From fa89872f2a7a4845e6d2a15e0a88486fb0dcfbf6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Milo=C5=A1=20Stojko?= Date: Mon, 9 Nov 2026 09:29:50 +0000 Subject: [PATCH 01/12] perf: kll_sketch grows its items buffer by 8x up to k A new sketch has room for min(k, 200) items instead of k, and a full buffer smaller than k grows by RESIZE_FACTOR (X8) up to k. Compaction still starts only when the buffer holds k items, and while there is one level the serialized levels are offsets into a buffer of k items, so results and serialized bytes do not change. Co-authored-by: Isaac --- kll/include/kll_sketch.hpp | 10 +- kll/include/kll_sketch_impl.hpp | 39 +++++- kll/test/kll_sketch_test.cpp | 212 ++++++++++++++++++++++++++++++++ 3 files changed, 255 insertions(+), 6 deletions(-) diff --git a/kll/include/kll_sketch.hpp b/kll/include/kll_sketch.hpp index 39cac445..287be17a 100644 --- a/kll/include/kll_sketch.hpp +++ b/kll/include/kll_sketch.hpp @@ -39,6 +39,10 @@ namespace kll_constants { const uint16_t MIN_K = DEFAULT_M; /// max value of parameter K const uint16_t MAX_K = (1 << 16) - 1; + /// initial size of the items buffer if K is larger + const uint16_t INITIAL_ITEMS_SIZE = DEFAULT_K; + /// growth factor of the items buffer until it reaches K + const resize_factor RESIZE_FACTOR = resize_factor::X8; } /** @@ -183,7 +187,9 @@ class kll_sketch { using quantile_return_type = typename quantiles_sorted_view::quantile_return_type; /** - * Constructor + * Constructor. + * The sketch starts with space for min(k, INITIAL_ITEMS_SIZE) items, + * which grows by RESIZE_FACTOR up to k items before the first compaction. * @param k affects the size of the sketch and its estimation error * @param comparator strict weak ordering function (see C++ named requirements: Compare) * @param allocator used by this sketch to allocate memory @@ -580,6 +586,8 @@ class kll_sketch { uint8_t find_level_to_compact() const; void add_empty_top_level_to_completely_full_sketch(); + void grow_items(); + uint32_t get_serialized_levels_offset() const; void sort_level_zero(); template void merge_higher_levels(O&& other, uint64_t final_n); diff --git a/kll/include/kll_sketch_impl.hpp b/kll/include/kll_sketch_impl.hpp index a1d0c4da..fec5dfbe 100644 --- a/kll/include/kll_sketch_impl.hpp +++ b/kll/include/kll_sketch_impl.hpp @@ -45,7 +45,7 @@ is_level_zero_sorted_(false), n_(0), levels_(2, 0, allocator), items_(nullptr), -items_size_(k_), +items_size_(std::min(k_, kll_constants::INITIAL_ITEMS_SIZE)), min_item_(), max_item_(), sorted_view_(nullptr) @@ -54,7 +54,7 @@ sorted_view_(nullptr) throw std::invalid_argument("K must be >= " + std::to_string(kll_constants::MIN_K) + " and <= " + std::to_string(kll_constants::MAX_K) + ": " + std::to_string(k)); } - levels_[0] = levels_[1] = k; + levels_[0] = levels_[1] = items_size_; items_ = allocator_.allocate(items_size_); } @@ -199,7 +199,13 @@ void kll_sketch::update_min_max(const T& item) { template uint32_t kll_sketch::internal_update() { - if (levels_[0] == 0) { compress_while_updating(); } + if (levels_[0] == 0) { + if (items_size_ < k_) { + grow_items(); + } else { + compress_while_updating(); + } + } n_++; is_level_zero_sorted_ = false; return --levels_[0]; @@ -389,7 +395,8 @@ void kll_sketch::serialize(std::ostream& os, const SerDe& sd) const { write(os, min_k_); write(os, num_levels_); write(os, unused); - write(os, levels_.data(), sizeof(levels_[0]) * num_levels_); + const uint32_t levels_offset = get_serialized_levels_offset(); + for (uint8_t i = 0; i < num_levels_; ++i) { write(os, levels_[i] + levels_offset); } sd.serialize(os, &*min_item_, 1); sd.serialize(os, &*max_item_, 1); } @@ -425,7 +432,8 @@ auto kll_sketch::serialize(unsigned header_size_bytes, const SerDe& sd) ptr += copy_to_mem(min_k_, ptr); ptr += copy_to_mem(num_levels_, ptr); ptr += sizeof(uint8_t); // unused - ptr += copy_to_mem(levels_.data(), ptr, sizeof(levels_[0]) * num_levels_); + const uint32_t levels_offset = get_serialized_levels_offset(); + for (uint8_t i = 0; i < num_levels_; ++i) { ptr += copy_to_mem(levels_[i] + levels_offset, ptr); } ptr += sd.serialize(ptr, end_ptr - ptr, &*min_item_, 1); ptr += sd.serialize(ptr, end_ptr - ptr, &*max_item_, 1); } @@ -753,6 +761,27 @@ void kll_sketch::add_empty_top_level_to_completely_full_sketch() { levels_[num_levels_] = new_total_cap; // initialize the new "extra" index at the top } +// Grows the items buffer of a one-level sketch when it is full and smaller than k. +// Compaction starts only when the buffer is full at k items, so growing does not change the result. +template +void kll_sketch::grow_items() { + const uint32_t new_size = std::min(items_size_ << static_cast(kll_constants::RESIZE_FACTOR), k_); + const uint32_t delta = new_size - items_size_; + T* new_items = allocator_.allocate(new_size); + kll_helper::move_construct(items_, levels_[0], levels_[1], new_items, levels_[0] + delta, true); + allocator_.deallocate(items_, items_size_); + items_ = new_items; + items_size_ = new_size; + levels_[0] += delta; + levels_[1] += delta; +} + +// Serialized levels are offsets into a buffer of k items while there is one level. +template +uint32_t kll_sketch::get_serialized_levels_offset() const { + return num_levels_ == 1 ? k_ - items_size_ : 0; +} + template void kll_sketch::sort_level_zero() { if (!is_level_zero_sorted_) { diff --git a/kll/test/kll_sketch_test.cpp b/kll/test/kll_sketch_test.cpp index cb09473e..d20db6b3 100644 --- a/kll/test/kll_sketch_test.cpp +++ b/kll/test/kll_sketch_test.cpp @@ -18,11 +18,13 @@ */ #include +#include #include #include #include #include #include +#include #include #include @@ -905,4 +907,214 @@ TEST_CASE("kll sketch: deserialize rejects inconsistent levels", "[kll_sketch]") } } +// A deserialized sketch has a buffer of the full size. +template +static Sketch with_full_buffer(const Sketch& sketch) { + auto bytes = sketch.serialize(); + return Sketch::deserialize(bytes.data(), bytes.size(), serde(), + sketch.get_comparator(), sketch.get_allocator()); +} + +// Applies fn to both sketches with the same random bits, so that they compact the same way. +template +static void apply_to_both(Sketch& growing, Sketch& full, const Fn& fn) { + const auto random_bit = random_utils::random_bit; + fn(growing); + random_utils::random_bit = random_bit; + fn(full); +} + +template +static std::vector> items_and_weights(const Sketch& sketch) { + std::vector> result; + for (const auto pair: sketch) { result.emplace_back(pair.first, pair.second); } + return result; +} + +// Requires the same retained items and weights, the same query results and the same bytes. +template +static void require_identical(const Sketch& growing, const Sketch& full) { + REQUIRE(growing.get_n() == full.get_n()); + REQUIRE(growing.get_num_retained() == full.get_num_retained()); + REQUIRE(growing.is_estimation_mode() == full.is_estimation_mode()); + REQUIRE(growing.get_normalized_rank_error(false) == full.get_normalized_rank_error(false)); + REQUIRE(growing.get_normalized_rank_error(true) == full.get_normalized_rank_error(true)); + REQUIRE(items_and_weights(growing) == items_and_weights(full)); + if (!growing.is_empty()) { + REQUIRE(growing.get_min_item() == full.get_min_item()); + REQUIRE(growing.get_max_item() == full.get_max_item()); + std::vector split_points; + for (int i = 0; i <= 100; ++i) { + const double rank = i / 100.0; + for (const bool inclusive: {false, true}) { + const auto quantile = growing.get_quantile(rank, inclusive); + REQUIRE(quantile == full.get_quantile(rank, inclusive)); + REQUIRE(growing.get_rank(quantile, inclusive) == full.get_rank(quantile, inclusive)); + } + split_points.push_back(growing.get_quantile(rank)); + } + split_points.erase(std::unique(split_points.begin(), split_points.end()), split_points.end()); + for (const bool inclusive: {false, true}) { + REQUIRE(growing.get_PMF(split_points.data(), split_points.size(), inclusive) + == full.get_PMF(split_points.data(), split_points.size(), inclusive)); + REQUIRE(growing.get_CDF(split_points.data(), split_points.size(), inclusive) + == full.get_CDF(split_points.data(), split_points.size(), inclusive)); + } + } + REQUIRE(growing.serialize() == full.serialize()); + std::stringstream growing_stream, full_stream; + growing.serialize(growing_stream); + full.serialize(full_stream); + REQUIRE(growing_stream.str() == full_stream.str()); +} + +// Sizes of the items buffer as it grows by RESIZE_FACTOR up to k. +static std::vector buffer_sizes(uint16_t k) { + std::vector sizes = {std::min(k, kll_constants::INITIAL_ITEMS_SIZE)}; + while (sizes.back() < k) { + sizes.push_back(std::min(sizes.back() << static_cast(kll_constants::RESIZE_FACTOR), k)); + } + return sizes; +} + +// Updates a sketch and a copy of it with a full buffer the same way, and requires the same results. +template +static void require_same_as_full_buffer(Sketch& growing, uint32_t num_items) { + auto full = with_full_buffer(growing); + random_utils::random_bit.seed(4); + for (uint32_t i = 0; i < num_items; ++i) { + const auto item = static_cast(i % 7777); + apply_to_both(growing, full, [&item](Sketch& sketch) { sketch.update(item); }); + if (i == 1000 || i == 20000) { require_identical(growing, full); } + } + require_identical(growing, full); +} + +TEST_CASE("kll sketch: lazy growth of the items buffer", "[kll_sketch]") { + test_allocator_total_bytes = 0; + const long long levels_bytes = 2 * sizeof(uint32_t); + + SECTION("k not above the initial size allocates k items") { + kll_float_sketch sketch(100, std::less(), 0); + REQUIRE(test_allocator_total_bytes == static_cast(levels_bytes + 100 * sizeof(float))); + for (int i = 0; i < 100; ++i) { + sketch.update(static_cast(i)); + REQUIRE(test_allocator_total_bytes == static_cast(levels_bytes + 100 * sizeof(float))); + } + } + + SECTION("buffer grows by RESIZE_FACTOR up to k") { + REQUIRE(buffer_sizes(kll_constants::MAX_K) == std::vector{200, 1600, 12800, 65535}); + kll_float_sketch sketch(kll_constants::MAX_K, std::less(), 0); + REQUIRE(test_allocator_total_bytes + == static_cast(levels_bytes + kll_constants::INITIAL_ITEMS_SIZE * sizeof(float))); + uint32_t n = 0; + for (const uint32_t size: buffer_sizes(kll_constants::MAX_K)) { + while (n < size) { + sketch.update(static_cast(n++)); + // the buffer grows when an item does not fit + REQUIRE(test_allocator_total_bytes == static_cast(levels_bytes + size * sizeof(float))); + } + } + REQUIRE_FALSE(sketch.is_estimation_mode()); + sketch.update(0); + REQUIRE(sketch.is_estimation_mode()); + } + + SECTION("merge grows the buffer like updates") { + kll_float_sketch other(kll_constants::MAX_K, std::less(), 0); + for (int i = 0; i < 1000; ++i) { other.update(static_cast(i)); } + const long long other_bytes = test_allocator_total_bytes; + kll_float_sketch sketch(kll_constants::MAX_K, std::less(), 0); + sketch.merge(other); + REQUIRE(test_allocator_total_bytes - other_bytes == static_cast(levels_bytes + 1600 * sizeof(float))); + REQUIRE(sketch.get_n() == 1000); + } + + SECTION("same as a full buffer through compactions") { + // the buffer grows 1, 2 and 3 times before it reaches these k + const uint16_t ks[] = {201, 5000, 20000}; + for (const uint16_t k: ks) { + random_utils::random_bit.seed(k); + kll_float_sketch growing(k, std::less(), 0); + growing.update(0); + const long long growing_bytes = test_allocator_total_bytes; + kll_float_sketch full = with_full_buffer(growing); + REQUIRE(test_allocator_total_bytes - growing_bytes == static_cast(levels_bytes + k * sizeof(float))); + const auto sizes = buffer_sizes(k); + for (uint32_t n = 2; n <= 10u * k; ++n) { + const float item = static_cast((n * 7919) % 10007) / 3; + apply_to_both(growing, full, [item](kll_float_sketch& sketch) { sketch.update(item); }); + const bool at_growth = std::find(sizes.begin(), sizes.end(), n) != sizes.end() + || std::find(sizes.begin(), sizes.end(), n - 1) != sizes.end(); + if (at_growth || n % 997 == 0 || n % k <= 1) { require_identical(growing, full); } + } + REQUIRE(growing.is_estimation_mode()); + require_identical(growing, full); + } + } + + SECTION("same as a full buffer when merging") { + random_utils::random_bit.seed(3); + // exact mode sketches first, so that merging grows the buffer before compactions start + const std::vector> k_and_n = { + {30000, 1000}, {30000, 3000}, {20000, 15000}, {2000, 10000}, {8, 100}, {20000, 60000}}; + std::vector others; + for (const auto& shape: k_and_n) { + others.emplace_back(shape.first, std::less(), 0); + for (uint32_t i = 0; i < shape.second; ++i) { others.back().update(static_cast((i * 31) % 4999)); } + } + for (const bool reversed: {false, true}) { + kll_float_sketch growing(20000, std::less(), 0); + growing.update(0); + kll_float_sketch full = with_full_buffer(growing); + for (size_t i = 0; i < others.size(); ++i) { + const auto& other = others[reversed ? others.size() - 1 - i : i]; + apply_to_both(growing, full, [&other](kll_float_sketch& sketch) { sketch.merge(other); }); + require_identical(growing, full); + } + for (int i = 0; i < 50000; ++i) { + apply_to_both(growing, full, [i](kll_float_sketch& sketch) { sketch.update(static_cast(i)); }); + } + require_identical(growing, full); + } + } + + SECTION("copies, moves and converted sketches keep growing the same") { + kll_float_sketch sketch(kll_constants::MAX_K, std::less(), 0); + for (int i = 0; i < 500; ++i) { sketch.update(static_cast(i)); } + kll_float_sketch copied(sketch); + kll_float_sketch assigned(8, std::less(), 0); + assigned = sketch; + kll_float_sketch to_move(sketch); + kll_float_sketch moved(std::move(to_move)); + kll_float_sketch move_assigned(8, std::less(), 0); + kll_float_sketch to_move_assign(sketch); + move_assigned = std::move(to_move_assign); + for (auto* growing: {&sketch, &copied, &assigned, &moved, &move_assigned}) { + require_same_as_full_buffer(*growing, 200000); + } + kll_float_sketch source(kll_constants::MAX_K, std::less(), 0); + for (int i = 0; i < 500; ++i) { source.update(static_cast(i)); } + kll_sketch, test_allocator> converted(source, std::less(), 0); + require_same_as_full_buffer(converted, 200000); + } + + SECTION("same as a full buffer with strings") { + random_utils::random_bit.seed(5); + kll_string_sketch growing(5000, std::less(), 0); + growing.update("0"); + kll_string_sketch full = with_full_buffer(growing); + for (int i = 1; i < 30000; ++i) { + const std::string item = std::to_string((i * 7919) % 10007); + apply_to_both(growing, full, [&item](kll_string_sketch& sketch) { sketch.update(item); }); + if (i == 199 || i == 200 || i == 1599 || i == 1600 || i == 4999 || i == 10000) { require_identical(growing, full); } + } + require_identical(growing, full); + } + + // cleanup + REQUIRE(test_allocator_total_bytes == 0); +} + } /* namespace datasketches */ From 74b60d15879af6fb3d355597f4771c570b06523d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Milo=C5=A1=20Stojko?= Date: Mon, 9 Nov 2026 09:29:52 +0000 Subject: [PATCH 02/12] KLL: add benchmarks for building, merging and serializing Co-authored-by: Isaac --- benchmarks/CMakeLists.txt | 17 +++ benchmarks/benchmark_kll_sketch.cpp | 163 ++++++++++++++++++++++++++++ 2 files changed, 180 insertions(+) create mode 100644 benchmarks/benchmark_kll_sketch.cpp diff --git a/benchmarks/CMakeLists.txt b/benchmarks/CMakeLists.txt index f543dfce..342e4478 100644 --- a/benchmarks/CMakeLists.txt +++ b/benchmarks/CMakeLists.txt @@ -50,3 +50,20 @@ target_sources(count_min_sketch_benchmark benchmark_count_min_sketch.cpp benchmark_count_min_sketch_serialization.cpp ) + +add_executable(kll_sketch_benchmark) + +target_link_libraries(kll_sketch_benchmark + PRIVATE + kll + benchmark::benchmark_main +) + +set_target_properties(kll_sketch_benchmark PROPERTIES + CXX_STANDARD_REQUIRED YES +) + +target_sources(kll_sketch_benchmark + PRIVATE + benchmark_kll_sketch.cpp +) diff --git a/benchmarks/benchmark_kll_sketch.cpp b/benchmarks/benchmark_kll_sketch.cpp new file mode 100644 index 00000000..fea5b979 --- /dev/null +++ b/benchmarks/benchmark_kll_sketch.cpp @@ -0,0 +1,163 @@ +/* + * 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 + +namespace +{ + +using Sketch = datasketches::kll_sketch; + +std::vector makeItems(size_t size) +{ + std::vector items; + items.reserve(size); + uint64_t state = 1; + for (size_t i = 0; i < size; ++i) + { + state = state * 6364136223846793005ULL + 1442695040888963407ULL; + items.push_back(static_cast(state >> 11)); + } + return items; +} + +// A new sketch with k = range(0) gets range(1) items. +void BM_KllBuild(benchmark::State & state) +{ + const auto k = static_cast(state.range(0)); + const auto items = makeItems(static_cast(state.range(1))); + for (auto _ : state) + { + Sketch sketch(k); + for (const double item : items) + sketch.update(item); + + benchmark::DoNotOptimize(sketch.get_n()); + benchmark::ClobberMemory(); + } + + state.SetItemsProcessed(state.iterations() * static_cast(items.size())); +} + +// Many sketches with k = range(0) are alive at once, each with range(1) items, like the groups of +// a GROUP BY. +void BM_KllBuildManySketches(benchmark::State & state) +{ + constexpr size_t NUM_SKETCHES = 1000; + const auto k = static_cast(state.range(0)); + const auto items = makeItems(static_cast(state.range(1))); + for (auto _ : state) + { + std::vector sketches(NUM_SKETCHES, Sketch(k)); + for (auto & sketch : sketches) + for (const double item : items) + sketch.update(item); + + benchmark::DoNotOptimize(sketches.data()); + benchmark::ClobberMemory(); + } + + state.SetItemsProcessed(state.iterations() * static_cast(NUM_SKETCHES * items.size())); +} + +// A new sketch with k = range(0) merges 100 sketches of range(1) items each. +void BM_KllMergeInto(benchmark::State & state) +{ + constexpr size_t NUM_INPUTS = 100; + const auto k = static_cast(state.range(0)); + const auto items = makeItems(static_cast(state.range(1))); + std::vector inputs(NUM_INPUTS, Sketch(k)); + for (auto & input : inputs) + for (const double item : items) + input.update(item); + + for (auto _ : state) + { + Sketch sketch(k); + for (const auto & input : inputs) + sketch.merge(input); + + benchmark::DoNotOptimize(sketch.get_n()); + benchmark::ClobberMemory(); + } + + state.SetItemsProcessed(state.iterations() * static_cast(NUM_INPUTS * items.size())); +} + +// Serializes a sketch with k = range(0) and range(1) items. +void BM_KllSerialize(benchmark::State & state) +{ + Sketch sketch(static_cast(state.range(0))); + for (const double item : makeItems(static_cast(state.range(1)))) + sketch.update(item); + + for (auto _ : state) + { + auto bytes = sketch.serialize(); + benchmark::DoNotOptimize(bytes.data()); + benchmark::ClobberMemory(); + } +} + +void buildArgs(benchmark::internal::Benchmark * b) +{ + for (const int64_t k : {200, 1000, 5000, 20000, 65535}) + { + // n = k fills the buffer up to k with no compaction, so it includes every growth step + for (const int64_t n : {int64_t{1}, int64_t{10}, int64_t{100}, int64_t{1000}, k, int64_t{10000}, + int64_t{100000}, int64_t{1000000}}) + { + if (n <= 100 * k) + b->Args({k, n}); + } + } +} + +void manySketchesArgs(benchmark::internal::Benchmark * b) +{ + for (const int64_t k : {200, 1000, 5000, 20000, 65535}) + for (const int64_t n : {1, 10, 100, 1000}) + b->Args({k, n}); +} + +void mergeArgs(benchmark::internal::Benchmark * b) +{ + for (const int64_t k : {200, 5000, 65535}) + for (const int64_t n : {10, 100, 1000}) + b->Args({k, n}); +} + +void serializeArgs(benchmark::internal::Benchmark * b) +{ + for (const int64_t k : {200, 5000, 65535}) + for (const int64_t n : {10, 1000, 100000}) + b->Args({k, n}); +} + +} + +BENCHMARK(BM_KllBuild)->Apply(buildArgs); +BENCHMARK(BM_KllBuildManySketches)->Apply(manySketchesArgs)->Unit(benchmark::kMicrosecond); +BENCHMARK(BM_KllMergeInto)->Apply(mergeArgs)->Unit(benchmark::kMicrosecond); +BENCHMARK(BM_KllSerialize)->Apply(serializeArgs); From 4a7d2966637cf21c9ef45a80696f55ac27c656aa Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Milo=C5=A1=20Stojko?= Date: Mon, 9 Nov 2026 10:40:16 +0000 Subject: [PATCH 03/12] KLL: merge a copy when merging into itself, keep to_string() capacity merge() reads the items of the other sketch while this sketch moves them, so merging a sketch into itself failed once the buffer grew or compacted. to_string() again prints k as the capacity of a one-level sketch, and the start size and growth factor are private constants. Co-authored-by: Isaac --- kll/include/kll_sketch.hpp | 14 +++++++------- kll/include/kll_sketch_impl.hpp | 19 ++++++++++++------- kll/test/kll_sketch_test.cpp | 27 +++++++++++++++++++++------ 3 files changed, 40 insertions(+), 20 deletions(-) diff --git a/kll/include/kll_sketch.hpp b/kll/include/kll_sketch.hpp index 287be17a..015a03e0 100644 --- a/kll/include/kll_sketch.hpp +++ b/kll/include/kll_sketch.hpp @@ -39,10 +39,6 @@ namespace kll_constants { const uint16_t MIN_K = DEFAULT_M; /// max value of parameter K const uint16_t MAX_K = (1 << 16) - 1; - /// initial size of the items buffer if K is larger - const uint16_t INITIAL_ITEMS_SIZE = DEFAULT_K; - /// growth factor of the items buffer until it reaches K - const resize_factor RESIZE_FACTOR = resize_factor::X8; } /** @@ -188,8 +184,8 @@ class kll_sketch { /** * Constructor. - * The sketch starts with space for min(k, INITIAL_ITEMS_SIZE) items, - * which grows by RESIZE_FACTOR up to k items before the first compaction. + * The sketch starts with space for min(k, 200) items, + * which grows by 8x up to k items before the first compaction. * @param k affects the size of the sketch and its estimation error * @param comparator strict weak ordering function (see C++ named requirements: Compare) * @param allocator used by this sketch to allocate memory @@ -555,6 +551,10 @@ class kll_sketch { static const uint8_t PREAMBLE_INTS_SHORT = 2; // for empty and single item static const uint8_t PREAMBLE_INTS_FULL = 5; + // the items buffer starts at this size if k is larger, and grows by RESIZE_FACTOR up to k + static const uint16_t INITIAL_ITEMS_SIZE = kll_constants::DEFAULT_K; + static const resize_factor RESIZE_FACTOR = resize_factor::X8; + C comparator_; A allocator_; uint16_t k_; @@ -587,7 +587,7 @@ class kll_sketch { uint8_t find_level_to_compact() const; void add_empty_top_level_to_completely_full_sketch(); void grow_items(); - uint32_t get_serialized_levels_offset() const; + uint32_t get_levels_offset() const; void sort_level_zero(); template void merge_higher_levels(O&& other, uint64_t final_n); diff --git a/kll/include/kll_sketch_impl.hpp b/kll/include/kll_sketch_impl.hpp index fec5dfbe..061bfa2b 100644 --- a/kll/include/kll_sketch_impl.hpp +++ b/kll/include/kll_sketch_impl.hpp @@ -45,7 +45,7 @@ is_level_zero_sorted_(false), n_(0), levels_(2, 0, allocator), items_(nullptr), -items_size_(std::min(k_, kll_constants::INITIAL_ITEMS_SIZE)), +items_size_(std::min(k_, INITIAL_ITEMS_SIZE)), min_item_(), max_item_(), sorted_view_(nullptr) @@ -215,6 +215,11 @@ template template void kll_sketch::merge(FwdSk&& other) { if (other.is_empty()) { return; } + // the loop below reads the items of other while this sketch moves them + if (static_cast(&other) == this) { + merge(kll_sketch(*this)); + return; + } if (m_ != other.m_) { throw std::invalid_argument("incompatible M: " + std::to_string(m_) + " and " + std::to_string(other.m_)); } @@ -395,7 +400,7 @@ void kll_sketch::serialize(std::ostream& os, const SerDe& sd) const { write(os, min_k_); write(os, num_levels_); write(os, unused); - const uint32_t levels_offset = get_serialized_levels_offset(); + const uint32_t levels_offset = get_levels_offset(); for (uint8_t i = 0; i < num_levels_; ++i) { write(os, levels_[i] + levels_offset); } sd.serialize(os, &*min_item_, 1); sd.serialize(os, &*max_item_, 1); @@ -432,7 +437,7 @@ auto kll_sketch::serialize(unsigned header_size_bytes, const SerDe& sd) ptr += copy_to_mem(min_k_, ptr); ptr += copy_to_mem(num_levels_, ptr); ptr += sizeof(uint8_t); // unused - const uint32_t levels_offset = get_serialized_levels_offset(); + const uint32_t levels_offset = get_levels_offset(); for (uint8_t i = 0; i < num_levels_; ++i) { ptr += copy_to_mem(levels_[i] + levels_offset, ptr); } ptr += sd.serialize(ptr, end_ptr - ptr, &*min_item_, 1); ptr += sd.serialize(ptr, end_ptr - ptr, &*max_item_, 1); @@ -765,7 +770,7 @@ void kll_sketch::add_empty_top_level_to_completely_full_sketch() { // Compaction starts only when the buffer is full at k items, so growing does not change the result. template void kll_sketch::grow_items() { - const uint32_t new_size = std::min(items_size_ << static_cast(kll_constants::RESIZE_FACTOR), k_); + const uint32_t new_size = std::min(items_size_ << static_cast(RESIZE_FACTOR), k_); const uint32_t delta = new_size - items_size_; T* new_items = allocator_.allocate(new_size); kll_helper::move_construct(items_, levels_[0], levels_[1], new_items, levels_[0] + delta, true); @@ -776,9 +781,9 @@ void kll_sketch::grow_items() { levels_[1] += delta; } -// Serialized levels are offsets into a buffer of k items while there is one level. +// Offset of levels_ in a buffer of k items, which serialize() and to_string() show while there is one level. template -uint32_t kll_sketch::get_serialized_levels_offset() const { +uint32_t kll_sketch::get_levels_offset() const { return num_levels_ == 1 ? k_ - items_size_ : 0; } @@ -989,7 +994,7 @@ string kll_sketch::to_string(bool print_levels, bool print_items) co os << " Estimation mode: " << (is_estimation_mode() ? "true" : "false") << std::endl; os << " Levels : " << (unsigned int) num_levels_ << std::endl; os << " Sorted : " << (is_level_zero_sorted_ ? "true" : "false") << std::endl; - os << " Capacity items : " << items_size_ << std::endl; + os << " Capacity items : " << items_size_ + get_levels_offset() << std::endl; os << " Retained items : " << get_num_retained() << std::endl; if (!is_empty()) { os << " Min item : " << *min_item_ << std::endl; diff --git a/kll/test/kll_sketch_test.cpp b/kll/test/kll_sketch_test.cpp index d20db6b3..c5048f86 100644 --- a/kll/test/kll_sketch_test.cpp +++ b/kll/test/kll_sketch_test.cpp @@ -966,13 +966,14 @@ static void require_identical(const Sketch& growing, const Sketch& full) { growing.serialize(growing_stream); full.serialize(full_stream); REQUIRE(growing_stream.str() == full_stream.str()); + REQUIRE(growing.to_string(true, false) == full.to_string(true, false)); } -// Sizes of the items buffer as it grows by RESIZE_FACTOR up to k. +// Sizes of the items buffer as it grows by 8x up to k. static std::vector buffer_sizes(uint16_t k) { - std::vector sizes = {std::min(k, kll_constants::INITIAL_ITEMS_SIZE)}; + std::vector sizes = {std::min(k, 200)}; while (sizes.back() < k) { - sizes.push_back(std::min(sizes.back() << static_cast(kll_constants::RESIZE_FACTOR), k)); + sizes.push_back(std::min(sizes.back() * 8, k)); } return sizes; } @@ -1003,11 +1004,10 @@ TEST_CASE("kll sketch: lazy growth of the items buffer", "[kll_sketch]") { } } - SECTION("buffer grows by RESIZE_FACTOR up to k") { + SECTION("buffer grows by 8x up to k") { REQUIRE(buffer_sizes(kll_constants::MAX_K) == std::vector{200, 1600, 12800, 65535}); kll_float_sketch sketch(kll_constants::MAX_K, std::less(), 0); - REQUIRE(test_allocator_total_bytes - == static_cast(levels_bytes + kll_constants::INITIAL_ITEMS_SIZE * sizeof(float))); + REQUIRE(test_allocator_total_bytes == static_cast(levels_bytes + 200 * sizeof(float))); uint32_t n = 0; for (const uint32_t size: buffer_sizes(kll_constants::MAX_K)) { while (n < size) { @@ -1113,6 +1113,21 @@ TEST_CASE("kll sketch: lazy growth of the items buffer", "[kll_sketch]") { require_identical(growing, full); } + SECTION("merge into itself") { + // with 150 items, merging into itself grows the buffer at MAX_K and compacts at DEFAULT_K + for (const uint16_t k: {kll_constants::DEFAULT_K, kll_constants::MAX_K}) { + kll_float_sketch sketch(k, std::less(), 0); + for (int i = 0; i < 150; ++i) { sketch.update(static_cast(i)); } + kll_float_sketch merged_with_copy(sketch); + const auto random_bit = random_utils::random_bit; + merged_with_copy.merge(kll_float_sketch(merged_with_copy)); + random_utils::random_bit = random_bit; + sketch.merge(sketch); + REQUIRE(sketch.get_n() == 300); + require_identical(sketch, merged_with_copy); + } + } + // cleanup REQUIRE(test_allocator_total_bytes == 0); } From 4b036fccca3b288b977f414c90c1a0a2f9516496 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Milo=C5=A1=20Stojko?= Date: Mon, 9 Nov 2026 11:21:38 +0000 Subject: [PATCH 04/12] KLL: start with k items if the resize factor is X1 With X1, grow_items() kept the buffer size, and the next update wrote before the start of the buffer. Like theta, the sketch now starts with the full buffer if the resize factor is X1. The start size is DEFAULT_K, the buffer of a sketch with the default k, so the INITIAL_ITEMS_SIZE constant is removed. Co-authored-by: Isaac --- kll/include/kll_sketch.hpp | 7 +++---- kll/include/kll_sketch_impl.hpp | 2 +- kll/test/kll_sketch_test.cpp | 19 ++++++++++--------- 3 files changed, 14 insertions(+), 14 deletions(-) diff --git a/kll/include/kll_sketch.hpp b/kll/include/kll_sketch.hpp index 015a03e0..5ebe01b2 100644 --- a/kll/include/kll_sketch.hpp +++ b/kll/include/kll_sketch.hpp @@ -184,8 +184,8 @@ class kll_sketch { /** * Constructor. - * The sketch starts with space for min(k, 200) items, - * which grows by 8x up to k items before the first compaction. + * The items buffer starts with space for min(k, DEFAULT_K) items (k items if RESIZE_FACTOR is X1), + * and grows by RESIZE_FACTOR up to k items before the first compaction. * @param k affects the size of the sketch and its estimation error * @param comparator strict weak ordering function (see C++ named requirements: Compare) * @param allocator used by this sketch to allocate memory @@ -551,8 +551,7 @@ class kll_sketch { static const uint8_t PREAMBLE_INTS_SHORT = 2; // for empty and single item static const uint8_t PREAMBLE_INTS_FULL = 5; - // the items buffer starts at this size if k is larger, and grows by RESIZE_FACTOR up to k - static const uint16_t INITIAL_ITEMS_SIZE = kll_constants::DEFAULT_K; + // growth factor of the items buffer until it holds k items static const resize_factor RESIZE_FACTOR = resize_factor::X8; C comparator_; diff --git a/kll/include/kll_sketch_impl.hpp b/kll/include/kll_sketch_impl.hpp index 061bfa2b..1e051111 100644 --- a/kll/include/kll_sketch_impl.hpp +++ b/kll/include/kll_sketch_impl.hpp @@ -45,7 +45,7 @@ is_level_zero_sorted_(false), n_(0), levels_(2, 0, allocator), items_(nullptr), -items_size_(std::min(k_, INITIAL_ITEMS_SIZE)), +items_size_(RESIZE_FACTOR == resize_factor::X1 ? k_ : std::min(k_, kll_constants::DEFAULT_K)), min_item_(), max_item_(), sorted_view_(nullptr) diff --git a/kll/test/kll_sketch_test.cpp b/kll/test/kll_sketch_test.cpp index c5048f86..ffb072f6 100644 --- a/kll/test/kll_sketch_test.cpp +++ b/kll/test/kll_sketch_test.cpp @@ -971,7 +971,7 @@ static void require_identical(const Sketch& growing, const Sketch& full) { // Sizes of the items buffer as it grows by 8x up to k. static std::vector buffer_sizes(uint16_t k) { - std::vector sizes = {std::min(k, 200)}; + std::vector sizes = {std::min(k, kll_constants::DEFAULT_K)}; while (sizes.back() < k) { sizes.push_back(std::min(sizes.back() * 8, k)); } @@ -995,19 +995,20 @@ TEST_CASE("kll sketch: lazy growth of the items buffer", "[kll_sketch]") { test_allocator_total_bytes = 0; const long long levels_bytes = 2 * sizeof(uint32_t); - SECTION("k not above the initial size allocates k items") { - kll_float_sketch sketch(100, std::less(), 0); - REQUIRE(test_allocator_total_bytes == static_cast(levels_bytes + 100 * sizeof(float))); - for (int i = 0; i < 100; ++i) { - sketch.update(static_cast(i)); - REQUIRE(test_allocator_total_bytes == static_cast(levels_bytes + 100 * sizeof(float))); - } + SECTION("k below DEFAULT_K allocates k items") { + const uint16_t k = kll_constants::DEFAULT_K / 2; + kll_float_sketch sketch(k, std::less(), 0); + sketch.update(0); + REQUIRE(test_allocator_total_bytes == static_cast(levels_bytes + k * sizeof(float))); + for (int i = 1; i < k; ++i) { sketch.update(static_cast(i)); } + REQUIRE(test_allocator_total_bytes == static_cast(levels_bytes + k * sizeof(float))); } SECTION("buffer grows by 8x up to k") { REQUIRE(buffer_sizes(kll_constants::MAX_K) == std::vector{200, 1600, 12800, 65535}); kll_float_sketch sketch(kll_constants::MAX_K, std::less(), 0); - REQUIRE(test_allocator_total_bytes == static_cast(levels_bytes + 200 * sizeof(float))); + REQUIRE(test_allocator_total_bytes + == static_cast(levels_bytes + kll_constants::DEFAULT_K * sizeof(float))); uint32_t n = 0; for (const uint32_t size: buffer_sizes(kll_constants::MAX_K)) { while (n < size) { From 5b246ccea5b96858bec66b0df24719a014048270 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Milo=C5=A1=20Stojko?= Date: Mon, 9 Nov 2026 11:31:39 +0000 Subject: [PATCH 05/12] KLL: check RESIZE_FACTOR at compile time Co-authored-by: Isaac --- kll/include/kll_sketch.hpp | 2 ++ 1 file changed, 2 insertions(+) diff --git a/kll/include/kll_sketch.hpp b/kll/include/kll_sketch.hpp index 5ebe01b2..6c388972 100644 --- a/kll/include/kll_sketch.hpp +++ b/kll/include/kll_sketch.hpp @@ -553,6 +553,8 @@ class kll_sketch { // growth factor of the items buffer until it holds k items static const resize_factor RESIZE_FACTOR = resize_factor::X8; + static_assert(RESIZE_FACTOR >= resize_factor::X1 && RESIZE_FACTOR <= resize_factor::X8, + "RESIZE_FACTOR must be X1, X2, X4 or X8"); C comparator_; A allocator_; From ed94f9c10a3843f97e6fcf4cc1705f3531a7b2bd Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Milo=C5=A1=20Stojko?= Date: Mon, 9 Nov 2026 11:45:39 +0000 Subject: [PATCH 06/12] KLL: throw when a sketch is merged into itself, as Java does merge() reads the items of the other sketch while it updates this one, so a self-merge failed once it compacted, and the growing buffer made it fail earlier. The Java KLL sketches throw "A sketch cannot merge with itself." (apache/datasketches-java#516), and so does count_min. The static_assert on RESIZE_FACTOR now checks only the lower bound. Co-authored-by: Isaac --- kll/include/kll_sketch.hpp | 4 ++-- kll/include/kll_sketch_impl.hpp | 6 +----- kll/test/kll_sketch_test.cpp | 24 +++++++++--------------- 3 files changed, 12 insertions(+), 22 deletions(-) diff --git a/kll/include/kll_sketch.hpp b/kll/include/kll_sketch.hpp index 6c388972..ebc77e1b 100644 --- a/kll/include/kll_sketch.hpp +++ b/kll/include/kll_sketch.hpp @@ -243,6 +243,7 @@ class kll_sketch { * Merges another sketch into this one. * If sketches contain strings, callers are responsible for ensuring that * both sketches were built using compatible string encodings. + * If other is this sketch, this throws std::invalid_argument. * @param other sketch to merge into this one */ template @@ -553,8 +554,7 @@ class kll_sketch { // growth factor of the items buffer until it holds k items static const resize_factor RESIZE_FACTOR = resize_factor::X8; - static_assert(RESIZE_FACTOR >= resize_factor::X1 && RESIZE_FACTOR <= resize_factor::X8, - "RESIZE_FACTOR must be X1, X2, X4 or X8"); + static_assert(RESIZE_FACTOR >= resize_factor::X1, "RESIZE_FACTOR must be at least X1"); C comparator_; A allocator_; diff --git a/kll/include/kll_sketch_impl.hpp b/kll/include/kll_sketch_impl.hpp index 1e051111..a1d8c1ea 100644 --- a/kll/include/kll_sketch_impl.hpp +++ b/kll/include/kll_sketch_impl.hpp @@ -214,12 +214,8 @@ uint32_t kll_sketch::internal_update() { template template void kll_sketch::merge(FwdSk&& other) { + if (this == &other) { throw std::invalid_argument("A sketch cannot merge with itself."); } if (other.is_empty()) { return; } - // the loop below reads the items of other while this sketch moves them - if (static_cast(&other) == this) { - merge(kll_sketch(*this)); - return; - } if (m_ != other.m_) { throw std::invalid_argument("incompatible M: " + std::to_string(m_) + " and " + std::to_string(other.m_)); } diff --git a/kll/test/kll_sketch_test.cpp b/kll/test/kll_sketch_test.cpp index ffb072f6..383782eb 100644 --- a/kll/test/kll_sketch_test.cpp +++ b/kll/test/kll_sketch_test.cpp @@ -568,6 +568,15 @@ TEST_CASE("kll sketch", "[kll_sketch]") { REQUIRE(sketch2.get_max_item() == 999999.0f); } + SECTION("merge into itself") { + kll_float_sketch sketch(200, std::less(), 0); + REQUIRE_THROWS_WITH(sketch.merge(sketch), "A sketch cannot merge with itself."); + for (int i = 0; i < 150; i++) sketch.update(static_cast(i)); + const auto bytes = sketch.serialize(); + REQUIRE_THROWS_WITH(sketch.merge(sketch), "A sketch cannot merge with itself."); + REQUIRE(sketch.serialize() == bytes); + } + SECTION("sketch of ints") { kll_sketch sketch; REQUIRE_THROWS_AS(sketch.get_quantile(0), std::runtime_error); @@ -1114,21 +1123,6 @@ TEST_CASE("kll sketch: lazy growth of the items buffer", "[kll_sketch]") { require_identical(growing, full); } - SECTION("merge into itself") { - // with 150 items, merging into itself grows the buffer at MAX_K and compacts at DEFAULT_K - for (const uint16_t k: {kll_constants::DEFAULT_K, kll_constants::MAX_K}) { - kll_float_sketch sketch(k, std::less(), 0); - for (int i = 0; i < 150; ++i) { sketch.update(static_cast(i)); } - kll_float_sketch merged_with_copy(sketch); - const auto random_bit = random_utils::random_bit; - merged_with_copy.merge(kll_float_sketch(merged_with_copy)); - random_utils::random_bit = random_bit; - sketch.merge(sketch); - REQUIRE(sketch.get_n() == 300); - require_identical(sketch, merged_with_copy); - } - } - // cleanup REQUIRE(test_allocator_total_bytes == 0); } From a82fd5b8280ee9745fc2914846f846d9c440190e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Milo=C5=A1=20Stojko?= Date: Mon, 9 Nov 2026 11:47:08 +0000 Subject: [PATCH 07/12] KLL: serialize the levels as master does once the buffer holds k items Only a one-level sketch whose buffer is still growing writes the start of level 0 as an offset into a buffer of k items. Otherwise both serializers run the same code as before. Co-authored-by: Isaac --- kll/include/kll_sketch.hpp | 1 - kll/include/kll_sketch_impl.hpp | 25 ++++++++++++++----------- 2 files changed, 14 insertions(+), 12 deletions(-) diff --git a/kll/include/kll_sketch.hpp b/kll/include/kll_sketch.hpp index ebc77e1b..5f840fcc 100644 --- a/kll/include/kll_sketch.hpp +++ b/kll/include/kll_sketch.hpp @@ -588,7 +588,6 @@ class kll_sketch { uint8_t find_level_to_compact() const; void add_empty_top_level_to_completely_full_sketch(); void grow_items(); - uint32_t get_levels_offset() const; void sort_level_zero(); template void merge_higher_levels(O&& other, uint64_t final_n); diff --git a/kll/include/kll_sketch_impl.hpp b/kll/include/kll_sketch_impl.hpp index a1d8c1ea..f5e003fd 100644 --- a/kll/include/kll_sketch_impl.hpp +++ b/kll/include/kll_sketch_impl.hpp @@ -396,8 +396,12 @@ void kll_sketch::serialize(std::ostream& os, const SerDe& sd) const { write(os, min_k_); write(os, num_levels_); write(os, unused); - const uint32_t levels_offset = get_levels_offset(); - for (uint8_t i = 0; i < num_levels_; ++i) { write(os, levels_[i] + levels_offset); } + if (items_size_ < k_) { + // one level in a buffer still growing to k items: write where it starts in a buffer of k items + write(os, levels_[0] + (k_ - items_size_)); + } else { + write(os, levels_.data(), sizeof(levels_[0]) * num_levels_); + } sd.serialize(os, &*min_item_, 1); sd.serialize(os, &*max_item_, 1); } @@ -433,8 +437,12 @@ auto kll_sketch::serialize(unsigned header_size_bytes, const SerDe& sd) ptr += copy_to_mem(min_k_, ptr); ptr += copy_to_mem(num_levels_, ptr); ptr += sizeof(uint8_t); // unused - const uint32_t levels_offset = get_levels_offset(); - for (uint8_t i = 0; i < num_levels_; ++i) { ptr += copy_to_mem(levels_[i] + levels_offset, ptr); } + if (items_size_ < k_) { + // one level in a buffer still growing to k items: write where it starts in a buffer of k items + ptr += copy_to_mem(levels_[0] + (k_ - items_size_), ptr); + } else { + ptr += copy_to_mem(levels_.data(), ptr, sizeof(levels_[0]) * num_levels_); + } ptr += sd.serialize(ptr, end_ptr - ptr, &*min_item_, 1); ptr += sd.serialize(ptr, end_ptr - ptr, &*max_item_, 1); } @@ -777,12 +785,6 @@ void kll_sketch::grow_items() { levels_[1] += delta; } -// Offset of levels_ in a buffer of k items, which serialize() and to_string() show while there is one level. -template -uint32_t kll_sketch::get_levels_offset() const { - return num_levels_ == 1 ? k_ - items_size_ : 0; -} - template void kll_sketch::sort_level_zero() { if (!is_level_zero_sorted_) { @@ -990,7 +992,8 @@ string kll_sketch::to_string(bool print_levels, bool print_items) co os << " Estimation mode: " << (is_estimation_mode() ? "true" : "false") << std::endl; os << " Levels : " << (unsigned int) num_levels_ << std::endl; os << " Sorted : " << (is_level_zero_sorted_ ? "true" : "false") << std::endl; - os << " Capacity items : " << items_size_ + get_levels_offset() << std::endl; + // a buffer still growing to k items shows the capacity of a full one + os << " Capacity items : " << std::max(items_size_, k_) << std::endl; os << " Retained items : " << get_num_retained() << std::endl; if (!is_empty()) { os << " Min item : " << *min_item_ << std::endl; From 297569601c7cc0795b3a9a121776f6b40e661516 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Milo=C5=A1=20Stojko?= Date: Mon, 9 Nov 2026 11:52:34 +0000 Subject: [PATCH 08/12] KLL: merge a sketch into itself as master does Throwing on a self-merge, as Java does, changes behavior, so it is left for a separate PR. A self-merge reads the items it moves, so it first grows the buffer to k items. It then runs on the same buffer as on master, with the same result or the same error. Co-authored-by: Isaac --- kll/include/kll_sketch.hpp | 1 - kll/include/kll_sketch_impl.hpp | 3 ++- kll/test/kll_sketch_test.cpp | 19 ++++++++++--------- 3 files changed, 12 insertions(+), 11 deletions(-) diff --git a/kll/include/kll_sketch.hpp b/kll/include/kll_sketch.hpp index 5f840fcc..196c963c 100644 --- a/kll/include/kll_sketch.hpp +++ b/kll/include/kll_sketch.hpp @@ -243,7 +243,6 @@ class kll_sketch { * Merges another sketch into this one. * If sketches contain strings, callers are responsible for ensuring that * both sketches were built using compatible string encodings. - * If other is this sketch, this throws std::invalid_argument. * @param other sketch to merge into this one */ template diff --git a/kll/include/kll_sketch_impl.hpp b/kll/include/kll_sketch_impl.hpp index f5e003fd..567566e0 100644 --- a/kll/include/kll_sketch_impl.hpp +++ b/kll/include/kll_sketch_impl.hpp @@ -214,8 +214,9 @@ uint32_t kll_sketch::internal_update() { template template void kll_sketch::merge(FwdSk&& other) { - if (this == &other) { throw std::invalid_argument("A sketch cannot merge with itself."); } if (other.is_empty()) { return; } + // a self-merge reads the items it moves, so it grows the buffer to k items first, as on master + if (this == &other) { while (items_size_ < k_) grow_items(); } if (m_ != other.m_) { throw std::invalid_argument("incompatible M: " + std::to_string(m_) + " and " + std::to_string(other.m_)); } diff --git a/kll/test/kll_sketch_test.cpp b/kll/test/kll_sketch_test.cpp index 383782eb..fe261d60 100644 --- a/kll/test/kll_sketch_test.cpp +++ b/kll/test/kll_sketch_test.cpp @@ -568,15 +568,6 @@ TEST_CASE("kll sketch", "[kll_sketch]") { REQUIRE(sketch2.get_max_item() == 999999.0f); } - SECTION("merge into itself") { - kll_float_sketch sketch(200, std::less(), 0); - REQUIRE_THROWS_WITH(sketch.merge(sketch), "A sketch cannot merge with itself."); - for (int i = 0; i < 150; i++) sketch.update(static_cast(i)); - const auto bytes = sketch.serialize(); - REQUIRE_THROWS_WITH(sketch.merge(sketch), "A sketch cannot merge with itself."); - REQUIRE(sketch.serialize() == bytes); - } - SECTION("sketch of ints") { kll_sketch sketch; REQUIRE_THROWS_AS(sketch.get_quantile(0), std::runtime_error); @@ -1123,6 +1114,16 @@ TEST_CASE("kll sketch: lazy growth of the items buffer", "[kll_sketch]") { require_identical(growing, full); } + SECTION("merge into itself as with a full buffer") { + // 300 items would grow the buffer during the merge + kll_float_sketch growing(kll_constants::MAX_K, std::less(), 0); + for (int i = 0; i < 150; ++i) { growing.update(static_cast(i)); } + kll_float_sketch full = with_full_buffer(growing); + apply_to_both(growing, full, [](kll_float_sketch& sketch) { sketch.merge(sketch); }); + REQUIRE(growing.get_n() == 300); + require_identical(growing, full); + } + // cleanup REQUIRE(test_allocator_total_bytes == 0); } From ea609a720fc799a74dd4a75d451ece236d22ff43 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Milo=C5=A1=20Stojko?= Date: Mon, 9 Nov 2026 12:26:36 +0000 Subject: [PATCH 09/12] KLL: leave self-merge to #570 #570 makes a self-merge throw, as Java does, so merge() is the same as on master here. Without #570, a self-merge of a sketch whose buffer is still growing can fail where master works, so this PR merges after #570. Co-authored-by: Isaac --- kll/include/kll_sketch_impl.hpp | 2 -- kll/test/kll_sketch_test.cpp | 10 ---------- 2 files changed, 12 deletions(-) diff --git a/kll/include/kll_sketch_impl.hpp b/kll/include/kll_sketch_impl.hpp index 567566e0..7e748ba5 100644 --- a/kll/include/kll_sketch_impl.hpp +++ b/kll/include/kll_sketch_impl.hpp @@ -215,8 +215,6 @@ template template void kll_sketch::merge(FwdSk&& other) { if (other.is_empty()) { return; } - // a self-merge reads the items it moves, so it grows the buffer to k items first, as on master - if (this == &other) { while (items_size_ < k_) grow_items(); } if (m_ != other.m_) { throw std::invalid_argument("incompatible M: " + std::to_string(m_) + " and " + std::to_string(other.m_)); } diff --git a/kll/test/kll_sketch_test.cpp b/kll/test/kll_sketch_test.cpp index fe261d60..2935a807 100644 --- a/kll/test/kll_sketch_test.cpp +++ b/kll/test/kll_sketch_test.cpp @@ -1114,16 +1114,6 @@ TEST_CASE("kll sketch: lazy growth of the items buffer", "[kll_sketch]") { require_identical(growing, full); } - SECTION("merge into itself as with a full buffer") { - // 300 items would grow the buffer during the merge - kll_float_sketch growing(kll_constants::MAX_K, std::less(), 0); - for (int i = 0; i < 150; ++i) { growing.update(static_cast(i)); } - kll_float_sketch full = with_full_buffer(growing); - apply_to_both(growing, full, [](kll_float_sketch& sketch) { sketch.merge(sketch); }); - REQUIRE(growing.get_n() == 300); - require_identical(growing, full); - } - // cleanup REQUIRE(test_allocator_total_bytes == 0); } From 78bc57768f1f8af16607b61dbba83696b5e8190c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Milo=C5=A1=20Stojko?= Date: Mon, 9 Nov 2026 12:30:40 +0000 Subject: [PATCH 10/12] KLL: remove the comments on the growing-buffer serialization branch Co-authored-by: Isaac --- kll/include/kll_sketch_impl.hpp | 2 -- 1 file changed, 2 deletions(-) diff --git a/kll/include/kll_sketch_impl.hpp b/kll/include/kll_sketch_impl.hpp index 7e748ba5..35e9c1f0 100644 --- a/kll/include/kll_sketch_impl.hpp +++ b/kll/include/kll_sketch_impl.hpp @@ -396,7 +396,6 @@ void kll_sketch::serialize(std::ostream& os, const SerDe& sd) const { write(os, num_levels_); write(os, unused); if (items_size_ < k_) { - // one level in a buffer still growing to k items: write where it starts in a buffer of k items write(os, levels_[0] + (k_ - items_size_)); } else { write(os, levels_.data(), sizeof(levels_[0]) * num_levels_); @@ -437,7 +436,6 @@ auto kll_sketch::serialize(unsigned header_size_bytes, const SerDe& sd) ptr += copy_to_mem(num_levels_, ptr); ptr += sizeof(uint8_t); // unused if (items_size_ < k_) { - // one level in a buffer still growing to k items: write where it starts in a buffer of k items ptr += copy_to_mem(levels_[0] + (k_ - items_size_), ptr); } else { ptr += copy_to_mem(levels_.data(), ptr, sizeof(levels_[0]) * num_levels_); From b4fcf81dc849d5d48b52d879c35e3e681970a316 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Milo=C5=A1=20Stojko?= Date: Mon, 9 Nov 2026 12:36:16 +0000 Subject: [PATCH 11/12] KLL benchmarks: build one sketch with many more items than k at k = 800 and 65535 Co-authored-by: Isaac --- benchmarks/benchmark_kll_sketch.cpp | 3 +++ 1 file changed, 3 insertions(+) diff --git a/benchmarks/benchmark_kll_sketch.cpp b/benchmarks/benchmark_kll_sketch.cpp index fea5b979..b44db5a0 100644 --- a/benchmarks/benchmark_kll_sketch.cpp +++ b/benchmarks/benchmark_kll_sketch.cpp @@ -132,6 +132,9 @@ void buildArgs(benchmark::internal::Benchmark * b) b->Args({k, n}); } } + // n much larger than k + b->Args({800, 1000000}); + b->Args({65535, 10000000}); } void manySketchesArgs(benchmark::internal::Benchmark * b) From 43a567340f00629dc4aaf2a96202a53fda009090 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Milo=C5=A1=20Stojko?= Date: Mon, 9 Nov 2026 13:02:26 +0000 Subject: [PATCH 12/12] KLL: remove the benchmark from this PR Co-authored-by: Isaac --- benchmarks/CMakeLists.txt | 17 --- benchmarks/benchmark_kll_sketch.cpp | 166 ---------------------------- 2 files changed, 183 deletions(-) delete mode 100644 benchmarks/benchmark_kll_sketch.cpp diff --git a/benchmarks/CMakeLists.txt b/benchmarks/CMakeLists.txt index 342e4478..f543dfce 100644 --- a/benchmarks/CMakeLists.txt +++ b/benchmarks/CMakeLists.txt @@ -50,20 +50,3 @@ target_sources(count_min_sketch_benchmark benchmark_count_min_sketch.cpp benchmark_count_min_sketch_serialization.cpp ) - -add_executable(kll_sketch_benchmark) - -target_link_libraries(kll_sketch_benchmark - PRIVATE - kll - benchmark::benchmark_main -) - -set_target_properties(kll_sketch_benchmark PROPERTIES - CXX_STANDARD_REQUIRED YES -) - -target_sources(kll_sketch_benchmark - PRIVATE - benchmark_kll_sketch.cpp -) diff --git a/benchmarks/benchmark_kll_sketch.cpp b/benchmarks/benchmark_kll_sketch.cpp deleted file mode 100644 index b44db5a0..00000000 --- a/benchmarks/benchmark_kll_sketch.cpp +++ /dev/null @@ -1,166 +0,0 @@ -/* - * 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 - -namespace -{ - -using Sketch = datasketches::kll_sketch; - -std::vector makeItems(size_t size) -{ - std::vector items; - items.reserve(size); - uint64_t state = 1; - for (size_t i = 0; i < size; ++i) - { - state = state * 6364136223846793005ULL + 1442695040888963407ULL; - items.push_back(static_cast(state >> 11)); - } - return items; -} - -// A new sketch with k = range(0) gets range(1) items. -void BM_KllBuild(benchmark::State & state) -{ - const auto k = static_cast(state.range(0)); - const auto items = makeItems(static_cast(state.range(1))); - for (auto _ : state) - { - Sketch sketch(k); - for (const double item : items) - sketch.update(item); - - benchmark::DoNotOptimize(sketch.get_n()); - benchmark::ClobberMemory(); - } - - state.SetItemsProcessed(state.iterations() * static_cast(items.size())); -} - -// Many sketches with k = range(0) are alive at once, each with range(1) items, like the groups of -// a GROUP BY. -void BM_KllBuildManySketches(benchmark::State & state) -{ - constexpr size_t NUM_SKETCHES = 1000; - const auto k = static_cast(state.range(0)); - const auto items = makeItems(static_cast(state.range(1))); - for (auto _ : state) - { - std::vector sketches(NUM_SKETCHES, Sketch(k)); - for (auto & sketch : sketches) - for (const double item : items) - sketch.update(item); - - benchmark::DoNotOptimize(sketches.data()); - benchmark::ClobberMemory(); - } - - state.SetItemsProcessed(state.iterations() * static_cast(NUM_SKETCHES * items.size())); -} - -// A new sketch with k = range(0) merges 100 sketches of range(1) items each. -void BM_KllMergeInto(benchmark::State & state) -{ - constexpr size_t NUM_INPUTS = 100; - const auto k = static_cast(state.range(0)); - const auto items = makeItems(static_cast(state.range(1))); - std::vector inputs(NUM_INPUTS, Sketch(k)); - for (auto & input : inputs) - for (const double item : items) - input.update(item); - - for (auto _ : state) - { - Sketch sketch(k); - for (const auto & input : inputs) - sketch.merge(input); - - benchmark::DoNotOptimize(sketch.get_n()); - benchmark::ClobberMemory(); - } - - state.SetItemsProcessed(state.iterations() * static_cast(NUM_INPUTS * items.size())); -} - -// Serializes a sketch with k = range(0) and range(1) items. -void BM_KllSerialize(benchmark::State & state) -{ - Sketch sketch(static_cast(state.range(0))); - for (const double item : makeItems(static_cast(state.range(1)))) - sketch.update(item); - - for (auto _ : state) - { - auto bytes = sketch.serialize(); - benchmark::DoNotOptimize(bytes.data()); - benchmark::ClobberMemory(); - } -} - -void buildArgs(benchmark::internal::Benchmark * b) -{ - for (const int64_t k : {200, 1000, 5000, 20000, 65535}) - { - // n = k fills the buffer up to k with no compaction, so it includes every growth step - for (const int64_t n : {int64_t{1}, int64_t{10}, int64_t{100}, int64_t{1000}, k, int64_t{10000}, - int64_t{100000}, int64_t{1000000}}) - { - if (n <= 100 * k) - b->Args({k, n}); - } - } - // n much larger than k - b->Args({800, 1000000}); - b->Args({65535, 10000000}); -} - -void manySketchesArgs(benchmark::internal::Benchmark * b) -{ - for (const int64_t k : {200, 1000, 5000, 20000, 65535}) - for (const int64_t n : {1, 10, 100, 1000}) - b->Args({k, n}); -} - -void mergeArgs(benchmark::internal::Benchmark * b) -{ - for (const int64_t k : {200, 5000, 65535}) - for (const int64_t n : {10, 100, 1000}) - b->Args({k, n}); -} - -void serializeArgs(benchmark::internal::Benchmark * b) -{ - for (const int64_t k : {200, 5000, 65535}) - for (const int64_t n : {10, 1000, 100000}) - b->Args({k, n}); -} - -} - -BENCHMARK(BM_KllBuild)->Apply(buildArgs); -BENCHMARK(BM_KllBuildManySketches)->Apply(manySketchesArgs)->Unit(benchmark::kMicrosecond); -BENCHMARK(BM_KllMergeInto)->Apply(mergeArgs)->Unit(benchmark::kMicrosecond); -BENCHMARK(BM_KllSerialize)->Apply(serializeArgs);