fix(kmv): read and merge by shared reference, report memory, skip the duplicate scan past the k-th hash

CHANGELOG.md

@@ -92,6 +92,14 @@ signals a backwards-compatible change.
 
 ### Fixed
 
+- **`KMV` reads, merges and reports memory without a mutable borrow or a
+  copy.** `estimate` takes `&self` and `merge` takes the other sketch as
+  `&KMV`, so a caller holding a shared state no longer clones it to read an
+  estimate or to merge it in; existing `&mut` call sites still compile.
+  `approx_memory_bytes` reports the struct plus the hash slots the heap has
+  allocated. A full sketch turns away a hash at or above its largest
+  retained one before scanning the `k` retained hashes for a duplicate, so
+  steady-state inserts no longer cost `O(k)`; the retained set is unchanged.
 - `KLLDynamic` panicked on every quantile query once a NaN had entered the
   stream: the CDF was sorted with `partial_cmp(..).unwrap()`. It sorts with
   `total_cmp`, as the three sibling sorts beside it already did.

docs/api/api_kmv.md

@@ -29,13 +29,18 @@ fn insert_by_hash(&mut self, hash_value: u64)
 ## Query
 
 ```rust
-fn estimate(&mut self) -> f64
+fn estimate(&self) -> f64
+fn approx_memory_bytes(&self) -> usize
 ```
 
+`approx_memory_bytes` is the struct plus the hash slots the heap has
+allocated: up to 1024 at construction, growing toward `k` as distinct hashes
+arrive.
+
 ## Merge
 
 ```rust
-fn merge(&mut self, other: &mut KMV<H>)
+fn merge(&mut self, other: &KMV<H>)
 ```
 
 ## Serialization

docs/tests.md

@@ -707,6 +707,8 @@ Wire tests: [`src/sketches/kmv/wire.rs`](../src/sketches/kmv/wire.rs)
 
 | test_name | test_description | what_is_tested |
 | --- | --- | --- |
+| `estimate_and_merge_borrow_inputs_shared` | Reading an estimate and merging another sketch need no mutable binding or copy. | Merges an immutable `k = 64` sketch over `400..1000` into a copy of one over `0..600`, verifies the merged estimate equals the build over `0..1000`, and that the left input's estimate is unchanged. |
+| `memory_report_covers_the_retained_hashes` | The memory report covers the retained hashes and does not move on a rejected insert. | For `k = 3000`, verifies the empty and the filled (20,000 keys) sketches each report at least the struct plus `8` bytes per retained hash, that the filled one reports more, and that a duplicate key and a hash of `u64::MAX` leave the report and the estimate unchanged. |
 | `assert_serialization_round_trip` | Assert serialization round trip. | Serialization/deserialization preserves component state and behavior after round trip. |
 | `kmv_round_trip_serialization` | The envelope frames a sketch under kind_id `0x0e 0x00`, and the bytes are stable across a round trip. | For `k = 64` filled by 5,000 distinct keys, verifies the bytes open with the ASAPv1 magic, a `kind_id_len` of `2`, and `0x0e 0x00`, that the metadata carries `k` `64` and the canonical seed index, that the 64 emitted hashes are strictly ascending, and that the decode matches on `k`, capacity, retained hashes, heap root, and `estimate`, re-serializing identically. |
 | `kmv_emitted_order_is_independent_of_insertion_order` | The emitted order follows the hash value, not the arrival order. | Fills two `k = 32` sketches with the same 2,000 keys forward and backward, verifies their heap arrays differ, and that both serialize to the same bytes, which a decode re-encodes exactly. |

src/common/structures/heap.rs

@@ -93,6 +93,13 @@ impl<T, O: CommonHeapOrder<T>> CommonHeap<T, O> {
         self.size
     }
 
+    /// Element slots the heap has allocated: at least [`Self::len`], and
+    /// grown on demand toward [`Self::capacity`]. For memory accounting.
+    #[inline]
+    pub fn allocated_slots(&self) -> usize {
+        self.data.capacity()
+    }
+
     /// Returns true if the heap is at full capacity.
     #[inline]
     pub fn is_full(&self) -> bool {

src/sketches/kmv.rs

@@ -46,6 +46,11 @@ impl<H: SketchHasher> KMV<H> {
     }
 
     pub fn insert_by_hash(&mut self, hash_value: u64) {
+        // A full sketch keeps only hashes below its largest retained one, so
+        // most inserts are turned away here without scanning the k hashes.
+        if self.k_vals.is_full() && self.k_vals.peek().is_some_and(|max| hash_value >= *max) {
+            return;
+        }
         if self.k_vals.iter().any(|value| *value == hash_value) {
             return;
         }
@@ -54,7 +59,7 @@ impl<H: SketchHasher> KMV<H> {
 
     /// `(k - 1) / U(k)`, where `U(k)` is the k-th smallest hash's position in
     /// the 64-bit range, mapped into (0, 1].
-    pub fn estimate(&mut self) -> f64 {
+    pub fn estimate(&self) -> f64 {
         if self.k_vals.len() < self.k {
             return self.k_vals.len() as f64;
         }
@@ -67,7 +72,7 @@ impl<H: SketchHasher> KMV<H> {
         (self.k - 1) as f64 / mapped
     }
 
-    pub fn merge(&mut self, other: &mut KMV<H>) {
+    pub fn merge(&mut self, other: &KMV<H>) {
         assert_eq!(
             self.k, other.k,
             "Two KMV sketch have different k size, not mergeable"
@@ -76,6 +81,13 @@ impl<H: SketchHasher> KMV<H> {
             self.insert_by_hash(value);
         }
     }
+
+    /// Approximate in-memory footprint in bytes: the struct plus the hash
+    /// slots its heap has allocated. The heap reserves up to 1024 slots at
+    /// construction and grows toward `k` as distinct hashes arrive.
+    pub fn approx_memory_bytes(&self) -> usize {
+        std::mem::size_of::<Self>().saturating_add(self.k_vals.allocated_slots().saturating_mul(8))
+    }
 }
 
 #[cfg(test)]
@@ -87,6 +99,51 @@ mod tests {
     const ERROR_TOLERANCE: f64 = 0.02;
     const SERDE_SAMPLE: usize = 100_000;
 
+    fn build(k: usize, items: std::ops::Range<u64>) -> KMV {
+        let mut sketch = KMV::new(k);
+        for item in items {
+            sketch.insert(&DataInput::U64(item));
+        }
+        sketch
+    }
+
+    /// Reading an estimate and merging in another sketch borrow their inputs
+    /// shared, so neither needs a mutable binding or a copy, and the merged
+    /// estimate equals the union build's.
+    #[test]
+    fn estimate_and_merge_borrow_inputs_shared() {
+        let left = build(64, 0..600);
+        let right = build(64, 400..1_000);
+        let union = build(64, 0..1_000);
+        let mut merged = left.clone();
+        merged.merge(&right);
+        assert_eq!(merged.estimate(), union.estimate());
+        assert_eq!(left.estimate(), build(64, 0..600).estimate());
+    }
+
+    /// The memory report covers the hashes the heap holds, grows as it fills,
+    /// and is the same after a duplicate or a hash past the retained k.
+    #[test]
+    fn memory_report_covers_the_retained_hashes() {
+        let k = 3_000;
+        let empty = KMV::new(k);
+        let full = build(k, 0..20_000);
+        assert_eq!(full.k_vals.len(), k);
+        for sketch in [&empty, &full] {
+            assert!(
+                sketch.approx_memory_bytes()
+                    >= std::mem::size_of::<KMV>() + sketch.k_vals.len() * 8
+            );
+        }
+        assert!(full.approx_memory_bytes() > empty.approx_memory_bytes());
+        let mut again = full.clone();
+        let before = again.approx_memory_bytes();
+        again.insert(&DataInput::U64(7));
+        again.insert_by_hash(u64::MAX);
+        assert_eq!(again.approx_memory_bytes(), before);
+        assert_eq!(again.estimate(), full.estimate());
+    }
+
     #[test]
     fn assert_serialization_round_trip() {
         let mut sketch: KMV = KMV::default();
@@ -103,7 +160,7 @@ mod tests {
             "KMV serialization output should not be empty"
         );
 
-        let mut decoded: KMV = KMV::deserialize_from_bytes(&encoded)
+        let decoded: KMV = KMV::deserialize_from_bytes(&encoded)
             .unwrap_or_else(|err| panic!("KMV deserialize_from_bytes failed: {err}"));
 
         let reencoded = decoded

src/sketches/kmv/wire.rs

@@ -242,7 +242,7 @@ mod tests {
 
     #[test]
     fn kmv_round_trip_serialization() {
-        let mut sketch = populated(64, 5_000);
+        let sketch = populated(64, 5_000);
         assert_eq!(sketch.k_vals.len(), 64, "the bound was never reached");
 
         let encoded = sketch.serialize_to_bytes().expect("serialize KMV");
@@ -260,7 +260,7 @@ mod tests {
             assert!(pair[0] < pair[1], "the emitted hashes are not ascending");
         }
 
-        let mut decoded = KMV::<DefaultXxHasher>::deserialize_from_bytes(&encoded).expect("decode");
+        let decoded = KMV::<DefaultXxHasher>::deserialize_from_bytes(&encoded).expect("decode");
         assert_eq!(decoded.k, sketch.k);
         assert_eq!(decoded.k_vals.capacity(), sketch.k_vals.capacity());
         assert_eq!(decoded.wire_hashes(), sketch.wire_hashes());
@@ -317,7 +317,7 @@ mod tests {
         assert_eq!(metadata_of(&bytes).k, 16);
         assert!(payload_of(&bytes).hashes.is_empty());
 
-        let mut decoded = KMV::<DefaultXxHasher>::deserialize_from_bytes(&bytes).expect("decode");
+        let decoded = KMV::<DefaultXxHasher>::deserialize_from_bytes(&bytes).expect("decode");
         assert_eq!(decoded.k, 16);
         assert_eq!(decoded.k_vals.len(), 0);
         assert_eq!(decoded.k_vals.capacity(), 16);

tests/e2e/experimental.rs

@@ -122,7 +122,7 @@ fn kmv_estimates_stay_inside_their_relative_standard_error_band_over_independent
             for regime in KMV_REGIMES {
                 let n = regime.n(k);
                 namespace = namespace.wrapping_add(NAMESPACE_STRIDE);
-                let mut sketch = kmv_trial(k, n, seed_idx, namespace);
+                let sketch = kmv_trial(k, n, seed_idx, namespace);
                 let estimate = sketch.estimate();
                 let outcome = spec.check(estimate, n);
                 tally.record(outcome.is_ok(), || {
@@ -175,7 +175,7 @@ fn kmv_is_exact_below_k_and_estimates_at_k() {
     );
 
     for n in [1usize, 2, K / 2, K - 2, K - 1] {
-        let mut sketch: KMV = kmv_trial(K, n, 0, NAMESPACE_STRIDE * 900 + n as u64 * 4096);
+        let sketch: KMV = kmv_trial(K, n, 0, NAMESPACE_STRIDE * 900 + n as u64 * 4096);
         assert_eq!(
             sketch.estimate(),
             n as f64,
@@ -188,7 +188,7 @@ fn kmv_is_exact_below_k_and_estimates_at_k() {
     // to k, so the exact-regime branch must not claim it.
     let mut estimates = Vec::new();
     for (i, &seed_idx) in KMV_HASH_SEEDS.iter().enumerate() {
-        let mut sketch = kmv_trial(K, K, seed_idx, NAMESPACE_STRIDE * (1_000 + i as u64));
+        let sketch = kmv_trial(K, K, seed_idx, NAMESPACE_STRIDE * (1_000 + i as u64));
         let est = sketch.estimate();
         assert!(
             spec.check(est, K).is_ok(),
@@ -255,8 +255,7 @@ fn kmv_duplicates_are_inert_and_a_shard_merge_reproduces_the_single_pass_exactly
     );
 
     let mut merged = a.clone();
-    let mut rhs = b.clone();
-    merged.merge(&mut rhs);
+    merged.merge(&b);
     assert_eq!(
         merged.estimate(),
         single,

tests/e2e/wire.rs

@@ -461,7 +461,7 @@ fn the_experimental_envelopes_round_trip_and_name_their_kinds() {
     }
     let kmv_bytes = kmv.serialize_to_bytes().expect("encode KMV");
     assert_kind(&kmv_bytes, 0x0e, 0x00, "KMV");
-    let mut decoded_kmv =
+    let decoded_kmv =
         KMV::<DefaultXxHasher>::deserialize_from_bytes(&kmv_bytes).expect("decode KMV");
     assert_eq!(decoded_kmv.estimate(), kmv.estimate(), "KMV estimate");
 

tests/pbt/kmv.rs

@@ -143,8 +143,8 @@ fn from_hashes(k: usize, stream: &[u64]) -> Kmv {
 
 fn merged(k: usize, left: &[u64], right: &[u64]) -> Kmv {
     let mut a = from_hashes(k, left);
-    let mut b = from_hashes(k, right);
-    a.merge(&mut b);
+    let b = from_hashes(k, right);
+    a.merge(&b);
     a
 }
 
@@ -241,7 +241,7 @@ proptest! {
     fn a_full_sketch_estimates_k_minus_one_over_its_mapped_k_th_smallest_hash(
         (k, stream, raise) in full_stream(),
     ) {
-        let mut sketch = from_hashes(k, &stream);
+        let sketch = from_hashes(k, &stream);
         prop_assert_eq!(sketch.k_vals.len(), k);
         let hashes = retained(&sketch);
         let kth = *hashes.last().expect("a full sketch retains k hashes");
@@ -333,7 +333,7 @@ proptest! {
     fn a_sketch_under_its_bound_estimates_its_distinct_count_exactly(
         (k, distinct, stream) in under_the_bound(48),
     ) {
-        let mut sketch = from_hashes(k, &stream);
+        let sketch = from_hashes(k, &stream);
 
         prop_assert_eq!(sketch.k_vals.len(), distinct.len());
         prop_assert_eq!(
@@ -353,8 +353,8 @@ proptest! {
     fn a_full_sketch_estimates_above_one_holding_a_larger_k_th_hash(
         (k, lower, higher) in two_full_streams(),
     ) {
-        let mut lower_sketch = from_hashes(k, &lower);
-        let mut higher_sketch = from_hashes(k, &higher);
+        let lower_sketch = from_hashes(k, &lower);
+        let higher_sketch = from_hashes(k, &higher);
         prop_assert_eq!(lower_sketch.k_vals.len(), k);
         prop_assert_eq!(higher_sketch.k_vals.len(), k);
 
@@ -379,9 +379,9 @@ proptest! {
         right in hash_stream(200),
     ) {
         let mut a = from_hashes(k, &left);
-        let mut b = from_hashes(k, &right);
+        let b = from_hashes(k, &right);
         let read_before = retained(&b);
-        a.merge(&mut b);
+        a.merge(&b);
 
         let both: Vec<u64> = left.iter().chain(right.iter()).copied().collect();
         prop_assert_eq!(
@@ -411,12 +411,12 @@ proptest! {
         );
 
         let mut left_first = merged(k, &first, &second);
-        let mut third_sketch = from_hashes(k, &third);
-        left_first.merge(&mut third_sketch);
+        let third_sketch = from_hashes(k, &third);
+        left_first.merge(&third_sketch);
 
         let mut right_first = from_hashes(k, &first);
-        let mut tail = merged(k, &second, &third);
-        right_first.merge(&mut tail);
+        let tail = merged(k, &second, &third);
+        right_first.merge(&tail);
 
         prop_assert_eq!(
             retained(&left_first),
@@ -433,8 +433,8 @@ proptest! {
     ) {
         let mut sketch = from_hashes(k, &stream);
         let before = retained(&sketch);
-        let mut copy = sketch.clone();
-        sketch.merge(&mut copy);
+        let copy = sketch.clone();
+        sketch.merge(&copy);
 
         prop_assert_eq!(
             retained(&sketch),