From e02de1b68b1329b341508c3ccf957fe28b9cc62d Mon Sep 17 00:00:00 2001 From: kedom Date: Thu, 25 Jun 2026 21:55:36 +0200 Subject: [PATCH 1/5] feat(frequencies): add borrowed-key update methods Add FrequentItemsSketch::update_borrowed and update_with_count_borrowed, backed by ReversePurgeItemHashMap::adjust_or_put_value_borrowed. They take the item by reference and only allocate an owned key on insertion, so a caller streaming many repeated items can increment an already-tracked item without allocating an owned key per occurrence. The owned update/update_with_count path allocates on every call, even when the item already exists and the owned value is immediately dropped. hash_item is relaxed to T: Hash + ?Sized so unsized borrowed keys such as str can be hashed directly. Co-Authored-By: Claude Opus 4.8 (1M context) Signed-off-by: kedom --- .../reverse_purge_item_hash_map.rs | 40 +++++++++++++- datasketches/src/frequencies/sketch.rs | 52 +++++++++++++++++++ 2 files changed, 91 insertions(+), 1 deletion(-) diff --git a/datasketches/src/frequencies/reverse_purge_item_hash_map.rs b/datasketches/src/frequencies/reverse_purge_item_hash_map.rs index 79ed290..16024eb 100644 --- a/datasketches/src/frequencies/reverse_purge_item_hash_map.rs +++ b/datasketches/src/frequencies/reverse_purge_item_hash_map.rs @@ -20,6 +20,7 @@ //! This linear-probing hash map supports a reverse purge operation that removes //! keys with non-positive counts by scanning clusters from the back to the front. +use std::borrow::Borrow; use std::hash::Hash; use std::hash::Hasher; @@ -94,6 +95,43 @@ impl ReversePurgeItemHashMap { } } + /// Adds `adjust_amount` to the value for a borrowed `key`, inserting if absent. + /// + /// Behaves like [`adjust_or_put_value`](Self::adjust_or_put_value) but takes + /// the key by reference and only allocates an owned key (via `key.to_owned()`) + /// on the insert path, so incrementing an already-present key is allocation + /// free. The `Borrow` contract guarantees the borrowed and owned forms hash + /// and compare identically, so lookups land on the same slot as the owned API. + pub fn adjust_or_put_value_borrowed(&mut self, key: &Q, adjust_amount: u64) + where + T: Borrow, + Q: Eq + Hash + ToOwned + ?Sized, + { + 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 { + let matches = self.keys[probe] + .as_ref() + .map(|existing| existing.borrow() == key) + .unwrap_or(false); + if matches { + break; + } + probe = (probe + 1) & mask; + drift += 1; + debug_assert!(drift < DRIFT_LIMIT, "drift limit exceeded"); + } + if self.states[probe] == 0 { + self.keys[probe] = Some(key.to_owned()); + self.values[probe] = adjust_amount; + self.states[probe] = drift as u16; + self.num_active += 1; + } else { + self.values[probe] += adjust_amount; + } + } + /// Removes all keys with non-positive counts. fn keep_only_positive_counts(&mut self) { let len = self.keys.len(); @@ -312,7 +350,7 @@ impl<'a, T> Iterator for ReversePurgeItemIter<'a, T> { } #[inline] -fn hash_item(item: &T) -> u64 { +fn hash_item(item: &T) -> u64 { let mut hasher = MurmurHash3X64128::default(); item.hash(&mut hasher); hasher.finish() diff --git a/datasketches/src/frequencies/sketch.rs b/datasketches/src/frequencies/sketch.rs index fc2b297..716f9f8 100644 --- a/datasketches/src/frequencies/sketch.rs +++ b/datasketches/src/frequencies/sketch.rs @@ -17,6 +17,7 @@ //! Frequent items sketch implementations. +use std::borrow::Borrow; use std::hash::Hash; use crate::codec::SketchBytes; @@ -263,6 +264,57 @@ impl FrequentItemsSketch { self.maybe_resize_or_purge(); } + /// Updates the sketch with a borrowed item and a count of one. + /// + /// Equivalent to [`update`](Self::update) but takes the item by reference and + /// only allocates an owned item when it is newly inserted, so updating an + /// already-tracked item is allocation free. + /// + /// # Examples + /// + /// ``` + /// # use datasketches::frequencies::FrequentItemsSketch; + /// let mut sketch = FrequentItemsSketch::::new(64); + /// sketch.update_borrowed("nginx"); + /// sketch.update_borrowed("nginx"); // no allocation on the second hit + /// assert!(sketch.estimate(&"nginx".to_string()) >= 2); + /// ``` + pub fn update_borrowed(&mut self, item: &Q) + where + T: Borrow, + Q: Eq + Hash + ToOwned + ?Sized, + { + self.update_with_count_borrowed(item, 1); + } + + /// Updates the sketch with a borrowed item and count. + /// + /// Equivalent to [`update_with_count`](Self::update_with_count) but takes the + /// item by reference and only allocates an owned item when it is newly + /// inserted. A count of zero is a no-op. + /// + /// # Examples + /// + /// ``` + /// # use datasketches::frequencies::FrequentItemsSketch; + /// let mut sketch = FrequentItemsSketch::::new(64); + /// sketch.update_with_count_borrowed("gzip", 3); + /// assert!(sketch.estimate(&"gzip".to_string()) >= 3); + /// ``` + pub fn update_with_count_borrowed(&mut self, item: &Q, count: u64) + where + T: Borrow, + Q: Eq + Hash + ToOwned + ?Sized, + { + if count == 0 { + return; + } + assert!(count > 0, "count may not be negative"); + self.stream_weight += count; + self.hash_map.adjust_or_put_value_borrowed(item, count); + self.maybe_resize_or_purge(); + } + /// Merges another sketch into this one. /// /// The other sketch may have a different map size. The merged sketch respects the From 5c88f6a6da2b9d554959364d2b990eeb04e44fd4 Mon Sep 17 00:00:00 2001 From: kedom Date: Fri, 26 Jun 2026 13:36:38 +0200 Subject: [PATCH 2/5] wip: methods generic over Borrow --- datasketches/src/frequencies/sketch.rs | 34 +++++++++++++++++--------- 1 file changed, 23 insertions(+), 11 deletions(-) diff --git a/datasketches/src/frequencies/sketch.rs b/datasketches/src/frequencies/sketch.rs index 716f9f8..5c1b110 100644 --- a/datasketches/src/frequencies/sketch.rs +++ b/datasketches/src/frequencies/sketch.rs @@ -159,7 +159,11 @@ impl FrequentItemsSketch { /// sketch.update_with_count(10, 2); /// assert!(sketch.estimate(&10) >= 2); /// ``` - pub fn estimate(&self, item: &T) -> u64 { + pub fn estimate(&self, item: &Q) -> u64 + where + T: Borrow, + Q: Eq + Hash + ?Sized, + { let value = self.hash_map.get(item); if value > 0 { value + self.offset } else { 0 } } @@ -168,7 +172,11 @@ impl FrequentItemsSketch { /// /// This value is guaranteed to be no larger than the true frequency. If the item is not /// tracked, the lower bound is zero. - pub fn lower_bound(&self, item: &T) -> u64 { + pub fn lower_bound(&self, item: &Q) -> u64 + where + T: Borrow, + Q: Eq + Hash + ?Sized, + { self.hash_map.get(item) } @@ -176,7 +184,11 @@ impl FrequentItemsSketch { /// /// This value is guaranteed to be no smaller than the true frequency. If the item is tracked, /// this is `item_count + offset`. - pub fn upper_bound(&self, item: &T) -> u64 { + pub fn upper_bound(&self, item: &Q) -> u64 + where + T: Borrow, + Q: Eq + Hash + ?Sized, + { self.hash_map.get(item) + self.offset } @@ -277,12 +289,12 @@ impl FrequentItemsSketch { /// let mut sketch = FrequentItemsSketch::::new(64); /// sketch.update_borrowed("nginx"); /// sketch.update_borrowed("nginx"); // no allocation on the second hit - /// assert!(sketch.estimate(&"nginx".to_string()) >= 2); + /// assert!(sketch.estimate("nginx") >= 2); /// ``` - pub fn update_borrowed(&mut self, item: &Q) + pub fn update_borrowed<'a, Q>(&mut self, item: &'a Q) where - T: Borrow, - Q: Eq + Hash + ToOwned + ?Sized, + T: Borrow + From<&'a Q>, + Q: Eq + Hash + ?Sized, { self.update_with_count_borrowed(item, 1); } @@ -299,12 +311,12 @@ impl FrequentItemsSketch { /// # use datasketches::frequencies::FrequentItemsSketch; /// let mut sketch = FrequentItemsSketch::::new(64); /// sketch.update_with_count_borrowed("gzip", 3); - /// assert!(sketch.estimate(&"gzip".to_string()) >= 3); + /// assert!(sketch.estimate("gzip") >= 3); /// ``` - pub fn update_with_count_borrowed(&mut self, item: &Q, count: u64) + pub fn update_with_count_borrowed<'a, Q>(&mut self, item: &'a Q, count: u64) where - T: Borrow, - Q: Eq + Hash + ToOwned + ?Sized, + T: Borrow + From<&'a Q>, + Q: Eq + Hash + ?Sized, { if count == 0 { return; From 2be97b8230b58e7983a8c70809d27d2e65dca8da Mon Sep 17 00:00:00 2001 From: kedom Date: Fri, 26 Jun 2026 13:37:18 +0200 Subject: [PATCH 3/5] wip: use From<&Q> instead of ToOwned --- .../reverse_purge_item_hash_map.rs | 27 +++++++++++++------ 1 file changed, 19 insertions(+), 8 deletions(-) diff --git a/datasketches/src/frequencies/reverse_purge_item_hash_map.rs b/datasketches/src/frequencies/reverse_purge_item_hash_map.rs index 16024eb..58cfd3f 100644 --- a/datasketches/src/frequencies/reverse_purge_item_hash_map.rs +++ b/datasketches/src/frequencies/reverse_purge_item_hash_map.rs @@ -60,7 +60,14 @@ impl ReversePurgeItemHashMap { } /// Returns the value for `key`, or zero if the key is not present. - pub fn get(&self, key: &T) -> u64 { + /// + /// The key may be any borrowed form of `T` (e.g. `&str` for a `String` map), + /// matching the `HashMap::get` convention. + pub fn get(&self, key: &Q) -> u64 + where + T: Borrow, + Q: Eq + Hash + ?Sized, + { let probe = self.hash_probe(key); if self.states[probe] > 0 { return self.values[probe]; @@ -98,14 +105,14 @@ impl ReversePurgeItemHashMap { /// Adds `adjust_amount` to the value for a borrowed `key`, inserting if absent. /// /// Behaves like [`adjust_or_put_value`](Self::adjust_or_put_value) but takes - /// the key by reference and only allocates an owned key (via `key.to_owned()`) + /// the key by reference and only allocates an owned key (via `T: From<&Q>`) /// on the insert path, so incrementing an already-present key is allocation /// free. The `Borrow` contract guarantees the borrowed and owned forms hash /// and compare identically, so lookups land on the same slot as the owned API. - pub fn adjust_or_put_value_borrowed(&mut self, key: &Q, adjust_amount: u64) + pub fn adjust_or_put_value_borrowed<'a, Q>(&mut self, key: &'a Q, adjust_amount: u64) where - T: Borrow, - Q: Eq + Hash + ToOwned + ?Sized, + T: Borrow + From<&'a Q>, + Q: Eq + Hash + ?Sized, { let mask = self.keys.len() - 1; let mut probe = (hash_item(key) as usize) & mask; @@ -123,7 +130,7 @@ impl ReversePurgeItemHashMap { debug_assert!(drift < DRIFT_LIMIT, "drift limit exceeded"); } if self.states[probe] == 0 { - self.keys[probe] = Some(key.to_owned()); + self.keys[probe] = Some(key.into()); self.values[probe] = adjust_amount; self.states[probe] = drift as u16; self.num_active += 1; @@ -266,13 +273,17 @@ impl ReversePurgeItemHashMap { self.states[probe] > 0 } - fn hash_probe(&self, key: &T) -> usize { + fn hash_probe(&self, key: &Q) -> usize + where + T: Borrow, + Q: Eq + Hash + ?Sized, + { let mask = self.keys.len() - 1; let mut probe = (hash_item(key) as usize) & mask; while self.states[probe] > 0 { let matches = self.keys[probe] .as_ref() - .map(|existing| existing == key) + .map(|existing| existing.borrow() == key) .unwrap_or(false); if matches { break; From 0e3864b1ce20a4ef02471485c9af7d6e6055a348 Mon Sep 17 00:00:00 2001 From: kedom Date: Sun, 28 Jun 2026 16:46:58 +0200 Subject: [PATCH 4/5] wip: align with hashmaps `entry_ref` --- .../reverse_purge_item_hash_map.rs | 17 +++++++------- datasketches/src/frequencies/sketch.rs | 22 +++++++++---------- 2 files changed, 20 insertions(+), 19 deletions(-) diff --git a/datasketches/src/frequencies/reverse_purge_item_hash_map.rs b/datasketches/src/frequencies/reverse_purge_item_hash_map.rs index 58cfd3f..9f853e5 100644 --- a/datasketches/src/frequencies/reverse_purge_item_hash_map.rs +++ b/datasketches/src/frequencies/reverse_purge_item_hash_map.rs @@ -105,14 +105,15 @@ impl ReversePurgeItemHashMap { /// Adds `adjust_amount` to the value for a borrowed `key`, inserting if absent. /// /// Behaves like [`adjust_or_put_value`](Self::adjust_or_put_value) but takes - /// the key by reference and only allocates an owned key (via `T: From<&Q>`) - /// on the insert path, so incrementing an already-present key is allocation - /// free. The `Borrow` contract guarantees the borrowed and owned forms hash - /// and compare identically, so lookups land on the same slot as the owned API. - pub fn adjust_or_put_value_borrowed<'a, Q>(&mut self, key: &'a Q, adjust_amount: u64) + /// the key by reference and only allocates an owned key (via + /// `Q: ToOwned`) on the insert path, so incrementing an + /// already-present key is allocation free. The `Borrow` contract guarantees + /// the borrowed and owned forms hash and compare identically, so lookups land + /// on the same slot as the owned API. + pub fn adjust_or_put_value_ref(&mut self, key: &Q, adjust_amount: u64) where - T: Borrow + From<&'a Q>, - Q: Eq + Hash + ?Sized, + T: Borrow, + Q: Eq + Hash + ToOwned + ?Sized, { let mask = self.keys.len() - 1; let mut probe = (hash_item(key) as usize) & mask; @@ -130,7 +131,7 @@ impl ReversePurgeItemHashMap { debug_assert!(drift < DRIFT_LIMIT, "drift limit exceeded"); } if self.states[probe] == 0 { - self.keys[probe] = Some(key.into()); + self.keys[probe] = Some(key.to_owned()); self.values[probe] = adjust_amount; self.states[probe] = drift as u16; self.num_active += 1; diff --git a/datasketches/src/frequencies/sketch.rs b/datasketches/src/frequencies/sketch.rs index 5c1b110..f46a231 100644 --- a/datasketches/src/frequencies/sketch.rs +++ b/datasketches/src/frequencies/sketch.rs @@ -287,16 +287,16 @@ impl FrequentItemsSketch { /// ``` /// # use datasketches::frequencies::FrequentItemsSketch; /// let mut sketch = FrequentItemsSketch::::new(64); - /// sketch.update_borrowed("nginx"); - /// sketch.update_borrowed("nginx"); // no allocation on the second hit + /// sketch.update_ref("nginx"); + /// sketch.update_ref("nginx"); // no allocation on the second hit /// assert!(sketch.estimate("nginx") >= 2); /// ``` - pub fn update_borrowed<'a, Q>(&mut self, item: &'a Q) + pub fn update_ref(&mut self, item: &Q) where - T: Borrow + From<&'a Q>, - Q: Eq + Hash + ?Sized, + T: Borrow, + Q: Eq + Hash + ToOwned + ?Sized, { - self.update_with_count_borrowed(item, 1); + self.update_with_count_ref(item, 1); } /// Updates the sketch with a borrowed item and count. @@ -310,20 +310,20 @@ impl FrequentItemsSketch { /// ``` /// # use datasketches::frequencies::FrequentItemsSketch; /// let mut sketch = FrequentItemsSketch::::new(64); - /// sketch.update_with_count_borrowed("gzip", 3); + /// sketch.update_with_count_ref("gzip", 3); /// assert!(sketch.estimate("gzip") >= 3); /// ``` - pub fn update_with_count_borrowed<'a, Q>(&mut self, item: &'a Q, count: u64) + pub fn update_with_count_ref(&mut self, item: &Q, count: u64) where - T: Borrow + From<&'a Q>, - Q: Eq + Hash + ?Sized, + T: Borrow, + Q: Eq + Hash + ToOwned + ?Sized, { if count == 0 { return; } assert!(count > 0, "count may not be negative"); self.stream_weight += count; - self.hash_map.adjust_or_put_value_borrowed(item, count); + self.hash_map.adjust_or_put_value_ref(item, count); self.maybe_resize_or_purge(); } From 70012b6e59b2504311ab110316c02124b38fb2e7 Mon Sep 17 00:00:00 2001 From: tison Date: Thu, 2 Jul 2026 10:05:50 +0800 Subject: [PATCH 5/5] fixup Signed-off-by: tison --- .../reverse_purge_item_hash_map.rs | 48 ++++--------------- datasketches/tests/frequencies_update_test.rs | 28 +++++++++++ 2 files changed, 36 insertions(+), 40 deletions(-) diff --git a/datasketches/src/frequencies/reverse_purge_item_hash_map.rs b/datasketches/src/frequencies/reverse_purge_item_hash_map.rs index 9f853e5..0e59745 100644 --- a/datasketches/src/frequencies/reverse_purge_item_hash_map.rs +++ b/datasketches/src/frequencies/reverse_purge_item_hash_map.rs @@ -68,7 +68,7 @@ impl ReversePurgeItemHashMap { T: Borrow, Q: Eq + Hash + ?Sized, { - let probe = self.hash_probe(key); + let (probe, _) = self.find_probe_or_empty(key); if self.states[probe] > 0 { return self.values[probe]; } @@ -77,21 +77,7 @@ impl ReversePurgeItemHashMap { /// Adds `adjust_amount` to the value for `key`, inserting if absent. pub fn adjust_or_put_value(&mut self, key: T, adjust_amount: 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 { - let matches = self.keys[probe] - .as_ref() - .map(|existing| existing == &key) - .unwrap_or(false); - if matches { - break; - } - probe = (probe + 1) & mask; - drift += 1; - debug_assert!(drift < DRIFT_LIMIT, "drift limit exceeded"); - } + let (probe, drift) = self.find_probe_or_empty(&key); if self.states[probe] == 0 { self.keys[probe] = Some(key); self.values[probe] = adjust_amount; @@ -103,33 +89,12 @@ impl ReversePurgeItemHashMap { } /// Adds `adjust_amount` to the value for a borrowed `key`, inserting if absent. - /// - /// Behaves like [`adjust_or_put_value`](Self::adjust_or_put_value) but takes - /// the key by reference and only allocates an owned key (via - /// `Q: ToOwned`) on the insert path, so incrementing an - /// already-present key is allocation free. The `Borrow` contract guarantees - /// the borrowed and owned forms hash and compare identically, so lookups land - /// on the same slot as the owned API. pub fn adjust_or_put_value_ref(&mut self, key: &Q, adjust_amount: u64) where T: Borrow, Q: Eq + Hash + ToOwned + ?Sized, { - 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 { - let matches = self.keys[probe] - .as_ref() - .map(|existing| existing.borrow() == key) - .unwrap_or(false); - if matches { - break; - } - probe = (probe + 1) & mask; - drift += 1; - debug_assert!(drift < DRIFT_LIMIT, "drift limit exceeded"); - } + let (probe, drift) = self.find_probe_or_empty(key); if self.states[probe] == 0 { self.keys[probe] = Some(key.to_owned()); self.values[probe] = adjust_amount; @@ -274,13 +239,14 @@ impl ReversePurgeItemHashMap { self.states[probe] > 0 } - fn hash_probe(&self, key: &Q) -> usize + fn find_probe_or_empty(&self, key: &Q) -> (usize, usize) where T: Borrow, Q: Eq + Hash + ?Sized, { 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 { let matches = self.keys[probe] .as_ref() @@ -290,8 +256,10 @@ impl ReversePurgeItemHashMap { break; } probe = (probe + 1) & mask; + drift += 1; + debug_assert!(drift < DRIFT_LIMIT, "drift limit exceeded"); } - probe + (probe, drift) } fn hash_delete(&mut self, mut delete_probe: usize) { diff --git a/datasketches/tests/frequencies_update_test.rs b/datasketches/tests/frequencies_update_test.rs index f15f219..87c702a 100644 --- a/datasketches/tests/frequencies_update_test.rs +++ b/datasketches/tests/frequencies_update_test.rs @@ -119,6 +119,34 @@ fn test_items_one_item() { assert_eq!(sketch.upper_bound(&item), 1); } +#[test] +fn test_items_borrowed_key_updates_and_queries() { + let mut sketch = FrequentItemsSketch::::new(16); + + sketch.update_ref("alpha"); + sketch.update_ref("alpha"); + sketch.update_with_count_ref("beta", 3); + sketch.update_with_count_ref("ignored", 0); + for item in [ + "gamma", "delta", "epsilon", "zeta", "eta", "theta", "iota", "kappa", "lambda", "mu", + ] { + sketch.update_ref(item); + } + + assert!(!sketch.is_empty()); + assert_eq!(sketch.total_weight(), 15); + assert_eq!(sketch.num_active_items(), 12); + assert_eq!(sketch.estimate("alpha"), 2); + assert_eq!(sketch.lower_bound("alpha"), 2); + assert_eq!(sketch.upper_bound("alpha"), 2); + assert_eq!(sketch.estimate("beta"), 3); + assert_eq!(sketch.estimate("ignored"), 0); + assert_eq!(sketch.estimate("missing"), 0); + + let owned = "alpha".to_string(); + assert_eq!(sketch.estimate(&owned), 2); +} + #[test] fn test_longs_several_items_no_resize_no_purge() { let mut sketch: FrequentItemsSketch = FrequentItemsSketch::new(8);