From 39f7bf33291dc6ec3af75b10454e42fb6e8e87ab Mon Sep 17 00:00:00 2001 From: Waterpine Date: Thu, 5 Nov 2026 12:32:00 -0600 Subject: [PATCH 1/3] perf(frequencies): insert known-absent keys without comparisons when rehashing Growing the map re-inserted distinct keys through `adjust_or_put_value`, which compares keys along each probe path. Rehashing now uses `insert_absent`, which follows the same probe sequence and drift counting without the comparisons, so the resulting table is identical. Add tests that grow the map through several doublings and check every estimate. Co-Authored-By: Claude Sonnet 5.5 --- .../reverse_purge_item_hash_map.rs | 20 ++++- .../tests/frequencies_test/main.rs | 1 + .../tests/frequencies_test/rehash.rs | 85 +++++++++++++++++++ 3 files changed, 105 insertions(+), 1 deletion(-) create mode 100644 tests-integration/tests/frequencies_test/rehash.rs diff --git a/datasketches/src/frequencies/reverse_purge_item_hash_map.rs b/datasketches/src/frequencies/reverse_purge_item_hash_map.rs index 7f430f65..bb5ebff3 100644 --- a/datasketches/src/frequencies/reverse_purge_item_hash_map.rs +++ b/datasketches/src/frequencies/reverse_purge_item_hash_map.rs @@ -171,12 +171,30 @@ impl ReversePurgeItemHashMap { for i in 0..old_keys.len() { if old_states[i] > 0 { if let Some(key) = old_keys[i].take() { - self.adjust_or_put_value(key, old_values[i]); + self.insert_absent(key, old_values[i]); } } } } + /// Inserts a key that is known to be absent, such as when rehashing distinct keys. + /// + /// This probes the same slots as [`Self::adjust_or_put_value`] but skips key comparisons. + fn insert_absent(&mut self, key: T, value: u64) { + let mask = self.keys.len() - 1; + let mut probe = (hash_item(&key) as usize) & mask; + let mut drift: usize = 1; + while self.states[probe] > 0 { + probe = (probe + 1) & mask; + drift += 1; + debug_assert!(drift < DRIFT_LIMIT, "drift limit exceeded"); + } + self.keys[probe] = Some(key); + self.values[probe] = value; + self.states[probe] = drift as u16; + self.num_active += 1; + } + /// Returns the length of the underlying arrays. pub fn len(&self) -> usize { self.keys.len() diff --git a/tests-integration/tests/frequencies_test/main.rs b/tests-integration/tests/frequencies_test/main.rs index 1dea33be..6181db4f 100644 --- a/tests-integration/tests/frequencies_test/main.rs +++ b/tests-integration/tests/frequencies_test/main.rs @@ -15,4 +15,5 @@ // specific language governing permissions and limitations // under the License. +mod rehash; mod update; diff --git a/tests-integration/tests/frequencies_test/rehash.rs b/tests-integration/tests/frequencies_test/rehash.rs new file mode 100644 index 00000000..5944c4ee --- /dev/null +++ b/tests-integration/tests/frequencies_test/rehash.rs @@ -0,0 +1,85 @@ +// 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. + +//! Growing the item map must keep every item with its exact weight. + +use datasketches::frequencies::FrequentItemsSketch; + +fn weight(i: u64) -> u64 { + i % 5 + 1 +} + +/// Feeds distinct items below the purge threshold, so the map doubles several times and no +/// counter is ever decremented, and checks every estimate after each batch. +fn check_growth( + max_map_size: usize, + distinct: u64, + item: impl Fn(u64) -> T, + estimate: impl Fn(&FrequentItemsSketch, &T) -> u64, +) { + let mut sketch = FrequentItemsSketch::new(max_map_size).unwrap(); + let initial_capacity = sketch.current_map_capacity(); + let mut total_weight = 0; + for i in 0..distinct { + sketch.update_with_count(item(i), weight(i)); + total_weight += weight(i); + if (i + 1) % 250 == 0 || i + 1 == distinct { + assert_eq!(sketch.total_weight(), total_weight); + assert_eq!(sketch.num_active_items() as u64, i + 1); + assert_eq!(sketch.maximum_error(), 0); + for j in 0..=i { + assert_eq!(estimate(&sketch, &item(j)), weight(j), "item {j} after {i}"); + } + } + } + assert!(sketch.current_map_capacity() > initial_capacity); +} + +#[test] +fn test_longs_growth_keeps_every_item() { + check_growth( + 1 << 12, + 2_000, + |i| (i * 7_919 + 13) as i64, + |sketch, item| sketch.estimate(item), + ); +} + +#[test] +fn test_items_growth_keeps_every_item() { + check_growth( + 1 << 12, + 2_000, + |i| format!("item-{i}"), + |sketch, item| sketch.estimate(item), + ); +} + +#[test] +fn test_growth_survives_serialization() { + let mut sketch: FrequentItemsSketch = FrequentItemsSketch::new(1 << 12).unwrap(); + for i in 0..2_000 { + sketch.update_with_count((i * 7_919 + 13) as i64, weight(i)); + } + let bytes = sketch.serialize(); + let restored = FrequentItemsSketch::::deserialize(&bytes).unwrap(); + assert_eq!(restored.serialize(), bytes); + assert_eq!(restored.total_weight(), sketch.total_weight()); + for i in 0..2_000 { + assert_eq!(restored.estimate(&((i * 7_919 + 13) as i64)), weight(i)); + } +} From 892b2b47c3fac55b0cb224498d86e2d39c110ee1 Mon Sep 17 00:00:00 2001 From: tison Date: Sat, 7 Nov 2026 14:52:25 +0800 Subject: [PATCH 2/3] test(frequencies): cover collision chains across rehash and purge Growth-only estimate checks cannot detect incorrect probe distances until a later purge deletes entries from a collision chain. Exercise a non-Clone key type with a constant hash through repeated growth, duplicate updates, a weighted purge, and reinsertion. Verify retained counts, error bounds, active entries, and total stream weight through the public sketch API. Check all existing items at each actual growth boundary instead of arbitrary batches, and remove the unused Clone bound and identical estimate callbacks from the growth helper. Keep the growth serialization round trip in the dedicated serialization test target, including its retained-count and zero-error checks. Validation: cargo x check, cargo x test, cargo x lint, and cargo x bench -- --test pass with the pinned TCK fixtures. Temporarily replacing rehash probe distances with 1 makes the collision test fail after purge, confirming that the regression test detects the intended invariant violation. --- .../tests/frequencies_test/rehash.rs | 90 ++++++++++++------- .../tests/serde_tests/frequencies.rs | 17 ++++ 2 files changed, 75 insertions(+), 32 deletions(-) diff --git a/tests-integration/tests/frequencies_test/rehash.rs b/tests-integration/tests/frequencies_test/rehash.rs index 5944c4ee..e166c55f 100644 --- a/tests-integration/tests/frequencies_test/rehash.rs +++ b/tests-integration/tests/frequencies_test/rehash.rs @@ -17,32 +17,30 @@ //! Growing the item map must keep every item with its exact weight. +use std::hash::Hash; +use std::hash::Hasher; + use datasketches::frequencies::FrequentItemsSketch; fn weight(i: u64) -> u64 { i % 5 + 1 } -/// Feeds distinct items below the purge threshold, so the map doubles several times and no -/// counter is ever decremented, and checks every estimate after each batch. -fn check_growth( - max_map_size: usize, - distinct: u64, - item: impl Fn(u64) -> T, - estimate: impl Fn(&FrequentItemsSketch, &T) -> u64, -) { +// Stay below the purge threshold and check all counts at each growth boundary. +fn check_growth(max_map_size: usize, distinct: u64, item: impl Fn(u64) -> T) { let mut sketch = FrequentItemsSketch::new(max_map_size).unwrap(); let initial_capacity = sketch.current_map_capacity(); let mut total_weight = 0; for i in 0..distinct { + let capacity = sketch.current_map_capacity(); sketch.update_with_count(item(i), weight(i)); total_weight += weight(i); - if (i + 1) % 250 == 0 || i + 1 == distinct { + if sketch.current_map_capacity() != capacity || i + 1 == distinct { assert_eq!(sketch.total_weight(), total_weight); assert_eq!(sketch.num_active_items() as u64, i + 1); assert_eq!(sketch.maximum_error(), 0); for j in 0..=i { - assert_eq!(estimate(&sketch, &item(j)), weight(j), "item {j} after {i}"); + assert_eq!(sketch.estimate(&item(j)), weight(j), "item {j} after {i}"); } } } @@ -51,35 +49,63 @@ fn check_growth( #[test] fn test_longs_growth_keeps_every_item() { - check_growth( - 1 << 12, - 2_000, - |i| (i * 7_919 + 13) as i64, - |sketch, item| sketch.estimate(item), - ); + check_growth(1 << 12, 2_000, |i| (i * 7_919 + 13) as i64); } #[test] fn test_items_growth_keeps_every_item() { - check_growth( - 1 << 12, - 2_000, - |i| format!("item-{i}"), - |sketch, item| sketch.estimate(item), - ); + check_growth(1 << 12, 2_000, |i| format!("item-{i}")); +} + +#[derive(Debug, PartialEq, Eq)] +struct CollidingItem(u64); + +impl Hash for CollidingItem { + fn hash(&self, state: &mut H) { + 0u64.hash(state); + } } #[test] -fn test_growth_survives_serialization() { - let mut sketch: FrequentItemsSketch = FrequentItemsSketch::new(1 << 12).unwrap(); - for i in 0..2_000 { - sketch.update_with_count((i * 7_919 + 13) as i64, weight(i)); +fn test_colliding_items_survive_growth_updates_and_purge() { + let mut sketch = FrequentItemsSketch::new(64).unwrap(); + for i in 0..48 { + sketch.update_with_count(CollidingItem(i), if i % 4 == 0 { 100 } else { 1 }); } - let bytes = sketch.serialize(); - let restored = FrequentItemsSketch::::deserialize(&bytes).unwrap(); - assert_eq!(restored.serialize(), bytes); - assert_eq!(restored.total_weight(), sketch.total_weight()); - for i in 0..2_000 { - assert_eq!(restored.estimate(&((i * 7_919 + 13) as i64)), weight(i)); + assert_eq!(sketch.current_map_capacity(), sketch.maximum_map_capacity()); + assert_eq!(sketch.maximum_error(), 0); + for i in 0..48 { + assert_eq!( + sketch.estimate(&CollidingItem(i)), + if i % 4 == 0 { 100 } else { 1 } + ); } + + for i in (0..48).step_by(4) { + sketch.update_with_count(CollidingItem(i), 7); + } + assert_eq!(sketch.num_active_items(), 48); + assert_eq!(sketch.maximum_error(), 0); + + // The new key triggers a purge with median 1. Deleting light items from + // the collision chain must preserve access to every displaced heavy item. + sketch.update(CollidingItem(48)); + assert_eq!(sketch.maximum_error(), 1); + assert_eq!(sketch.num_active_items(), 12); + for i in 0..49 { + if i < 48 && i % 4 == 0 { + assert_eq!(sketch.estimate(&CollidingItem(i)), 107); + assert_eq!(sketch.lower_bound(&CollidingItem(i)), 106); + } else { + assert_eq!(sketch.estimate(&CollidingItem(i)), 0); + assert_eq!(sketch.upper_bound(&CollidingItem(i)), 1); + } + } + + sketch.update_with_count(CollidingItem(44), 3); + sketch.update_with_count(CollidingItem(1), 5); + assert_eq!(sketch.estimate(&CollidingItem(44)), 110); + assert_eq!(sketch.estimate(&CollidingItem(1)), 6); + assert_eq!(sketch.num_active_items(), 13); + assert_eq!(sketch.total_weight(), 12 * 107 + 37 + 3 + 5); } diff --git a/tests-integration/tests/serde_tests/frequencies.rs b/tests-integration/tests/serde_tests/frequencies.rs index 9c1df36b..dd301ccf 100644 --- a/tests-integration/tests/serde_tests/frequencies.rs +++ b/tests-integration/tests/serde_tests/frequencies.rs @@ -64,6 +64,23 @@ fn test_longs_round_trip() { assert_eq!(restored.maximum_error(), sketch.maximum_error()); } +#[test] +fn test_growth_survives_serialization() { + let mut sketch = FrequentItemsSketch::::new(1 << 12).unwrap(); + for i in 0..2_000 { + sketch.update_with_count(i * 7_919 + 13, (i % 5 + 1) as u64); + } + let bytes = sketch.serialize(); + let restored = FrequentItemsSketch::::deserialize(&bytes).unwrap(); + assert_eq!(restored.serialize(), bytes); + assert_eq!(restored.total_weight(), sketch.total_weight()); + assert_eq!(restored.num_active_items(), 2_000); + assert_eq!(restored.maximum_error(), 0); + for i in 0..2_000 { + assert_eq!(restored.estimate(&(i * 7_919 + 13)), (i % 5 + 1) as u64); + } +} + #[test] fn test_items_round_trip() { let mut sketch = FrequentItemsSketch::new(32).unwrap(); From 1507af237139308e6087f9e4ac38379b02245eb1 Mon Sep 17 00:00:00 2001 From: tison Date: Sat, 7 Nov 2026 14:52:25 +0800 Subject: [PATCH 3/3] docs(changelog): record faster growing frequency sketches Document the user-visible improvement to FrequentItemsSketch updates for distinct-item streams while the map grows. Scope the claim to the measured workload; repeated-key streams and sketches at their maximum map size do not generally benefit from skipping comparisons during rehash. Compared main at 4618600 with the updated PR at 6185c87 on an Apple M4 Max running macOS 26.6.2 and rustc 1.101.0-nightly (8006091eb). Ran the unchanged frequencies::update workloads with --repeat 5 --raw in three baseline/PR pairs, reversing order for the middle pair. Each revision has 15 samples at the registered fixed round counts (2000 for u64, 900 for bytes_32). Aggregate median and full range in milliseconds: u64 301 [264, 362] to 176 [168, 198], ratio 0.58; bytes_32 323 [318, 329] to 268 [265, 285], ratio 0.83. These are whole-workload elapsed times, including construction and cleanup, on one machine. No production code or benchmark workload changed after the measured PR revision. --- CHANGELOG.md | 1 + 1 file changed, 1 insertion(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 7d7158fb..545914c5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,6 +12,7 @@ All significant changes to this project will be documented in this file. ### Improvements +* `FrequentItemsSketch` updates are faster for streams of distinct items while the map is growing. * `HllUnion` is faster when adopting HLL8 inputs without downsampling or merging multiple dense inputs before querying or serializing the result. * `KllSketch::rank` is faster for repeated queries on compacted sketches. * Repeated scalar `TDigest::rank` queries are faster, including small digests and data with repeated values.