diff --git a/CHANGELOG.md b/CHANGELOG.md index 80d85f98..12e00ce4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -66,6 +66,9 @@ signals a backwards-compatible change. worker's seed and the aggregator's from one seed. The unseeded constructors draw from the thread generator. The seed is not serialized, and the ASAPv1 payload is unchanged. +- **An ASAPv1 golden fixture for CSHeap** (`0x0a 0x00`), + `csheap_i64_regular_2x4_strkeys`: the Count Sketch 2x4 signed matrix with a + string-keyed heap at `k = 5`, pinned in `tests/asapv1_golden.rs`. ### Changed @@ -88,6 +91,19 @@ signals a backwards-compatible change. as an empty window like any other, and that window takes a place in `max_windows`. +### Removed + +- **The portable Count-Sketch-with-heap type.** `CountSketchWithHeap` and + `CsHeapItem` (crate root) and the module + `message_pack_format::portable::countsketch_topk` with `WireHeapItem`, + `SketchlibCSHeap`, `new_sketchlib_cs_heap`, + `sketchlib_cs_heap_from_matrix_and_heap`, `matrix_from_sketchlib_cs_heap`, + `heap_to_wire`, `sketchlib_cs_heap_update`, `sketchlib_cs_heap_query`, + `CountSketchInnerWire` and `CountSketchWithHeapWire`. Its MessagePack + `{sketch, topk_heap, heap_size}` encoding is gone with it; `CSHeap` encodes + and decodes ASAPv1 only, through `serialize_to_bytes` / + `deserialize_from_bytes`. + ### Fixed - `KLLDynamic` panicked on every quantile query once a NaN had entered the diff --git a/asapv1_golden b/asapv1_golden index 8287a4ae..5760846e 160000 --- a/asapv1_golden +++ b/asapv1_golden @@ -1 +1 @@ -Subproject commit 8287a4ae04e6cfbfb3ed33dafbf1dac77d8f7fc1 +Subproject commit 5760846eb1485defbd93aa5a7e89f1f9ed2dd104 diff --git a/docs/asapv1_wire_format.md b/docs/asapv1_wire_format.md index fd14e70a..3e538119 100644 --- a/docs/asapv1_wire_format.md +++ b/docs/asapv1_wire_format.md @@ -1341,14 +1341,14 @@ Good direction (more compact, higher fidelity, less Rust-internal duplication), 2. **Golden byte-vector fixtures** in one shared repo, [`sketchlib-golden-bytes`](https://github.com/ProjectASAP/sketchlib-golden-bytes), which each implementation mounts at `asapv1_golden/`; both languages decode and re-encode them byte-identically. These replace the `portable`-as-oracle round-trip test. 3. **This registry**, mirrored, never independently allocated. -Fixtures exist for six `kind_id`s — HLL's three estimators, Count-Min, Count Sketch and compact KLL — and `sketchlib-go` mirrors those. Every other kind here has a payload and **no fixture**, so this document is its only contract; `asapv1_golden/README.md` lists the gap. `portable` carries its own Go goldens. +`asapv1_golden/README.md`'s Coverage section lists which `kind_id`s have fixtures; for every other kind, this document is the only contract. `portable` carries its own Go goldens. **Hash profile on the Go side.** Rust derives the hash spec from a generic `HashProfile` bound on the hasher type; Go has no generic hasher type, so there is nothing to derive from. On the Go side the profile is simply **written into** the metadata on encode and **read from** it on decode. Go MUST validate the profile it reads (same fail-closed intent as Rust): a sketch is only mergeable/queryable if its `hash_profile_id` + seeds match the profile Go is prepared to reproduce. -Sequencing: (2) covers six `kind_id`s, and the one `native bytes == portable bytes` test is HLL-only, since no other `portable` type emits ASAPv1 bytes. +Sequencing: the one `native bytes == portable bytes` test is HLL-only, since no other `portable` type emits ASAPv1 bytes. Retire `portable` once the fixtures cover the rest. --- diff --git a/docs/message_pack_format.md b/docs/message_pack_format.md index 12a2b91c..8dae3de7 100644 --- a/docs/message_pack_format.md +++ b/docs/message_pack_format.md @@ -169,10 +169,10 @@ decode → re-encode them byte-identically. The `kind_id` registry is mirrored verbatim with Go's `wire/asapmsgpack/magic_ids.go`, never independently allocated. -That proof covers the six `kind_id`s a fixture exists for — HLL's three -estimators, Count-Min, Count Sketch and compact KLL. Every other implemented -kind has **no ASAPv1 golden and therefore no ASAPv1 drift guard**; -[its README](https://github.com/ProjectASAP/sketchlib-golden-bytes#coverage) lists what is covered. +That proof covers only the `kind_id`s with a fixture, which +[its README's Coverage section](https://github.com/ProjectASAP/sketchlib-golden-bytes#coverage) +lists. Every other implemented kind has **no ASAPv1 golden and therefore no +ASAPv1 drift guard**. The older `portable` path is guarded separately, by the `sketchlib-go` goldens of its own listed above. diff --git a/docs/tests.md b/docs/tests.md index ccb1571c..086738c4 100644 --- a/docs/tests.md +++ b/docs/tests.md @@ -1279,21 +1279,6 @@ Test file: [`src/message_pack_format/portable/countsketch.rs`](../src/message_pa | `test_compute_delta_then_apply_matches_current` | A delta between two non-empty snapshots reconstructs the later one. | For a `3x64` base fed 40 updates of `x{i % 8}` and a current fed 30 more, verifies the delta applied to the base reproduces the current matrix. | | `test_compute_delta_matches_go_golden_bytes` | The packed delta bytes and per-row L2 are pinned by a golden. | For a `3x512` sketch fed the 25 `k-a`..`k-e` updates, verifies the delta against an empty sketch at threshold `1.0` equals a 93-byte inline golden carrying the packed cell arrays and a per-row L2 of `125.0`. | -### Portable CountSketchWithHeap - -Test file: [`src/message_pack_format/portable/countsketch_topk.rs`](../src/message_pack_format/portable/countsketch_topk.rs) - -| test_name | test_description | what_is_tested | -| --- | --- | --- | -| `test_creation` | A fresh composite has the requested matrix and an empty heap. | For `CountSketchWithHeap::new(4, 1000, 20)`, verifies `rows` `4`, `cols` `1000`, `heap_size` `20`, a `sketch_matrix()` of 4 rows of 1,000, and no heap items. | -| `test_query_empty` | An empty composite answers zero. | Verifies a `2x10` composite with a 5-slot heap reports `estimate("anything")` of `0.0`. | -| `test_merge` | Merge sums the matrices and adopts the peer's heap bound. | Merges a `2x10` composite holding `5.0` at `(0,0)`, `15.0` at `(1,1)`, heap `{key3: 75.0, key1: 80.0}` and `heap_size` `3` into one holding `10.0` at `(0,0)`, `20.0` at `(1,1)`, heap `{key1: 100.0, key2: 50.0}` and `heap_size` `5`, and verifies the merged cells read `15.0` and `35.0`, `heap_size` becomes `3`, and the heap holds at most 3 items. | -| `test_merge_dimension_mismatch` | Mismatched row counts are refused. | Verifies merging a `3x10` composite into a `2x10` one fails. | -| `test_msgpack_round_trip` | The round trip carries the matrix, the heap bound, and the heap itself. | Round-trips a `4x128` composite with a 3-slot heap fed `"hot"` at `100.0` and `"cold"` at `1.0`, and verifies the decode reports `rows` `4`, `cols` `128`, `heap_size` `3`, a non-empty heap holding `"hot"` at or above `100.0`, and estimates at or above `100.0` and `1.0`. | -| `test_aggregate_topk` | `aggregate_topk` builds a decodable composite honouring its heap bound. | For `aggregate_topk(4, 100, 2, ["a", "b", "a", "c"], [1.0, 2.0, 3.0, 0.5])`, verifies the decode reports `heap_size` `2` and at most 2 heap items; the heap contents themselves are not checked. | -| `test_aggregate_topk_empty` | An empty stream produces nothing. | Verifies `aggregate_topk(4, 100, 10, &[], &[])` returns `None`. | -| `test_median_estimator_differs_from_cms_style_min` | The median-over-rows estimator tracks the true count rather than inflating. | For a `5x64` composite with a 10-slot heap fed `"heavy"` 50 times at weight `1.0` and 20 keys `light-0`..`light-19` once each, verifies `estimate("heavy")` falls in `40.0..=60.0`. | - ### Portable DdSketch Test file: [`src/message_pack_format/portable/ddsketch.rs`](../src/message_pack_format/portable/ddsketch.rs) @@ -1957,7 +1942,6 @@ Test file: [`tests/e2e/composition.rs`](../tests/e2e/composition.rs) | test_name | test_description | what_is_tested | | --- | --- | --- | | `portable_count_min_with_heap_satisfies_the_count_min_bound_through_merge_and_wire` | The portable Count-Min-plus-heap holds Count-Min's bound and an agreeing heap through a merge and a MessagePack round trip. | Feeds `40_000` Zipf(1.1) keys over a `2_048` domain (seed `0xC090_5101`) as `"k{key}"` strings into `CountMinSketchWithHeap::new(3, 4_096, 32)`, plus even/odd halves into two more. Runs `CountMinSpec::new(3, 4_096).assert_contract` on the single sketch, asserts every `topk_heap_items()` entry's `value` equals `estimate(&item.key)` with zero violations, re-runs the contract on `CountMinSketchWithHeap::merge_refs(&[&left, &right])` against the full truth, and after `to_msgpack`/`from_msgpack` asserts the decoded sketch's `estimate` is exactly equal to the original's for every observed key and still satisfies the contract. | -| `portable_count_sketch_with_heap_satisfies_the_l2_bound_through_merge_and_wire` | The portable Count-Sketch-plus-heap holds the L2 bound and an agreeing heap through a merge and a MessagePack round trip. | Feeds the same `40_000`-key Zipf(1.1) stream (domain `2_048`, seed `0xC090_5101`) as `"k{key}"` strings into `CountSketchWithHeap::new(5, 4_096, 32)`, plus even/odd halves into two more, and runs `CountSketchSpec::new(5, 4_096).assert_contract` (simultaneous L2 band at `delta = 1e-3` with zero violations, marginal `sqrt(3/4096)*\|\|f_-i\|\|_2` rate pinned at `P[Bin(5, 1/3) >= 3] = 51/243`) on the single sketch, on `merge_refs(&[&left, &right])` and on the `from_msgpack` decode; every `topk_heap_items()` entry's `value` must equal `estimate(&item.key)`. Unlike the Count-Min twin, the wire round trip is only re-checked against the band - no per-key equality across the encode/decode is asserted. | | `portable_kll_sketch_satisfies_the_rank_error_characterization_through_merge_and_wire` | The portable KLL facade holds the rank-error characterization single-pass and after a two-shard merge, and its wire round trip is bit-for-bit. | Over `12` compaction seeds `0x5EED_0400 + t * 0x9E37_79B9_7F4A_7C15`, builds `KllSketch::with_seed(200, seed)` plus even/odd shards at `seed ^ 0xAAAA` and `seed ^ 0x5555`, feeds all of them `40_000` uniform values over `1_000_000` (seed `0xC090_5101`), and merges the shards with `merge`. Each seed records two trials - the single-pass sketch and the merged pair - of the worst normalized rank error over the grid `[0.1, 0.25, 0.5, 0.75, 0.9]` against `KllRankSpec::datasketches(200)`'s `eps = 2.446/200^0.9433 = 0.0165`, and the `24` outcomes are accepted by `assert_independent_binomial` at `p = 0.01`. Per seed it also asserts `to_msgpack`/`from_msgpack` preserves `k() == 200`, `count()`, and every grid quantile exactly, and that merging a `k = 400` sketch into the single-pass one errs. | ### Composition: Documented Input Matrix @@ -2126,6 +2110,7 @@ Test file: [`tests/asapv1_golden.rs`](../tests/asapv1_golden.rs) | `cs_i32_regular_2x4_matches_golden` | The i32 counter type reaches the metadata, and an i32 golden is not an i64 sketch. | Verifies `Count::, RegularPath>::from_storage` over `[[0, 127, 128, 65536], [-1, -33, -32768, -2147483648]]` cast to `i32` serializes to `asapv1_golden/cs_i32_regular_2x4.hex`, that decoding it recovers the eight cells and re-serializes identically, that the i32 and i64 regular fixtures have equal length yet differ, and that the i32 fixture fails to decode as `Count, RegularPath>`. | | `cs_i64_fast_2x4_matches_golden` | The fast-path mode reaches the bytes, so the same matrix encodes differently per path. | Verifies `Count::, FastPath>::from_storage` over `[[0, 127, 128, 65536], [-1, -33, -32768, -2147483648]]` serializes to `asapv1_golden/cs_i64_fast_2x4.hex`, that decoding it recovers the eight cells and re-serializes identically, and that the fixture differs from `cs_i64_regular_2x4.hex` built from the same matrix. | | `cs_i64_regular_2x4_matches_golden` | A known signed Count-Sketch matrix serializes to its checked-in golden bytes. | Verifies `Count::, RegularPath>::from_storage` over `[[0, 127, 128, 65536], [-1, -33, -32768, -2147483648]]` — sweeping the msgpack integer widths in both directions — serializes to `asapv1_golden/cs_i64_regular_2x4.hex`, and that decoding the fixture reports `rows()` `2`, `cols()` `4`, the eight cells, and a byte-identical re-serialization. | +| `csheap_i64_regular_2x4_strkeys_matches_golden` | A known CSHeap matrix and heap serialize to their checked-in golden bytes. | Builds `CSHeap::, RegularPath>::from_storage` over `[[0, 127, 128, 65536], [-1, -33, -32768, -2147483648]]` at `k` `5` and seats the heap directly through `heap_mut().update` with `Str` keys `gamma` `-33`, `delta` `127`, `alpha` `4294967296`, `beta` `127`; verifies the bytes equal `asapv1_golden/csheap_i64_regular_2x4_strkeys.hex`, that decoding recovers the eight cells, `rows()` `2`, `cols()` `4`, capacity `5`, four entries each at its count, and a byte-identical re-serialization, and that the fixture fails to decode as `Count, RegularPath>`. | | `hll_classic_p12_matches_golden` | A known P12 Classic register pattern serializes to its golden bytes on both the portable and native paths. | Builds `HllSketch::from_raw(HllVariant::Regular, 12, regs, 0.0, 0.0, 0.0)` where `regs` is 4,096 zeros with `[0] = 1`, `[1] = 7`, `[100] = 42` and `[4095] = 3`, verifies `to_msgpack()` equals `asapv1_golden/hll_classic_p12.hex`, that `HyperLogLogP12::::deserialize_from_bytes` on the fixture recovers those registers and re-serializes identically, and that `HllSketch::from_msgpack` reports the same registers, `precision` `12` and `HllVariant::Regular`. | | `hll_ertl_mle_p12_matches_golden` | The Datafusion variant tag reaches the bytes for the same P12 register pattern. | Builds `HllSketch::from_raw(HllVariant::Datafusion, 12, regs, 0.0, 0.0, 0.0)` over the shared P12 pattern (`[0] = 1`, `[1] = 7`, `[100] = 42`, `[4095] = 3` in 4,096 registers), verifies `to_msgpack()` equals `asapv1_golden/hll_ertl_mle_p12.hex`, that `HyperLogLogP12::::deserialize_from_bytes` recovers the registers and re-serializes identically, and that `HllSketch::from_msgpack` reports `HllVariant::Datafusion`. | | `hll_hip_p12_matches_golden` | The HIP variant's three running scalars travel on the wire beside its registers. | Builds `HllSketch::from_raw(HllVariant::Hip, 12, regs, 1.5, 2.5, 3.0)` over the shared P12 pattern, verifies `to_msgpack()` equals `asapv1_golden/hll_hip_p12.hex`, that `HyperLogLogHIPP12::deserialize_from_bytes` re-serializes the fixture identically, and that `HllSketch::from_msgpack` reports the registers, `HllVariant::Hip`, `hip_kxq0` `1.5`, `hip_kxq1` `2.5` and `hip_est` `3.0`. | diff --git a/src/lib.rs b/src/lib.rs index b062b8b9..92003b6d 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -46,7 +46,6 @@ pub use message_pack_format::portable::countminsketch_topk::{CmsHeapItem, CountM pub use message_pack_format::portable::countsketch::{ COUNT_SKETCH_TOPK_CAPACITY, CountSketch, CountSketchDelta, }; -pub use message_pack_format::portable::countsketch_topk::{CountSketchWithHeap, CsHeapItem}; pub use message_pack_format::portable::ddsketch::{DDSKETCH_GROW_CHUNK, DdSketch, DdSketchDelta}; pub use message_pack_format::portable::delta_set_aggregator::DeltaResult; pub use message_pack_format::portable::hll::{HllSketch, HllSketchDelta, HllVariant}; diff --git a/src/message_pack_format/portable/countsketch_topk.rs b/src/message_pack_format/portable/countsketch_topk.rs deleted file mode 100644 index 82ec8958..00000000 --- a/src/message_pack_format/portable/countsketch_topk.rs +++ /dev/null @@ -1,471 +0,0 @@ -//! Wire-format-aligned Count Sketch + top-k heap composite. -//! -//! No `sketches::*` equivalent — this combines `sketches::CSHeap` with -//! the wire-format `CountSketch` shape and exposes it as a single -//! Go-interop type. Mirrors [`crate::message_pack_format::portable::countminsketch_topk::CountMinSketchWithHeap`] -//! structurally, but backed by `CSHeap` (median estimator, signed rows) -//! instead of `CMSHeap` (min-over-rows estimator) — the two are -//! different sketch algorithms that happen to produce a same-shaped -//! `rows x cols` matrix, not interchangeable data. - -use serde::{Deserialize, Serialize}; - -use crate::message_pack_format::{Error as MsgPackError, MessagePackCodec}; -use crate::sketches::countsketch_topk::CSHeap; -use crate::{DataInput, RegularPath, Vector2D}; - -/// Wire-format heap item (key, value) used by the dispatch helpers below. -pub struct WireHeapItem { - pub key: String, - pub value: f64, -} - -/// Concrete Count-Sketch-with-Heap type backing the wire-format -/// `CountSketchWithHeap`. -pub type SketchlibCSHeap = CSHeap, RegularPath>; - -/// Creates a fresh CSHeap with the given dimensions and heap capacity. -pub fn new_sketchlib_cs_heap(row_num: usize, col_num: usize, heap_size: usize) -> SketchlibCSHeap { - CSHeap::new(row_num, col_num, heap_size) -} - -/// Builds a CSHeap from an existing sketch matrix and optional heap -/// items. Used when deserializing wire-format state. -pub fn sketchlib_cs_heap_from_matrix_and_heap( - row_num: usize, - col_num: usize, - heap_size: usize, - sketch: &[Vec], - topk_heap: &[WireHeapItem], -) -> SketchlibCSHeap { - let matrix = Vector2D::from_fn(row_num, col_num, |r, c| { - sketch - .get(r) - .and_then(|row| row.get(c)) - .copied() - .unwrap_or(0.0) - .round() as i64 - }); - let mut cs_heap = CSHeap::from_storage(matrix, heap_size); - - for item in topk_heap { - let count = item.value.round() as i64; - if count > 0 { - let input = DataInput::Str(&item.key); - cs_heap.heap_mut().update(&input, count); - } - } - - cs_heap -} - -/// Converts a CSHeap's storage into a `Vec>` matrix. -pub fn matrix_from_sketchlib_cs_heap(cs_heap: &SketchlibCSHeap) -> Vec> { - let storage = cs_heap.cs().as_storage(); - let rows = storage.rows(); - let cols = storage.cols(); - let mut sketch = vec![vec![0.0; cols]; rows]; - - for (r, row) in sketch.iter_mut().enumerate().take(rows) { - for (c, cell) in row.iter_mut().enumerate().take(cols) { - if let Some(v) = storage.get(r, c) { - *cell = *v as f64; - } - } - } - - sketch -} - -/// Converts sketchlib HHHeap items to wire-format (key, value) pairs. -pub fn heap_to_wire(cs_heap: &SketchlibCSHeap) -> Vec { - cs_heap - .heap() - .heap() - .iter() - .map(|hh_item| { - let key = match &hh_item.key { - crate::HeapItem::String(s) => s.clone(), - other => format!("{other:?}"), - }; - WireHeapItem { - key, - value: hh_item.count as f64, - } - }) - .collect() -} - -/// Updates a CSHeap with a weighted key. Automatically updates the heap. -pub fn sketchlib_cs_heap_update(cs_heap: &mut SketchlibCSHeap, key: &str, value: f64) { - let many = value.round() as i64; - if many <= 0 { - return; - } - cs_heap.insert_many(&DataInput::String(key.to_owned()), many); -} - -/// Queries a CSHeap for a key's frequency estimate (median-of-signed-rows, -/// not CMS's min-over-rows). -pub fn sketchlib_cs_heap_query(cs_heap: &SketchlibCSHeap, key: &str) -> f64 { - cs_heap.estimate(&DataInput::String(key.to_owned())) -} - -/// Item in the top-k heap representing a key-value pair. -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct CsHeapItem { - pub key: String, - pub value: f64, -} - -/// Count Sketch with Heap for top-k tracking. Combines the balanced/ -/// zero-mean-error frequency estimator with efficient top-k maintenance. -pub struct CountSketchWithHeap { - pub rows: usize, - pub cols: usize, - pub heap_size: usize, - pub(crate) backend: SketchlibCSHeap, -} - -impl std::fmt::Debug for CountSketchWithHeap { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - f.debug_struct("CountSketchWithHeap") - .field("rows", &self.rows) - .field("cols", &self.cols) - .field("heap_size", &self.heap_size) - .finish() - } -} - -impl Clone for CountSketchWithHeap { - fn clone(&self) -> Self { - // Structural clone: `CSHeap` derives `Clone` (its `Count` + - // `HHHeap` fields are `Clone`), so copy the backend directly - // instead of extracting the matrix + heap to wire form and - // rebuilding it. - Self { - rows: self.rows, - cols: self.cols, - heap_size: self.heap_size, - backend: self.backend.clone(), - } - } -} - -impl CountSketchWithHeap { - pub fn new(rows: usize, cols: usize, heap_size: usize) -> Self { - Self { - rows, - cols, - heap_size, - backend: new_sketchlib_cs_heap(rows, cols, heap_size), - } - } - - pub fn rows(&self) -> usize { - self.rows - } - - pub fn cols(&self) -> usize { - self.cols - } - - /// Create from a sketch matrix and heap (e.g. from JSON deserialization). - pub fn from_legacy_matrix( - sketch: Vec>, - topk_heap: Vec, - rows: usize, - cols: usize, - heap_size: usize, - ) -> Self { - let wire_heap: Vec = topk_heap - .into_iter() - .map(|h| WireHeapItem { - key: h.key, - value: h.value, - }) - .collect(); - Self { - rows, - cols, - heap_size, - backend: sketchlib_cs_heap_from_matrix_and_heap( - rows, cols, heap_size, &sketch, &wire_heap, - ), - } - } - - /// Get the top-k heap items. - pub fn topk_heap_items(&self) -> Vec { - heap_to_wire(&self.backend) - .into_iter() - .map(|w| CsHeapItem { - key: w.key, - value: w.value, - }) - .collect() - } - - /// Get the sketch matrix. - pub fn sketch_matrix(&self) -> Vec> { - matrix_from_sketchlib_cs_heap(&self.backend) - } - - pub fn update(&mut self, key: &str, value: f64) { - sketchlib_cs_heap_update(&mut self.backend, key, value); - } - - pub fn estimate(&self, key: &str) -> f64 { - sketchlib_cs_heap_query(&self.backend, key) - } - - /// Merge another CountSketchWithHeap into self in place. Both - /// operands must have identical dimensions; the resulting - /// `heap_size` is the minimum of the two. - pub fn merge( - &mut self, - other: &CountSketchWithHeap, - ) -> Result<(), Box> { - if self.rows != other.rows || self.cols != other.cols { - return Err(format!( - "CountSketchWithHeap dimension mismatch: self={}x{}, other={}x{}", - self.rows, self.cols, other.rows, other.cols - ) - .into()); - } - self.backend.merge(&other.backend); - self.heap_size = self.heap_size.min(other.heap_size); - Ok(()) - } - - pub fn merge_refs( - inputs: &[&CountSketchWithHeap], - ) -> Result> { - let first = inputs - .first() - .ok_or("CountSketchWithHeap::merge_refs called with empty input")?; - let mut merged = (*first).clone(); - for h in inputs.iter().skip(1) { - merged.merge(h)?; - } - Ok(merged) - } - - pub fn aggregate_topk( - rows: usize, - cols: usize, - heap_size: usize, - keys: &[&str], - values: &[f64], - ) -> Option> { - if keys.is_empty() { - return None; - } - let mut sketch = Self::new(rows, cols, heap_size); - for (key, &value) in keys.iter().zip(values.iter()) { - sketch.update(key, value); - } - sketch.to_msgpack().ok() - } -} - -// ----- Wire format ----- - -/// Inner CS payload nested inside [`CountSketchWithHeapWire`]. -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct CountSketchInnerWire { - pub sketch: Vec>, - #[serde(rename = "row_num")] - pub rows: usize, - #[serde(rename = "col_num")] - pub cols: usize, -} - -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct CountSketchWithHeapWire { - pub sketch: CountSketchInnerWire, - pub topk_heap: Vec, - pub heap_size: usize, -} - -impl MessagePackCodec for CountSketchWithHeap { - fn to_msgpack(&self) -> Result, MsgPackError> { - let wire = CountSketchWithHeapWire { - sketch: CountSketchInnerWire { - sketch: self.sketch_matrix(), - rows: self.rows, - cols: self.cols, - }, - topk_heap: self.topk_heap_items(), - heap_size: self.heap_size, - }; - Ok(rmp_serde::to_vec(&wire)?) - } - - fn from_msgpack(bytes: &[u8]) -> Result { - let wire: CountSketchWithHeapWire = rmp_serde::from_slice(bytes)?; - - let mut sorted_topk_heap = wire.topk_heap; - sorted_topk_heap.sort_by(|a, b| b.value.partial_cmp(&a.value).unwrap()); - - let wire_heap: Vec = sorted_topk_heap - .iter() - .map(|h| WireHeapItem { - key: h.key.clone(), - value: h.value, - }) - .collect(); - let backend = sketchlib_cs_heap_from_matrix_and_heap( - wire.sketch.rows, - wire.sketch.cols, - wire.heap_size, - &wire.sketch.sketch, - &wire_heap, - ); - - Ok(Self { - rows: wire.sketch.rows, - cols: wire.sketch.cols, - heap_size: wire.heap_size, - backend, - }) - } -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn test_creation() { - let cs = CountSketchWithHeap::new(4, 1000, 20); - assert_eq!(cs.rows, 4); - assert_eq!(cs.cols, 1000); - assert_eq!(cs.heap_size, 20); - assert_eq!(cs.sketch_matrix().len(), 4); - assert_eq!(cs.sketch_matrix()[0].len(), 1000); - assert_eq!(cs.topk_heap_items().len(), 0); - } - - #[test] - fn test_query_empty() { - let cs = CountSketchWithHeap::new(2, 10, 5); - assert_eq!(cs.estimate("anything"), 0.0); - } - - #[test] - fn test_merge() { - let mut sketch1 = vec![vec![0.0; 10]; 2]; - sketch1[0][0] = 10.0; - sketch1[1][1] = 20.0; - let mut cs1 = CountSketchWithHeap::from_legacy_matrix( - sketch1, - vec![ - CsHeapItem { - key: "key1".to_string(), - value: 100.0, - }, - CsHeapItem { - key: "key2".to_string(), - value: 50.0, - }, - ], - 2, - 10, - 5, - ); - let mut sketch2 = vec![vec![0.0; 10]; 2]; - sketch2[0][0] = 5.0; - sketch2[1][1] = 15.0; - let cs2 = CountSketchWithHeap::from_legacy_matrix( - sketch2, - vec![ - CsHeapItem { - key: "key3".to_string(), - value: 75.0, - }, - CsHeapItem { - key: "key1".to_string(), - value: 80.0, - }, - ], - 2, - 10, - 3, - ); - - cs1.merge(&cs2).unwrap(); - - assert_eq!(cs1.sketch_matrix()[0][0], 15.0); - assert_eq!(cs1.sketch_matrix()[1][1], 35.0); - assert_eq!(cs1.heap_size, 3); - assert!(cs1.topk_heap_items().len() <= 3); - } - - #[test] - fn test_merge_dimension_mismatch() { - let mut cs1 = CountSketchWithHeap::new(2, 10, 5); - let cs2 = CountSketchWithHeap::new(3, 10, 5); - assert!(cs1.merge(&cs2).is_err()); - } - - #[test] - fn test_msgpack_round_trip() { - let mut cs = CountSketchWithHeap::new(4, 128, 3); - cs.update("hot", 100.0); - cs.update("cold", 1.0); - - let bytes = cs.to_msgpack().unwrap(); - let deserialized = CountSketchWithHeap::from_msgpack(&bytes).unwrap(); - - assert_eq!(deserialized.rows, 4); - assert_eq!(deserialized.cols, 128); - assert_eq!(deserialized.heap_size, 3); - let items = deserialized.topk_heap_items(); - assert!(!items.is_empty()); - let hot = items - .iter() - .find(|item| item.key == "hot") - .expect("'hot' should be in the heap"); - assert!(hot.value >= 100.0); - assert!(deserialized.estimate("hot") >= 100.0); - assert!(deserialized.estimate("cold") >= 1.0); - } - - #[test] - fn test_aggregate_topk() { - let keys = ["a", "b", "a", "c"]; - let values = [1.0, 2.0, 3.0, 0.5]; - let bytes = CountSketchWithHeap::aggregate_topk(4, 100, 2, &keys, &values).unwrap(); - let cs = CountSketchWithHeap::from_msgpack(&bytes).unwrap(); - assert_eq!(cs.heap_size, 2); - assert!(cs.topk_heap_items().len() <= 2); - } - - #[test] - fn test_aggregate_topk_empty() { - assert!(CountSketchWithHeap::aggregate_topk(4, 100, 10, &[], &[]).is_none()); - } - - /// The whole point of splitting this from `CountMinSketchWithHeap`: - /// CS's median estimator must behave differently from CMS's - /// min-over-rows estimator on the same kind of skewed insert - /// pattern, not just share a struct shape. - #[test] - fn test_median_estimator_differs_from_cms_style_min() { - let mut cs = CountSketchWithHeap::new(5, 64, 10); - for _ in 0..50 { - cs.update("heavy", 1.0); - } - for i in 0..20 { - cs.update(&format!("light-{i}"), 1.0); - } - let est = cs.estimate("heavy"); - assert!( - (40.0..=60.0).contains(&est), - "median estimate {est} should track the true count (50) reasonably \ - closely, not collide-and-inflate the way a single min-over-rows \ - outlier could" - ); - } -} diff --git a/src/message_pack_format/portable/mod.rs b/src/message_pack_format/portable/mod.rs index 97e0d372..6524c04c 100644 --- a/src/message_pack_format/portable/mod.rs +++ b/src/message_pack_format/portable/mod.rs @@ -10,7 +10,6 @@ pub mod countminsketch; pub mod countminsketch_topk; pub mod countsketch; -pub mod countsketch_topk; pub mod ddsketch; pub mod delta_set_aggregator; pub mod hll; diff --git a/tests/asapv1_golden.rs b/tests/asapv1_golden.rs index 30b41651..29acbf67 100644 --- a/tests/asapv1_golden.rs +++ b/tests/asapv1_golden.rs @@ -9,8 +9,8 @@ //! See `asapv1_golden/README.md`. use asap_sketchlib::{ - Classic, Count, CountMin, ErtlMLE, FastPath, HllSketch, HllVariant, HyperLogLogHIPP12, - HyperLogLogP12, KLL, MessagePackCodec, RegularPath, Vector2D, + CSHeap, Classic, Count, CountMin, DataInput, ErtlMLE, FastPath, HllSketch, HllVariant, + HyperLogLogHIPP12, HyperLogLogP12, KLL, MessagePackCodec, RegularPath, Vector2D, }; fn decode_hex(s: &str) -> Vec { @@ -30,6 +30,7 @@ const GOLDEN_CMS_F64: &str = include_str!("../asapv1_golden/cms_f64_fast_2x3.hex const GOLDEN_CS_REGULAR: &str = include_str!("../asapv1_golden/cs_i64_regular_2x4.hex"); const GOLDEN_CS_FAST: &str = include_str!("../asapv1_golden/cs_i64_fast_2x4.hex"); const GOLDEN_CS_I32: &str = include_str!("../asapv1_golden/cs_i32_regular_2x4.hex"); +const GOLDEN_CS_HEAP: &str = include_str!("../asapv1_golden/csheap_i64_regular_2x4_strkeys.hex"); const GOLDEN_KLL_F64: &str = include_str!("../asapv1_golden/kll_f64_k200.hex"); const GOLDEN_KLL_I64: &str = include_str!("../asapv1_golden/kll_i64_k200.hex"); @@ -242,6 +243,59 @@ fn cs_i32_regular_2x4_matches_golden() { ); } +// --------------------------------------------------------------------------- +// CSHeap: the Count Sketch matrix above plus a heap whose entries are seated +// directly, never by inserting through the sketch. +// --------------------------------------------------------------------------- + +/// Heap entries of the CSHeap fixture, in seating order. The counts span +/// uint64, a positive fixint tie (ordered by key on the wire) and a negative +/// int8; a CSHeap heap count is a signed median and may be negative. +const CS_HEAP_ENTRIES: [(&str, i64); 4] = [ + ("gamma", -33), + ("delta", 127), + ("alpha", 4_294_967_296), + ("beta", 127), +]; +const CS_HEAP_K: usize = 5; + +#[test] +fn csheap_i64_regular_2x4_strkeys_matches_golden() { + let want = decode_hex(GOLDEN_CS_HEAP); + + let mut sketch = CSHeap::, RegularPath>::from_storage( + Vector2D::from_fn(2, 4, |r, c| CS_VALS[r][c]), + CS_HEAP_K, + ); + for (key, count) in CS_HEAP_ENTRIES { + sketch.heap_mut().update(&DataInput::Str(key), count); + } + let got = sketch.serialize_to_bytes().expect("serialize"); + assert_eq!(got, want, "CSHeap i64/regular bytes diverge from golden"); + + let decoded = + CSHeap::, RegularPath>::deserialize_from_bytes(&want).expect("decode"); + let flat: Vec = CS_VALS.iter().flatten().copied().collect(); + assert_eq!(decoded.cs().as_storage().as_slice(), flat.as_slice()); + assert_eq!((decoded.rows(), decoded.cols()), (2, 4)); + assert_eq!(decoded.heap().capacity(), CS_HEAP_K); + assert_eq!(decoded.heap().len(), CS_HEAP_ENTRIES.len()); + for (key, count) in CS_HEAP_ENTRIES { + let seat = decoded + .heap() + .find(&DataInput::Str(key)) + .unwrap_or_else(|| panic!("heap key {key:?} missing after decode")); + assert_eq!(decoded.heap().heap()[seat].count, count, "heap key {key:?}"); + } + assert_eq!(decoded.serialize_to_bytes().expect("re-serialize"), want); + + // A CSHeap golden is not a Count Sketch envelope. + assert!( + Count::, RegularPath>::deserialize_from_bytes(&want).is_err(), + "the CSHeap golden must not decode as a plain Count Sketch" + ); +} + // --------------------------------------------------------------------------- // KLL: build known state (k=200, seed 42, integers 1..=50 — below the level-0 // capacity, so no compaction fires and the retained set is deterministic) -> diff --git a/tests/e2e/composition.rs b/tests/e2e/composition.rs index 23e42c41..3c45a728 100644 --- a/tests/e2e/composition.rs +++ b/tests/e2e/composition.rs @@ -16,9 +16,9 @@ use common::specs::{CardinalityConfidenceSpec, CountMinSpec, CountSketchSpec, Kl use common::{FreqTruth, NumericTruth, uniform_u64, zipf_u64}; use asap_sketchlib::{ - Classic, Count, CountMin, CountMinSketchWithHeap, CountSketchWithHeap, DataInput, - EnsembleSketch, ErtlMLE, FastPath, HashSketchEnsemble, HyperLogLog, HyperLogLogHIP, KllSketch, - MessagePackCodec, UnivMonQ, UnivMonQConfig, Vector2D, + Classic, Count, CountMin, CountMinSketchWithHeap, DataInput, EnsembleSketch, ErtlMLE, FastPath, + HashSketchEnsemble, HyperLogLog, HyperLogLogHIP, KllSketch, MessagePackCodec, UnivMonQ, + UnivMonQConfig, Vector2D, }; const ROWS: usize = 3; @@ -636,67 +636,6 @@ fn portable_count_min_with_heap_satisfies_the_count_min_bound_through_merge_and_ ); } -#[test] -fn portable_count_sketch_with_heap_satisfies_the_l2_bound_through_merge_and_wire() { - const HEAP: usize = 32; - const CS_ROWS: usize = 5; - let stream = zipf_u64(N, DOMAIN, 1.1, STREAM_SEED); - let mut truth = FreqTruth::default(); - let mut single = CountSketchWithHeap::new(CS_ROWS, COLS, HEAP); - let mut left = CountSketchWithHeap::new(CS_ROWS, COLS, HEAP); - let mut right = CountSketchWithHeap::new(CS_ROWS, COLS, HEAP); - for (i, k) in stream.iter().enumerate() { - truth.observe(*k as i64); - let key = format!("k{k}"); - single.update(&key, 1.0); - if i % 2 == 0 { - left.update(&key, 1.0); - } else { - right.update(&key, 1.0); - } - } - - let context = format!( - "rows={CS_ROWS} cols={COLS} heap={HEAP} zipf(1.1) domain={DOMAIN} n={N} seed={STREAM_SEED:#x}" - ); - let spec = CountSketchSpec::new(CS_ROWS, COLS); - spec.assert_contract( - "portable CountSketchWithHeap", - &truth, - |k| single.estimate(&format!("k{k}")), - &context, - ); - - let mut heap_tally = Tally::default(); - for item in single.topk_heap_items() { - let est = single.estimate(&item.key); - heap_tally.record(item.value == est, || { - format!( - "key {}: heap holds {} but the sketch estimates {est}", - item.key, item.value - ) - }); - } - heap_tally.assert_none("portable CountSketchWithHeap heap consistency", &context); - - let merged = CountSketchWithHeap::merge_refs(&[&left, &right]).expect("merge"); - spec.assert_contract( - "portable CountSketchWithHeap after merge", - &truth, - |k| merged.estimate(&format!("k{k}")), - &context, - ); - - let bytes = single.to_msgpack().expect("encode"); - let decoded = CountSketchWithHeap::from_msgpack(&bytes).expect("decode"); - spec.assert_contract( - "portable CountSketchWithHeap after a wire round trip", - &truth, - |k| decoded.estimate(&format!("k{k}")), - &context, - ); -} - /// The portable KLL facade under the DataSketches maximum-rank-error /// characterization, seeded so a failure reproduces. `KllSketch::new` seeds /// from the wall clock; `with_seed` is what an accuracy test must use.