From 60298033ffe1f0c28c24c454a5399463f84cd20d Mon Sep 17 00:00:00 2001 From: Christoph Viebig Date: Fri, 10 Jul 2026 18:37:41 +0000 Subject: [PATCH 1/4] Add micro-benchmark for count_min_sketch serialization --- CMakeLists.txt | 6 + benchmarks/CMakeLists.txt | 59 ++++++++ ...nchmark_count_min_sketch_serialization.cpp | 140 ++++++++++++++++++ 3 files changed, 205 insertions(+) create mode 100644 benchmarks/CMakeLists.txt create mode 100644 benchmarks/benchmark_count_min_sketch_serialization.cpp diff --git a/CMakeLists.txt b/CMakeLists.txt index e3505c19..d8e70383 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -74,6 +74,8 @@ if (BUILD_TESTS) enable_testing() endif() +option(BUILD_BENCHMARKS "Build benchmarks" OFF) + option(COVERAGE "Enable code coverage reporting (g++/clang only)" OFF) if(COVERAGE AND CMAKE_CXX_COMPILER_ID MATCHES "GNU|Clang") set(CMAKE_BUILD_TYPE "Debug" FORCE) @@ -120,6 +122,10 @@ add_subdirectory(count) add_subdirectory(density) add_subdirectory(tdigest) +if (BUILD_BENCHMARKS) + add_subdirectory(benchmarks) +endif() + if (WITH_PYTHON) add_subdirectory(python) endif() diff --git a/benchmarks/CMakeLists.txt b/benchmarks/CMakeLists.txt new file mode 100644 index 00000000..b0e67835 --- /dev/null +++ b/benchmarks/CMakeLists.txt @@ -0,0 +1,59 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +if (NOT TARGET benchmark::benchmark_main) + include(FetchContent) + + set(BENCHMARK_ENABLE_TESTING OFF CACHE BOOL "Disable google-benchmark tests" FORCE) + set(BENCHMARK_ENABLE_GTEST_TESTS OFF CACHE BOOL "Disable google-benchmark gtest tests" FORCE) + set(BENCHMARK_ENABLE_INSTALL OFF CACHE BOOL "Disable google-benchmark install rules" FORCE) + set(BENCHMARK_INSTALL_DOCS OFF CACHE BOOL "Disable google-benchmark documentation install" FORCE) + set(BENCHMARK_ENABLE_WERROR OFF CACHE BOOL "Disable google-benchmark warnings as errors" FORCE) + + FetchContent_Declare( + googlebenchmark + GIT_REPOSITORY https://github.com/google/benchmark.git + GIT_TAG v1.7.1 + ) + + FetchContent_MakeAvailable(googlebenchmark) +endif() + +add_executable(count_min_sketch_benchmark) + +target_link_libraries(count_min_sketch_benchmark + PRIVATE + count + benchmark::benchmark_main +) + +set_target_properties(count_min_sketch_benchmark PROPERTIES + CXX_STANDARD_REQUIRED YES +) + +set(COUNT_MIN_SKETCH_BENCHMARK_SOURCES + benchmark_count_min_sketch_serialization.cpp +) + +if (EXISTS "${CMAKE_CURRENT_SOURCE_DIR}/benchmark_count_min_sketch.cpp") + list(APPEND COUNT_MIN_SKETCH_BENCHMARK_SOURCES benchmark_count_min_sketch.cpp) +endif() + +target_sources(count_min_sketch_benchmark + PRIVATE + ${COUNT_MIN_SKETCH_BENCHMARK_SOURCES} +) diff --git a/benchmarks/benchmark_count_min_sketch_serialization.cpp b/benchmarks/benchmark_count_min_sketch_serialization.cpp new file mode 100644 index 00000000..b6d502e8 --- /dev/null +++ b/benchmarks/benchmark_count_min_sketch_serialization.cpp @@ -0,0 +1,140 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#include +#include + +#include +#include +#include +#include +#include +#include +#include + +namespace +{ + +using Sketch = datasketches::count_min_sketch; + +// Roughly 99.9% confidence. Serialization cost then scales with the bucket count. +constexpr uint8_t NUM_HASHES = 7; + +Sketch makeSketch(uint32_t num_buckets) +{ + Sketch sketch(NUM_HASHES, num_buckets); + for (uint64_t i = 0; i < NUM_HASHES; ++i) + sketch.update(i, i + 1); + return sketch; +} + +class fixed_buffer_streambuf final: public std::streambuf +{ +public: + explicit fixed_buffer_streambuf(std::vector& buffer): buffer_(buffer), position_(0) {} + + void reset() + { + position_ = 0; + } + + size_t bytes_written() const + { + return position_; + } + +protected: + std::streamsize xsputn(const char* data, std::streamsize size) override + { + const size_t requested = static_cast(size); + const size_t available = buffer_.size() - position_; + const size_t bytes_to_write = std::min(requested, available); + if (bytes_to_write > 0) { + std::memcpy(buffer_.data() + position_, data, bytes_to_write); + position_ += bytes_to_write; + } + return static_cast(bytes_to_write); + } + + int_type overflow(int_type ch) override + { + if (traits_type::eq_int_type(ch, traits_type::eof())) + return traits_type::not_eof(ch); + if (position_ == buffer_.size()) + return traits_type::eof(); + buffer_[position_++] = static_cast(traits_type::to_char_type(ch)); + return ch; + } + +private: + std::vector& buffer_; + size_t position_; +}; + +void BM_CountMinGetSerializedSizeBytes(benchmark::State& state) +{ + const auto sketch = makeSketch(static_cast(state.range(0))); + + for (auto _ : state) { + const auto size = sketch.get_serialized_size_bytes(); + benchmark::DoNotOptimize(size); + } + + state.SetItemsProcessed(state.iterations()); +} + +void BM_CountMinSerializeVector(benchmark::State& state) +{ + const auto sketch = makeSketch(static_cast(state.range(0))); + const auto serialized_size = sketch.get_serialized_size_bytes(); + + for (auto _ : state) { + auto bytes = sketch.serialize(); + benchmark::DoNotOptimize(bytes.data()); + benchmark::DoNotOptimize(bytes.size()); + benchmark::ClobberMemory(); + } + + state.SetBytesProcessed(state.iterations() * static_cast(serialized_size)); +} + +void BM_CountMinSerializeOStream(benchmark::State& state) +{ + const auto sketch = makeSketch(static_cast(state.range(0))); + const auto serialized_size = sketch.get_serialized_size_bytes(); + std::vector bytes(serialized_size); + fixed_buffer_streambuf stream_buffer(bytes); + std::ostream os(&stream_buffer); + + for (auto _ : state) { + stream_buffer.reset(); + os.clear(); + sketch.serialize(os); + benchmark::DoNotOptimize(stream_buffer.bytes_written()); + benchmark::ClobberMemory(); + } + + state.SetBytesProcessed(state.iterations() * static_cast(serialized_size)); +} + +} + +BENCHMARK(BM_CountMinGetSerializedSizeBytes)->RangeMultiplier(8)->Range(1024, 65536); +BENCHMARK(BM_CountMinSerializeVector)->RangeMultiplier(8)->Range(1024, 65536); +BENCHMARK(BM_CountMinSerializeOStream)->RangeMultiplier(8)->Range(1024, 65536); From 4e66c11a247d1512f095e2c85e4efb846692fe0a Mon Sep 17 00:00:00 2001 From: Christoph Viebig Date: Thu, 9 Jul 2026 18:53:15 +0200 Subject: [PATCH 2/4] Add serialize_to sink API for count-min sketch --- count/include/count_min.hpp | 9 ++++ count/include/count_min_impl.hpp | 84 +++++++++++++------------------- count/test/count_min_test.cpp | 23 +++++++++ 3 files changed, 66 insertions(+), 50 deletions(-) diff --git a/count/include/count_min.hpp b/count/include/count_min.hpp index 1fbba0e6..d1cec7fb 100644 --- a/count/include/count_min.hpp +++ b/count/include/count_min.hpp @@ -326,6 +326,15 @@ class count_min_sketch{ */ void serialize(std::ostream& os) const; + /** + * This method serializes the sketch by passing binary fragments to a callback. + * The callback must be callable as write_bytes(const void* data, size_t size). + * The data pointer is valid only until the callback returns. + * @return size in bytes written to the callback + */ + template + size_t serialize_to(WriteBytes&& write_bytes) const; + // This is a convenience alias for users // The type returned by the following serialize method using vector_bytes = std::vector::template rebind_alloc>; diff --git a/count/include/count_min_impl.hpp b/count/include/count_min_impl.hpp index 45376e7b..cb7235d6 100644 --- a/count/include/count_min_impl.hpp +++ b/count/include/count_min_impl.hpp @@ -268,6 +268,22 @@ return _sketch_array.end(); template void count_min_sketch::serialize(std::ostream& os) const { + serialize_to([&os](const void* data, size_t size) { + os.write(static_cast(data), size); + }); +} + +template +static inline void write_count_min_value(WriteBytes& write_bytes, size_t& bytes_written, const T& value) { + write_bytes(&value, sizeof(value)); + bytes_written += sizeof(value); +} + +template +template +size_t count_min_sketch::serialize_to(WriteBytes&& write_bytes) const { + size_t bytes_written = 0; + // Long 0 //const uint8_t preamble_longs = is_empty() ? PREAMBLE_LONGS_SHORT : PREAMBLE_LONGS_FULL; const uint8_t preamble_longs = PREAMBLE_LONGS_SHORT; @@ -275,32 +291,31 @@ void count_min_sketch::serialize(std::ostream& os) const { const uint8_t family_id = FAMILY_ID; const uint8_t flags_byte = (is_empty() ? 1 << flags::IS_EMPTY : 0); const uint32_t unused32 = NULL_32; - write(os, preamble_longs); - write(os, ser_ver); - write(os, family_id); - write(os, flags_byte); - write(os, unused32); + write_count_min_value(write_bytes, bytes_written, preamble_longs); + write_count_min_value(write_bytes, bytes_written, ser_ver); + write_count_min_value(write_bytes, bytes_written, family_id); + write_count_min_value(write_bytes, bytes_written, flags_byte); + write_count_min_value(write_bytes, bytes_written, unused32); // Long 1 const uint32_t nbuckets = _num_buckets; const uint8_t nhashes = _num_hashes; const uint16_t seed_hash(compute_seed_hash(_seed)); const uint8_t unused8 = NULL_8; - write(os, nbuckets); - write(os, nhashes); - write(os, seed_hash); - write(os, unused8); - if (is_empty()) return; // sketch is empty, no need to write further bytes. + write_count_min_value(write_bytes, bytes_written, nbuckets); + write_count_min_value(write_bytes, bytes_written, nhashes); + write_count_min_value(write_bytes, bytes_written, seed_hash); + write_count_min_value(write_bytes, bytes_written, unused8); + if (is_empty()) return bytes_written; // sketch is empty, no need to write further bytes. // Long 2 - write(os, _total_weight); + const W t_weight = _total_weight; + write_count_min_value(write_bytes, bytes_written, t_weight); // Long 3 onwards: remaining bytes are consumed by writing the weight and the array values. - auto it = _sketch_array.begin(); - while (it != _sketch_array.end()) { - write(os, *it); - ++it; - } + for (const auto& value: _sketch_array) write_count_min_value(write_bytes, bytes_written, value); + + return bytes_written; } template @@ -351,40 +366,9 @@ template auto count_min_sketch::serialize(unsigned header_size_bytes) const -> vector_bytes { vector_bytes bytes(header_size_bytes + get_serialized_size_bytes(), 0, _allocator); uint8_t *ptr = bytes.data() + header_size_bytes; - - // Long 0 - const uint8_t preamble_longs = PREAMBLE_LONGS_SHORT; - ptr += copy_to_mem(preamble_longs, ptr); - const uint8_t ser_ver = SERIAL_VERSION_1; - ptr += copy_to_mem(ser_ver, ptr); - const uint8_t family_id = FAMILY_ID; - ptr += copy_to_mem(family_id, ptr); - const uint8_t flags_byte = (is_empty() ? 1 << flags::IS_EMPTY : 0); - ptr += copy_to_mem(flags_byte, ptr); - const uint32_t unused32 = NULL_32; - ptr += copy_to_mem(unused32, ptr); - - // Long 1 - const uint32_t nbuckets = _num_buckets; - const uint8_t nhashes = _num_hashes; - const uint16_t seed_hash(compute_seed_hash(_seed)); - const uint8_t null_characters_8 = NULL_8; - ptr += copy_to_mem(nbuckets, ptr); - ptr += copy_to_mem(nhashes, ptr); - ptr += copy_to_mem(seed_hash, ptr); - ptr += copy_to_mem(null_characters_8, ptr); - if (is_empty()) return bytes; // sketch is empty, no need to write further bytes. - - // Long 2 - const W t_weight = _total_weight; - ptr += copy_to_mem(t_weight, ptr); - - // Long 3 onwards: remaining bytes are consumed by writing the weight and the array values. - auto it = _sketch_array.begin(); - while (it != _sketch_array.end()) { - ptr += copy_to_mem(*it, ptr); - ++it; - } + serialize_to([&ptr](const void* data, size_t size) { + ptr += copy_to_mem(data, ptr, size); + }); return bytes; } diff --git a/count/test/count_min_test.cpp b/count/test/count_min_test.cpp index 143be1b8..ea1ca44f 100644 --- a/count/test/count_min_test.cpp +++ b/count/test/count_min_test.cpp @@ -300,4 +300,27 @@ TEST_CASE("CountMin sketch: bytes serialize-deserialize non-empty", "[cm_sketch] } +TEST_CASE("CountMin sketch: sink serialize matches bytes", "[cm_sketch]") { + auto check_sink_serialize = [](const count_min_sketch& c) { + auto bytes = c.serialize(); + std::vector sink_bytes; + sink_bytes.reserve(c.get_serialized_size_bytes()); + + const size_t bytes_written = c.serialize_to([&sink_bytes](const void* data, size_t size) { + const auto* begin = static_cast(data); + sink_bytes.insert(sink_bytes.end(), begin, begin + size); + }); + + REQUIRE(bytes_written == bytes.size()); + REQUIRE(sink_bytes == bytes); + }; + + count_min_sketch empty(3, 32); + check_sink_serialize(empty); + + count_min_sketch non_empty(5, 64); + for (uint64_t i=0; i < 10; ++i) non_empty.update(i, 10 * i * i); + check_sink_serialize(non_empty); +} + } /* namespace datasketches */ From 764cdf1e725b8158ba5b3eb137a557870671ce29 Mon Sep 17 00:00:00 2001 From: Christoph Viebig Date: Fri, 10 Jul 2026 18:37:41 +0000 Subject: [PATCH 3/4] Extend micro-benchmark for count_min_sketch serialization --- ...nchmark_count_min_sketch_serialization.cpp | 21 +++++++++++++++++++ 1 file changed, 21 insertions(+) diff --git a/benchmarks/benchmark_count_min_sketch_serialization.cpp b/benchmarks/benchmark_count_min_sketch_serialization.cpp index b6d502e8..bd62f0c2 100644 --- a/benchmarks/benchmark_count_min_sketch_serialization.cpp +++ b/benchmarks/benchmark_count_min_sketch_serialization.cpp @@ -133,8 +133,29 @@ void BM_CountMinSerializeOStream(benchmark::State& state) state.SetBytesProcessed(state.iterations() * static_cast(serialized_size)); } +void BM_CountMinSerializeToSink(benchmark::State& state) +{ + const auto sketch = makeSketch(static_cast(state.range(0))); + const auto serialized_size = sketch.get_serialized_size_bytes(); + std::vector bytes(serialized_size); + + for (auto _ : state) { + uint8_t* ptr = bytes.data(); + const auto bytes_written = sketch.serialize_to([&ptr](const void* data, size_t size) { + std::memcpy(ptr, data, size); + ptr += size; + }); + benchmark::DoNotOptimize(ptr); + benchmark::DoNotOptimize(bytes_written); + benchmark::ClobberMemory(); + } + + state.SetBytesProcessed(state.iterations() * static_cast(serialized_size)); +} + } BENCHMARK(BM_CountMinGetSerializedSizeBytes)->RangeMultiplier(8)->Range(1024, 65536); BENCHMARK(BM_CountMinSerializeVector)->RangeMultiplier(8)->Range(1024, 65536); BENCHMARK(BM_CountMinSerializeOStream)->RangeMultiplier(8)->Range(1024, 65536); +BENCHMARK(BM_CountMinSerializeToSink)->RangeMultiplier(8)->Range(1024, 65536); From 63e7f75771bbf168704cb53956b0a7d0dd90bf30 Mon Sep 17 00:00:00 2001 From: Christoph Viebig Date: Fri, 10 Jul 2026 17:00:46 +0000 Subject: [PATCH 4/4] Serialize count_min_sketch sketch array in a single write call --- count/include/count_min.hpp | 3 +++ count/include/count_min_impl.hpp | 6 +++++- count/test/count_min_test.cpp | 14 ++++++++++---- 3 files changed, 18 insertions(+), 5 deletions(-) diff --git a/count/include/count_min.hpp b/count/include/count_min.hpp index d1cec7fb..63ec3beb 100644 --- a/count/include/count_min.hpp +++ b/count/include/count_min.hpp @@ -21,6 +21,8 @@ #define COUNT_MIN_HPP_ #include +#include +#include #include "common_defs.hpp" namespace datasketches { @@ -36,6 +38,7 @@ template > class count_min_sketch{ static_assert(std::is_arithmetic::value, "Arithmetic type expected"); + static_assert(!std::is_same::value, "Boolean weight type is not supported"); public: using allocator_type = Allocator; using const_iterator = typename std::vector::const_iterator; diff --git a/count/include/count_min_impl.hpp b/count/include/count_min_impl.hpp index cb7235d6..f1dcb4c5 100644 --- a/count/include/count_min_impl.hpp +++ b/count/include/count_min_impl.hpp @@ -313,7 +313,11 @@ size_t count_min_sketch::serialize_to(WriteBytes&& write_bytes) const { write_count_min_value(write_bytes, bytes_written, t_weight); // Long 3 onwards: remaining bytes are consumed by writing the weight and the array values. - for (const auto& value: _sketch_array) write_count_min_value(write_bytes, bytes_written, value); + const size_t sketch_array_bytes = sizeof(W) * _sketch_array.size(); + if (sketch_array_bytes > 0) { + write_bytes(_sketch_array.data(), sketch_array_bytes); + bytes_written += sketch_array_bytes; + } return bytes_written; } diff --git a/count/test/count_min_test.cpp b/count/test/count_min_test.cpp index ea1ca44f..ee0b7d4a 100644 --- a/count/test/count_min_test.cpp +++ b/count/test/count_min_test.cpp @@ -301,26 +301,32 @@ TEST_CASE("CountMin sketch: bytes serialize-deserialize non-empty", "[cm_sketch] } TEST_CASE("CountMin sketch: sink serialize matches bytes", "[cm_sketch]") { - auto check_sink_serialize = [](const count_min_sketch& c) { + auto check_sink_serialize = [](const count_min_sketch& c, size_t expected_fragments, size_t expected_last_fragment_size = 0) { auto bytes = c.serialize(); std::vector sink_bytes; sink_bytes.reserve(c.get_serialized_size_bytes()); + std::vector fragment_sizes; - const size_t bytes_written = c.serialize_to([&sink_bytes](const void* data, size_t size) { + const size_t bytes_written = c.serialize_to([&sink_bytes, &fragment_sizes](const void* data, size_t size) { const auto* begin = static_cast(data); sink_bytes.insert(sink_bytes.end(), begin, begin + size); + fragment_sizes.push_back(size); }); REQUIRE(bytes_written == bytes.size()); REQUIRE(sink_bytes == bytes); + REQUIRE(fragment_sizes.size() == expected_fragments); + if (expected_last_fragment_size > 0) { + REQUIRE(fragment_sizes.back() == expected_last_fragment_size); + } }; count_min_sketch empty(3, 32); - check_sink_serialize(empty); + check_sink_serialize(empty, 9); count_min_sketch non_empty(5, 64); for (uint64_t i=0; i < 10; ++i) non_empty.update(i, 10 * i * i); - check_sink_serialize(non_empty); + check_sink_serialize(non_empty, 11, sizeof(uint64_t) * non_empty.get_num_hashes() * non_empty.get_num_buckets()); } } /* namespace datasketches */