From 747ed0934516d7d854a45c057750931eeed17427 Mon Sep 17 00:00:00 2001 From: GordonYuanyc Date: Fri, 2 Oct 2026 00:40:42 -0600 Subject: [PATCH 1/2] refactor(countsketch)!: drop the proto and portable formats Count Sketch serializes as ASAPv1 only. Removes the portable CountSketch, CountSketchDelta and COUNT_SKETCH_TOPK_CAPACITY, the native MessagePackCodec shim for Count, the proto CountSketchDelta and CountSketchCell messages, and SketchEnvelope's count_sketch field (11, now reserved). CountSketchState stays: UnivMon and Hydra embed it. Tests that exercised Count Sketch through the removed formats are dropped (proto parity probe, msgpack round trip, xtest consumer case, portable e2e bound) or ported to the native Count (the four portable property laws). Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 15 + docs/message_pack_format.md | 4 +- docs/tests.md | 33 +- proto/countminsketch/countminsketch.proto | 10 +- proto/countsketch/countsketch.proto | 53 +- proto/sketchlib.proto | 3 +- src/common/hashspec.rs | 8 +- src/lib.rs | 5 +- src/message_pack_format/native/countsketch.rs | 32 - src/message_pack_format/native/mod.rs | 1 - .../portable/countminsketch.rs | 24 +- .../portable/countsketch.rs | 974 ------------------ src/message_pack_format/portable/mod.rs | 3 +- src/proto/generated/sketchlib.v1.rs | 78 +- tests/PBT_COVERAGE.md | 6 +- tests/cs_envelope_golden.hex | 1 - tests/e2e/frequency.rs | 25 +- tests/msgpack_compat.rs | 15 +- tests/pbt/count_sketch.rs | 115 +-- tests/sketches_go_parity_probe.rs | 66 +- tests/xtest_consumer.rs | 71 -- 21 files changed, 115 insertions(+), 1427 deletions(-) delete mode 100644 src/message_pack_format/native/countsketch.rs delete mode 100644 src/message_pack_format/portable/countsketch.rs delete mode 100644 tests/cs_envelope_golden.hex diff --git a/CHANGELOG.md b/CHANGELOG.md index 80d85f98..85cd72de 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -88,6 +88,21 @@ signals a backwards-compatible change. as an empty window like any other, and that window takes a place in `max_windows`. +### Removed + +- **Count Sketch's proto and portable formats.** Count Sketch (`Count`) + serializes as ASAPv1 only (`serialize_to_bytes` / `deserialize_from_bytes`, + kind `0x04 0x00`). Removed: the portable `CountSketch`, `CountSketchDelta` + and `COUNT_SKETCH_TOPK_CAPACITY` (module + `message_pack_format::portable::countsketch` and their crate-root + re-exports), including `CountSketch`'s proto and MessagePack delta codecs + (`compute_delta`, `apply_delta_bytes`, `compute_delta_msgpack`, + `apply_delta_msgpack_bytes`); the `MessagePackCodec` impl for `Count` + (module `message_pack_format::native::countsketch`); the proto messages + `CountSketchDelta` and `CountSketchCell`; and `SketchEnvelope`'s + `count_sketch` field (11, now reserved). `CountSketchState` stays, as + UnivMon and Hydra embed it. + ### Fixed - `KLLDynamic` panicked on every quantile query once a NaN had entered the diff --git a/docs/message_pack_format.md b/docs/message_pack_format.md index 12a2b91c..a229a8d9 100644 --- a/docs/message_pack_format.md +++ b/docs/message_pack_format.md @@ -150,8 +150,8 @@ being **phased out** in favor of the per-sketch `wire.rs` + the shared `envelope.rs`. - `portable/` holds per-sketch wire types that predate the envelope. They - carry their own byte-parity goldens against `sketchlib-go` — HLL, Count-Min, - Count Sketch and KLL (both directions) — separate from `asapv1_golden/`. + carry their own byte-parity goldens against `sketchlib-go` — HLL, Count-Min + and KLL (both directions) — separate from `asapv1_golden/`. - `native/` is a set of thin `MessagePackCodec` shims over the sketches' `serialize_to_bytes` / `deserialize_from_bytes`. Every type it wraps emits the ASAPv1 envelope, so each shim is a pass-through. diff --git a/docs/tests.md b/docs/tests.md index ccb1571c..47732b4d 100644 --- a/docs/tests.md +++ b/docs/tests.md @@ -1254,31 +1254,6 @@ Test file: [`src/message_pack_format/portable/countminsketch_topk.rs`](../src/me | `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`. | -### Portable CountSketch - -Test file: [`src/message_pack_format/portable/countsketch.rs`](../src/message_pack_format/portable/countsketch.rs) - -| test_name | test_description | what_is_tested | -| --- | --- | --- | -| `msgpack_delta_against_empty_round_trips` | A msgpack delta against an empty snapshot carries the signed cells exactly. | For a `5x1024` sketch fed 200 updates of `"hot"` at weight `1.0`, verifies the delta against a fresh sketch at threshold `1.0`, applied to another fresh sketch, reproduces `sketch()` cell for cell. | -| `msgpack_delta_wire_layout_is_parallel_arrays` | The delta is a five-element array of scalars plus parallel index arrays. | For a `3x8` sketch fed `"x"` at weight `5.0`, verifies the delta bytes equal `rmp_serde::to_vec(&(rows as u64, cols as u64, row_idx: Vec, col_idx: Vec, d_count: Vec))` recomputed from the non-zero cells, and that decoding reports `rows` `3` and `cols` `8`. | -| `test_new_empty` | A fresh sketch has the requested shape and zeroed cells. | For `CountSketch::new(2, 3)`, verifies `rows` is `2`, `cols` is `3`, and `sketch()` is `[[0.0, 0.0, 0.0], [0.0, 0.0, 0.0]]`. | -| `test_from_legacy_matrix` | A legacy matrix is adopted unchanged, signs included. | Verifies `from_legacy_matrix([[1.0, -2.0, 3.0], [-4.0, 5.0, -6.0]], 2, 3)` reports exactly that matrix from `sketch()`. | -| `test_merge_element_wise` | Merge sums signed cells element-wise. | Merges the `2x2` matrix `[[-1.0, -2.0], [-3.0, -4.0]]` into `[[1.0, 2.0], [3.0, 4.0]]` and verifies every merged cell is `0.0`. | -| `test_merge_dimension_mismatch` | Mismatched row counts are refused. | Verifies merging a `3x3` sketch into a `2x3` one fails. | -| `test_merge_refs` | `merge_refs` sums every input. | Merges three `1x2` sketches holding `[1.0, 2.0]`, `[3.0, 4.0]`, and `[5.0, 6.0]` and verifies the result is `[[9.0, 12.0]]`. | -| `test_apply_delta_additive` | Delta cells add into the matrix, negative weights included. | For a `2x3` sketch holding `[[1.0, -2.0, 3.0], [-4.0, 5.0, -6.0]]`, applies cells `(0,0,10)`, `(0,2,-3)`, and `(1,1,-15)` and verifies the matrix becomes `[[11.0, -2.0, 0.0], [-4.0, -10.0, -6.0]]`. | -| `test_apply_delta_matches_full_merge` | Applying a delta lands where merging the same addition lands. | For a `2x2` base `[[1.0, 2.0], [3.0, 4.0]]`, verifies merging the addition `[[10.0, 0.0], [0.0, 20.0]]` and applying the delta cells `(0,0,10)` and `(1,1,20)` give the same matrix. | -| `test_apply_delta_out_of_range` | A cell outside the declared geometry is refused. | Verifies a delta carrying cell `(2, 0, 1)` against a two-row `2x3` sketch fails to apply. | -| `test_apply_delta_rebuilds_topk_from_hh_keys` | The `hh_keys` channel rebuilds Top-K by re-querying the merged matrix. | For a `3x16` sketch fed `"alpha"` at `5.0` and `"beta"` at `3.0`, applies a delta with no cells and `hh_keys` `["alpha", "beta"]` and verifies `topk` holds both keys and that `"alpha"` reads above `"beta"`. | -| `test_apply_delta_hh_keys_topk_capacity` | The rebuilt Top-K is bounded by `COUNT_SKETCH_TOPK_CAPACITY`. | For a `3x1024` sketch fed 105 keys `k0000`..`k0104` at weights `1.0`..`105.0`, applies a delta whose `hh_keys` carries all 105 and verifies `topk` holds exactly `COUNT_SKETCH_TOPK_CAPACITY` (`100`) entries; which entries survive is not checked. | -| `test_msgpack_round_trip` | The msgpack round trip preserves the signed matrix and its shape. | Round-trips a `2x2` sketch holding `[[1.5, -2.5], [3.5, -4.5]]` and verifies the matrix, `rows`, and `cols` are unchanged. | -| `test_update_then_envelope_matches_sketchlib_go_bytes` | The proto envelope for a populated matrix is byte-identical to the golden. | For a `3x512` sketch fed 25 updates over the five keys `k-a`..`k-e` (each five times) at weight `1.0`, builds a `CountSketchState` with `counter_type` `INT64`, row-major `counts_int`, per-row `l2` of `125.0`, and no `topk`, wraps it in a `SketchEnvelope` with `format_version` `1`, and verifies the prost encoding equals a 1,577-byte inline golden. | -| `test_hh_keys_matches_go_golden_bytes` | Heavy-hitter keys at tag 6 reach the wire in the golden's shape. | For the same `3x512` sketch over `k-a`..`k-e`, decodes the packed delta against an empty sketch as a proto `CountSketchDelta`, sets `hh_keys` to `["flow-0", "flow-3", "flow-7"]`, and verifies the re-encoding equals a 117-byte inline golden. | -| `test_compute_delta_against_empty_round_trips` | A proto delta against an empty snapshot carries the whole signed matrix. | For a `3x512` sketch fed 200 updates of `k-{i % 23}` at weight `1.0`, verifies the delta against an empty sketch at threshold `1.0`, applied to a fresh `3x512` sketch, reproduces `sketch()` cell for cell. | -| `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) @@ -1606,13 +1581,13 @@ Test file: [`tests/e2e/frequency.rs`](../tests/e2e/frequency.rs) | `fold_cms_foldcs_counts_and_hierarchical_merge` | Weighted counts are exact on sparse dimensions and survive a same-level merge, a signed update, and a hierarchical merge. | On `FoldCMS::::new(3, 2048, 0, 32)` sketches: 100 inserts of `Str("alpha")` at weight 2 into each of two sketches plus 100 of `Str("beta")` at weight 1 into the second, verifying `query` reads `200`, `200` and `100`; after `merge_same_level`, `400` for alpha and `100` for beta. On `FoldCS::::new(3, 2048, 0, 32)`, `Str("gamma")` at `+60` then `-20` is verified to query `40`. Finally `FoldCMS::hierarchical_merge` of two sketches built by `build_fold_cms` (30 and 45 inserts at weight 3) is verified to query `90` and `135`. | | `folded_sketches_keep_their_own_bounds_through_a_sixteen_way_merge` | Folded sketches keep their own theorems, evaluated at the folded width, through a sixteen-way hierarchical merge. | Splits `zipf_u64(240_000, 10_000, 1.1, 1009)` into 16 windows of 15,000, giving each window a `FoldCMS` and a `FoldCS` of `3x4096` at `fold_level = 4` and `top_k = 20`, then `hierarchical_merge`s each family. Asserts `CountMinSpec::new(3, 256)` over `FoldCMS::query` and `CountSketchSpec::new(3, 256)` over `FoldCS::query` - the bounds are evaluated at the folded width `4096 >> 4 = 256`, not the unfolded 4096 - with the specs' usual rules: zero violations for one-sidedness and both simultaneous bounds at `delta = 1e-3`, and the marginal bounds as rate pins at `e^-3` and `P[Bin(3, 1/3) >= 2]`. | -### Frequency: Portable Count-Min and Count Sketch +### Frequency: Portable Count-Min Test file: [`tests/e2e/frequency.rs`](../tests/e2e/frequency.rs) | test_name | test_description | what_is_tested | | --- | --- | --- | -| `portable_cms_and_cs_string_keys_satisfy_their_own_bounds` | The portable wire twins satisfy their own theorems over string keys with no extra slack. | Feeds `zipf_u64(50_000, 2048, 1.1, 1006)` as string keys `"k{k}"` through `update(&key, 1.0)` into `portable::countminsketch::CountMinSketch::new(3, 4096)` and `portable::countsketch::CountSketch::new(5, 4096)`, then asserts `CountMinSpec::new(3, 4096).assert_contract` over the portable Count-Min's `estimate` and `CountSketchSpec::new(5, 4096).assert_contract` over the portable Count Sketch's `estimate`, against a `FreqTruth` keyed on the underlying `i64` identities. | +| `portable_cms_string_keys_satisfy_their_own_bound` | The portable Count-Min satisfies its own theorem over string keys with no extra slack. | Feeds `zipf_u64(50_000, 2048, 1.1, 1006)` as string keys `"k{k}"` through `update(&key, 1.0)` into `portable::countminsketch::CountMinSketch::new(3, 4096)`, then asserts `CountMinSpec::new(3, 4096).assert_contract` over its `estimate`, against a `FreqTruth` keyed on the underlying `i64` identities. | ### Frequency: Documented Input Matrix @@ -2143,7 +2118,6 @@ Test file: [`tests/msgpack_compat.rs`](../tests/msgpack_compat.rs) | `count_min_sketch_with_heap_round_trip` | A portable `CountMinSketchWithHeap` survives a msgpack round trip with its geometry and heap size intact. | Updates `CountMinSketchWithHeap::new(3, 64, 8)` with `hot` `+100.0` and `warm` `+10.0`, then verifies the decode reports `rows` `3`, `cols` `64` and `heap_size` `8`; the heap's contents are not checked. | | `count_min_wire_shape` | The `CountMinSketchWire` DTO round-trips through `rmp_serde` with its field shape intact. | Serializes `CountMinSketchWire { sketch: vec![vec![1.0, 2.0]; 3], rows: 3, cols: 2 }` with `rmp_serde::to_vec` and reads it back with `rmp_serde::from_slice`, verifying `rows` `3`, `cols` `2` and a `sketch` of length `3`; the cell values are not checked. | | `count_min_with_heap_wire_shape` | The nested `CountMinSketchWithHeapWire` DTO round-trips through `rmp_serde` with its nesting intact. | Serializes a wire struct holding a `CountMinSketchInnerWire` of `2x4` zeros, one `CmsHeapItem { key: "hot", value: 42.0 }` and `heap_size: 8`, then verifies the `rmp_serde` decode reports `heap_size` `8`, one heap entry and the key `hot`; the item's value and the inner matrix are not re-checked. | -| `count_sketch_round_trip` | A portable `CountSketch` survives a msgpack round trip with its geometry intact. | Updates `CountSketch::new(3, 64)` with `k1` `+1.0` and `k2` `+2.0`, then verifies the decode reports `rows` `3` and `cols` `64`; the signed counters and the estimates are not checked. | | `dd_sketch_decodes_go_bytes` | The test asserts nothing. | The test asserts nothing: the body is empty and the test is marked `#[ignore]` with the reason that no cross-language msgpack fixture exists in the repository. | | `dd_sketch_round_trip` | A portable `DdSketch` survives a msgpack round trip carrying its full count in the bucket array. | Updates `DdSketch::new(0.01)` with `1.0`, `10.0` and `100.0`, then verifies the decode's `total_count()` — recovered by summing the buckets, since the count is not on the wire — is `3`; `alpha`, the bucket offset and the quantiles are not checked. | | `delta_result_round_trip` | A `DeltaResult` survives a msgpack round trip with both of its sets. | Serializes a `DeltaResult` whose `added` holds `a` and whose `removed` holds `b`, then verifies the decode's `added` contains `a` and its `removed` contains `b`. | @@ -2191,7 +2165,6 @@ Test file: [`tests/sketches_go_parity_probe.rs`](../tests/sketches_go_parity_pro | test_name | test_description | what_is_tested | | --- | --- | --- | -| `sketches_count_fastpath_matches_go_count_sketch_envelope` | A `sketches::Count` fast-path matrix encodes to the checked-in Count-Sketch envelope bytes. | Fills `Count, FastPath, DefaultXxHasher>` at `3x512` with 25 `insert_many` calls over the five keys `k-a`..`k-e`, packs the row-major cells and per-row L2 sums into a `CountSketchState` with `counter_type` `INT64` inside a `SketchEnvelope { format_version: 1 }` with no producer or hash_spec, and verifies the prost encoding equals the 1,577-byte fixture `tests/cs_envelope_golden.hex`. | | `sketches_countmin_fastpath_matches_go_count_min_envelope` | A `sketches::CountMin` fast-path matrix encodes to the checked-in Count-Min envelope bytes. | Fills `CountMin, FastPath, DefaultXxHasher>` at `4x2048` with 50 `insert_many` calls over the ten keys `flow-0`..`flow-9`, packs the cells cast to `i64` plus per-row L1 and L2 sums into a `CountMinState` with `counter_type` `INT64`, and verifies the prost encoding matches the 8,275-byte fixture `src/sketches/testdata/cms_envelope_golden.hex` in length and then byte for byte. | | `sketches_ddsketch_matches_go_envelope` | A `sketches::DDSketch` encodes to the DDSketch envelope hex constant held in the test file. | Fills `DDSketch::new(0.01)` with `1..=50` as `f64`, encodes a `DdSketchState` carrying `store_counts`, `store_offset` and an alpha round-tripped through gamma (`gamma = (1 + a) / (1 - a)`, `alpha_wire = (gamma - 1) / (gamma + 1)`), and verifies the envelope matches the 403-byte inline hex constant in length and then byte for byte; the comment on that constant records it as this crate's own output rather than a fixture from another implementation, so no cross-language parity is pinned. | | `sketches_hll_classic_matches_go_envelope` | An ErtlMLE P14 register set encodes to the checked-in HyperLogLog envelope bytes. | Feeds `HyperLogLogImpl` the `hash64_seeded(CANONICAL_HASH_SEED, ...)` digests of the IEEE-754 little-endian bytes of `1.0..=50.0` through `insert_with_hash`, encodes a `HyperLogLogState` with `variant` `2`, `precision` `14`, the dense registers, zeroed HIP scalars and no sparse form, and verifies the envelope matches the 16,398-byte fixture `src/sketches/testdata/hll_envelope_golden.hex` in length and then byte for byte; despite the name, the variant under test is `ErtlMLE`, not `Classic`. | @@ -2202,7 +2175,7 @@ Test file: [`tests/xtest_consumer.rs`](../tests/xtest_consumer.rs) | test_name | test_description | what_is_tested | | --- | --- | --- | -| `cross_language_proto` | Protobuf envelopes read from the cross-language fixture directory answer their point queries in band. | Asserts nothing unless fixtures are available: with no `XTEST_DIR` and no discoverable producer directory holding `tests/cross_language/xtest_producer_test.go`, it prints a skip line and returns. With fixtures it decodes nine `SketchEnvelope` files and records a pass or failure for each — `countmin.pb` min-frequency for `item:42` at least `101`, `kll.pb` p50 within 5% of `5000` and p99 within 5% of `9900`, `ddsketch.pb` the same two quantiles within 2%, `hll.pb` an ErtlMLE cardinality inside `[40_000, 65_000]`, `countsketch.pb` `cs:hot` at least `200`, `coco.pb` `coco:hot` at least `500`, `elastic.pb` `elephant` at least `900`, `univmon.pb` a cardinality inside `[1_000, 15_000]`, `hydra.pb` `hydra:42` at least `51` — using the hash seeds `0xcafe3553`, `0x6a09e667` and `0xbb67ae85`, and, when `countmin_sampled.pb` and `hll_sampled.pb` exist, requires `effective_sample_p` of `0.1` and a `rescale_count` by `1/p` within 5% of `100_000` and 6% of `200_000`. A single `assert!` at the end fails if any check failed; a missing or mistyped envelope panics on the spot. | +| `cross_language_proto` | Protobuf envelopes read from the cross-language fixture directory answer their point queries in band. | Asserts nothing unless fixtures are available: with no `XTEST_DIR` and no discoverable producer directory holding `tests/cross_language/xtest_producer_test.go`, it prints a skip line and returns. With fixtures it decodes eight `SketchEnvelope` files and records a pass or failure for each — `countmin.pb` min-frequency for `item:42` at least `101`, `kll.pb` p50 within 5% of `5000` and p99 within 5% of `9900`, `ddsketch.pb` the same two quantiles within 2%, `hll.pb` an ErtlMLE cardinality inside `[40_000, 65_000]`, `coco.pb` `coco:hot` at least `500`, `elastic.pb` `elephant` at least `900`, `univmon.pb` a cardinality inside `[1_000, 15_000]`, `hydra.pb` `hydra:42` at least `51` — using the hash seeds `0xcafe3553`, `0x6a09e667` and `0xbb67ae85`, and, when `countmin_sampled.pb` and `hll_sampled.pb` exist, requires `effective_sample_p` of `0.1` and a `rescale_count` by `1/p` within 5% of `100_000` and 6% of `200_000`. A single `assert!` at the end fails if any check failed; a missing or mistyped envelope panics on the spot. | ### HLL Custom Precision diff --git a/proto/countminsketch/countminsketch.proto b/proto/countminsketch/countminsketch.proto index 27e09d3b..18a9dd8f 100644 --- a/proto/countminsketch/countminsketch.proto +++ b/proto/countminsketch/countminsketch.proto @@ -77,10 +77,9 @@ message CountMinState { // byte-identical delta frames for identical window state (cross-language byte // parity). // -// CountMinDelta is structurally identical to CountSketchDelta: rows, cols, the -// packed cell encoding, per-row norm deltas, and an optional repeated hh_keys -// at the SAME tag number (6). Both sketches can track heavy hitters; whether -// hh_keys is populated is a control-plane decision (it is empty/omitted when +// CountMinDelta carries rows, cols, the packed cell encoding, per-row norm +// deltas, and an optional repeated hh_keys (tag 6). Whether hh_keys is +// populated is a control-plane decision (it is empty/omitted when // heavy-hitter tracking is not enabled for this sketch). message CountMinDelta { uint32 rows = 1; @@ -92,8 +91,7 @@ message CountMinDelta { repeated double l1 = 4 [packed = true]; // Per-row L2 norm deltas, length = rows. repeated double l2 = 5 [packed = true]; - // Heavy-hitter candidate keys from an upstream tracker, mirroring - // CountSketchDelta.hh_keys (same tag number 6, same wire shape). Downstream + // Heavy-hitter candidate keys from an upstream tracker. Downstream // queries the merged Count-Min matrix for each key to (re)build its Top-K // with globally-merged estimates. Empty/omitted when heavy-hitter tracking // is not enabled. diff --git a/proto/countsketch/countsketch.proto b/proto/countsketch/countsketch.proto index ae921100..71e05513 100644 --- a/proto/countsketch/countsketch.proto +++ b/proto/countsketch/countsketch.proto @@ -1,5 +1,5 @@ -// countsketch.proto — Portable wire format for Count Sketch state. -// Also contains TopKState and HeapEntry used by CountSketch and UnivMon. +// countsketch.proto — CountSketchState, TopKState and HeapEntry, the +// messages UnivMon and Hydra embed. syntax = "proto3"; @@ -72,52 +72,3 @@ message HeapEntry { reserved 3; // Reserved for bytes_key alternative. } - -// ============================================================================ -// Count Sketch Delta -// ============================================================================ - -// CountSketchDelta carries only the matrix cells that changed by at least the -// threshold between two consecutive snapshots, plus the full per-row L2 norm -// deltas and any heavy-hitter candidate keys. -// -// Cells apply additively on the receiver: matrix[row][col] += d_count. Count -// Sketch counters are signed (±1 increments), so d_count is a signed integer -// (sint64). Heavy-hitter candidate keys (hh_keys) are re-queried against the -// merged matrix downstream to rebuild the receiver's Top-K with accurate, -// globally-merged estimates. -// -// The packed cell encoding (cell_rows / cell_cols / d_counts, tags 9-11) plus -// l2 (tag 12) is the canonical form; the legacy per-cell message (tag 3), -// l2_legacy (tag 4) and topk (tag 5) are retained only so payloads from older -// producers still decode. This message is byte-identical to the Go reference -// implementation's CountSketchDelta so the two runtimes emit byte-identical -// delta frames for identical window state (cross-language byte parity). -message CountSketchDelta { - uint32 rows = 1; - uint32 cols = 2; - // Deprecated: use cell_rows/cell_cols/d_counts (tags 9-11). Retained for - // backward-compatible decode of payloads from older producers. - repeated CountSketchCell cells_legacy = 3; - // Deprecated: use l2 (tag 12) instead. - repeated double l2_legacy = 4 [packed = true]; - // Deprecated: use hh_keys (tag 6) instead. - TopKState topk = 5; - // Heavy-hitter candidate keys from the upstream tracker. Downstream queries - // the merged Count Sketch matrix for each key to build its Top-K. - repeated string hh_keys = 6; - // Packed cell encoding (canonical form): - repeated uint32 cell_rows = 9 [packed = true]; // row index of each changed cell - repeated uint32 cell_cols = 10 [packed = true]; // col index of each changed cell - repeated sint64 d_counts = 11 [packed = true]; // signed integer count delta - repeated double l2 = 12 [packed = true]; // per-row L2 norm deltas -} - -// CountSketchCell is the deprecated per-cell delta record. Producers emit the -// packed cell arrays on CountSketchDelta instead; this message exists only to -// decode payloads from older producers. -message CountSketchCell { - uint32 row = 1; - uint32 col = 2; - double d_count = 3; // deprecated: use CountSketchDelta.d_counts -} diff --git a/proto/sketchlib.proto b/proto/sketchlib.proto index 0a6d8193..e96d83de 100644 --- a/proto/sketchlib.proto +++ b/proto/sketchlib.proto @@ -67,7 +67,6 @@ message SketchEnvelope { // The sketch payload. Exactly one field must be set. oneof sketch_state { CountMinState count_min = 10; - CountSketchState count_sketch = 11; HyperLogLogState hll = 12; KLLState kll = 13; DDSketchState ddsketch = 14; @@ -77,6 +76,8 @@ message SketchEnvelope { ElasticState elastic = 18; } + reserved 11; + reserved "count_sketch"; reserved 19 to 31; // Reserved for future sketch types. // Geometric / NitroSketch sampling probability for the producing sketch. diff --git a/src/common/hashspec.rs b/src/common/hashspec.rs index 13c44c19..0c653029 100644 --- a/src/common/hashspec.rs +++ b/src/common/hashspec.rs @@ -13,8 +13,7 @@ //! ±1 signs. //! //! This module re-exposes that pipeline as a small, byte-key API so -//! sketches that consume keys as raw bytes (e.g. the wire-format -//! `CountSketch::update`) can match Go's emitted matrix +//! sketches that consume keys as raw bytes can match Go's emitted matrix //! cell-for-cell. CountMinSketch's pending byte-parity fix will reuse //! the same primitives — the seed table, derive_index, and derive_sign //! are agnostic to whether ±1 signing is applied. @@ -75,10 +74,7 @@ pub const CANONICAL_HASH_SEED: usize = 5; /// derivation strategy so consumers can validate hash compatibility /// before merging. /// -/// `asap_sketchlib`'s wire-format CountSketch constructs a -/// `HashSpec::default()` and feeds it to [`derive_index`] / -/// [`derive_sign`] on the hot path. The default matches Go's -/// `portableHashSpec()`: +/// The default matches Go's `portableHashSpec()`: /// /// - `seed_list = CANONICAL_HASH_SEED_TABLE` /// - `canonical_seed_index = CANONICAL_HASH_SEED` (= 5) diff --git a/src/lib.rs b/src/lib.rs index b062b8b9..47af9d8e 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -10,7 +10,7 @@ //! UnivMon, tumbling windows, and batch/parallel execution helpers. //! - [`message_pack_format`]: MessagePack/proto wire format shared with //! `sketchlib-go`. Owns the wire-DTO sketch types consumed by the ASAP -//! query engine (`CountMinSketch`, `CountSketch`, `KllSketch`, `HllSketch`, +//! query engine (`CountMinSketch`, `KllSketch`, `HllSketch`, //! `DdSketch`, `HydraKllSketch`, `CountMinSketchWithHeap`, //! `SetAggregator`, `DeltaResult`). //! @@ -43,9 +43,6 @@ pub use common::*; pub use message_pack_format::MessagePackCodec; pub use message_pack_format::portable::countminsketch::{CountMinSketch, CountMinSketchDelta}; pub use message_pack_format::portable::countminsketch_topk::{CmsHeapItem, CountMinSketchWithHeap}; -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; diff --git a/src/message_pack_format/native/countsketch.rs b/src/message_pack_format/native/countsketch.rs deleted file mode 100644 index 89375065..00000000 --- a/src/message_pack_format/native/countsketch.rs +++ /dev/null @@ -1,32 +0,0 @@ -//! Native MessagePack codec impl for [`crate::sketches::countsketch::Count`]. -//! -//! Only the canonical wire configs — `i32` or `i64` counters (`CsWireCounter`) -//! with a fast/regular mode (`CsWireMode`) — are serializable. There is no `f64` -//! counterpart to Count-Min's, because `CountSketchCounter` is implemented only -//! for `i32`, `i64` and `i128`; `i128` and non-`Vector2D` storage must be -//! converted first. - -use serde::{Deserialize, Serialize}; - -use crate::message_pack_format::{Error, MessagePackCodec}; -use crate::sketches::countsketch::{Count, CountSketchCounter, CsWireCounter, CsWireMode}; -use crate::{HashProfile, SketchHasher, Vector2D}; - -impl MessagePackCodec for Count, Mode, H> -where - T: CsWireCounter - + CountSketchCounter - + std::ops::AddAssign - + Serialize - + for<'de> Deserialize<'de>, - Mode: CsWireMode, - H: SketchHasher + HashProfile, -{ - fn to_msgpack(&self) -> Result, Error> { - Ok(self.serialize_to_bytes()?) - } - - fn from_msgpack(bytes: &[u8]) -> Result { - Ok(Self::deserialize_from_bytes(bytes)?) - } -} diff --git a/src/message_pack_format/native/mod.rs b/src/message_pack_format/native/mod.rs index 0a83ef96..2d01b364 100644 --- a/src/message_pack_format/native/mod.rs +++ b/src/message_pack_format/native/mod.rs @@ -8,7 +8,6 @@ //! specifies the format `sketchlib-go` mirrors. pub mod countminsketch; -pub mod countsketch; pub mod countsketch_topk; pub mod ddsketch; pub mod hll; diff --git a/src/message_pack_format/portable/countminsketch.rs b/src/message_pack_format/portable/countminsketch.rs index 16087bdb..e6b000a1 100644 --- a/src/message_pack_format/portable/countminsketch.rs +++ b/src/message_pack_format/portable/countminsketch.rs @@ -90,9 +90,8 @@ pub fn sketchlib_cms_query(inner: &SketchlibCms, key: &str) -> f64 { /// L1 and L2 norm deltas are carried for downstream error-accounting /// but are not consumed by `apply_delta` itself. /// -/// Structurally identical to [`super::countsketch::CountSketchDelta`]: the -/// optional `hh_keys` channel carries heavy-hitter candidate keys forwarded by -/// an upstream tracker. Count-Min can track heavy hitters, but whether +/// The optional `hh_keys` channel carries heavy-hitter candidate keys +/// forwarded by an upstream tracker. Count-Min can track heavy hitters, but whether /// `hh_keys` is populated is a control-plane decision — it is empty when /// heavy-hitter tracking is not enabled. The receiver re-queries the merged /// matrix for each key to (re)build its Top-K. Mirrors the Go reference @@ -266,9 +265,8 @@ impl CountMinSketch { /// Go producer maintains incrementally. Heavy-hitter candidate keys /// (`hh_keys`) are sourced from an upstream tracker that this minimal /// wrapper does not maintain, so the field is left empty (an empty repeated - /// field encodes to nothing on the wire), exactly as CountSketch's - /// `compute_delta` does. The returned bytes are a `prost`-encoded - /// [`crate::proto::sketchlib::CountMinDelta`], byte-identical to the Go + /// field encodes to nothing on the wire). The returned bytes are a + /// `prost`-encoded [`crate::proto::sketchlib::CountMinDelta`], byte-identical to the Go /// `proto.Marshal(CountMinDelta)` output for the same inputs when no /// heavy-hitter keys are forwarded (cross-language byte parity). /// @@ -334,7 +332,7 @@ impl CountMinSketch { l1, l2, // Heavy-hitter keys are control-plane-gated; this wrapper has no - // tracker, so the field is empty (mirrors CountSketch::compute_delta). + // tracker, so the field is empty. hh_keys: Vec::new(), cell_rows, cell_cols, @@ -348,9 +346,9 @@ impl CountMinSketch { /// Go reference implementation's `DeserializeDelta` + `ApplyDelta`. /// Reads the packed cell arrays (`cell_rows`/`cell_cols`/`d_counts`) /// and falls back to the legacy per-cell records for payloads from - /// older producers. Heavy-hitter keys (`hh_keys`) are read symmetrically - /// (mirroring CountSketch) and carried onto the decoded delta; a plain CMS - /// keeps no Top-K, so they are not otherwise consumed here. + /// older producers. Heavy-hitter keys (`hh_keys`) are read and carried + /// onto the decoded delta; a plain CMS keeps no Top-K, so they are not + /// otherwise consumed here. /// /// Returns `Err` if `bytes` is not a valid `CountMinDelta` proto or a /// cell is out of range for this sketch's dimensions. @@ -383,10 +381,8 @@ impl CountMinSketch { cells, l1: proto.l1, l2: proto.l2, - // Heavy-hitter keys are carried through symmetrically (mirrors - // CountSketch). This wrapper keeps no Top-K, so they are not - // otherwise consumed; a control-plane sink would re-query the - // merged matrix for each key. Empty for a plain CMS. + // This wrapper keeps no Top-K, so heavy-hitter keys are carried + // through but not otherwise consumed. Empty for a plain CMS. hh_keys: proto.hh_keys, }; self.apply_delta(&delta) diff --git a/src/message_pack_format/portable/countsketch.rs b/src/message_pack_format/portable/countsketch.rs deleted file mode 100644 index a94ad981..00000000 --- a/src/message_pack_format/portable/countsketch.rs +++ /dev/null @@ -1,974 +0,0 @@ -//! Wire-format-aligned Count Sketch types. The wire DTO and runtime -//! ops live together here. - -use serde::{Deserialize, Serialize}; - -use crate::message_pack_format::{Error as MsgPackError, MessagePackCodec}; - -// ===================================================================== -// ASAP runtime wire-format-aligned variant . -// -// `CountSketch` and `CountSketchDelta` below are the public-field, -// proto-decode-friendly types consumed by the ASAP query engine -// accumulators. The high-throughput in-process variant is -// `crate::sketches::countsketch::Count`, which keeps its own design. -// ===================================================================== - -// Count Sketch (a.k.a. Count-Min-style signed-counter sketch) — -// element-wise mergeable frequency estimator. -// -// Parallel to `count_min::CountMinSketch` but with **signed** counters, -// matching the `asap_sketchlib::proto::sketchlib::CountSketchState` wire -// format that DataCollector's `countsketchprocessor` emits via the -// modified OTLP `Metric.data = CountSketch{…}` variant. -// -// The surface is: construct from a decoded proto state, merge -// element-wise with another sketch, emit the matrix for queries and -// serialization. The richer query semantics of Count Sketch (median- -// of-estimators heavy-hitter tracking, `TopKState` integration) are not -// implemented here — the wire format carries the matrix losslessly, so -// the merge/store round-trip works with just a matrix. - -/// Default Top-K capacity. Mirrors sketchlib-go `TOPK_SIZE = 100`. -pub const COUNT_SKETCH_TOPK_CAPACITY: usize = 100; - -/// Sparse delta between two consecutive CountSketch snapshots — -/// the input shape for [`CountSketch::apply_delta`]. Mirrors the -/// `CountSketchDelta` proto in -/// `sketchlib-go/proto/countsketch/countsketch.proto` and the native -/// Go `Delta` in `sketchlib-go/sketches/CountSketch/delta.go`. -/// -/// Cells apply additively: `matrix[row][col] += d_count` for each -/// `(row, col, d_count)` triple. Per-row L2 norm deltas apply -/// additively. Heavy-hitter candidate keys (`hh_keys`) are queried -/// against the post-merge matrix and used to rebuild the receiver's -/// Top-K heap. -#[derive(Debug, Clone, Default)] -pub struct CountSketchDelta { - pub rows: u32, - pub cols: u32, - /// `(row, col, d_count)` cell updates, additive on the CS matrix. - pub cells: Vec<(u32, u32, i64)>, - /// Per-row L2 norm deltas. Additive, one scalar per row of the - /// base sketch. Kept on the delta surface for downstream - /// error-accounting; `apply_delta` itself ignores L2. - pub l2: Vec, - /// Heavy-hitter candidate keys forwarded by the upstream - /// Space-Saving tracker. The receiver re-queries the merged CS - /// matrix for each key and updates its Top-K heap with the - /// resulting estimate. Mirrors Go's `Delta.HHKeys`. - pub hh_keys: Vec, -} - -/// Minimal Count Sketch state — a flat `rows × cols` matrix of signed -/// counts. Element-wise mergeable (sum over aligned cells). Mirrors -/// sketchlib-go's `CountSketch.Count`/`TopK` pair (the on-the-wire -/// `L2` field is a derived value and is recomputed on load). -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct CountSketch { - #[serde(rename = "row_num")] - pub rows: usize, - #[serde(rename = "col_num")] - pub cols: usize, - /// Row-major matrix of signed counts. `matrix[r][c]` is the value of - /// hash row `r`, column `c`. - pub matrix: Vec>, - /// Top-K heavy hitters as `(key, count)` pairs, capped at - /// [`COUNT_SKETCH_TOPK_CAPACITY`]. Order is not guaranteed (heap - /// shape is not preserved on the wire). Mirrors Go's - /// `CountSketch.TopK` slot. Defaults to empty on legacy payloads. - #[serde(default)] - pub topk: Vec<(String, f64)>, -} - -impl CountSketch { - /// Construct an all-zero sketch with the given dimensions. - pub fn new(rows: usize, cols: usize) -> Self { - Self { - rows, - cols, - matrix: vec![vec![0.0; cols]; rows], - topk: Vec::new(), - } - } - - /// Construct from a pre-built matrix (used by the modified-OTLP - /// proto-decode path). `topk` is zero-initialised; callers that - /// need non-zero auxiliary state should use the msgpack/proto path. - pub fn from_legacy_matrix(matrix: Vec>, rows: usize, cols: usize) -> Self { - debug_assert_eq!(matrix.len(), rows, "row count mismatch"); - debug_assert!( - matrix.iter().all(|r| r.len() == cols), - "column count mismatch in at least one row" - ); - Self { - rows, - cols, - matrix, - topk: Vec::new(), - } - } - - /// Borrow the inner matrix. - pub fn sketch(&self) -> &Vec> { - &self.matrix - } - - /// Update the in-memory Top-K heap with `(key, count)`. Keeps the - /// heap bounded by [`COUNT_SKETCH_TOPK_CAPACITY`]; on overflow, - /// drops the smallest-count entry. If `key` is already present, - /// the new count replaces the old (max semantics). Used by - /// `apply_delta` to rebuild Top-K from `hh_keys`. - fn topk_update(&mut self, key: &str, count: f64) { - if let Some(slot) = self.topk.iter_mut().find(|(k, _)| k == key) { - if count > slot.1 { - slot.1 = count; - } - return; - } - if self.topk.len() < COUNT_SKETCH_TOPK_CAPACITY { - self.topk.push((key.to_owned(), count)); - return; - } - // Capacity hit: replace the minimum if `count` exceeds it. - if let Some((min_idx, min_count)) = self - .topk - .iter() - .enumerate() - .min_by(|a, b| { - a.1.1 - .partial_cmp(&b.1.1) - .unwrap_or(std::cmp::Ordering::Equal) - }) - .map(|(i, e)| (i, e.1)) - { - if count > min_count { - self.topk[min_idx] = (key.to_owned(), count); - } - } - } - - /// Insert a single weighted observation. Uses the shared - /// [`crate::MatrixHashType::Packed64`] fast-path hash so the matrix-cell - /// layout matches `sketchlib-go::CountSketch.UpdateString` - /// bit-for-bit. Locked in by - /// `tests/sketches_go_parity_probe.rs`. - /// - /// Constraints inherited from Go's Packed64 mode: - /// - `cols` must be a power of two; the column mask is `cols - 1`. - /// - `rows * (log2(cols) + 1) ≤ 64` so the packed hash holds all - /// row indices and sign bits. - pub fn update(&mut self, key: &str, value: f64) { - if self.rows == 0 || self.cols == 0 { - return; - } - let input = crate::common::DataInput::String(key.to_owned()); - let hashed = crate::common::MatrixHashType::Packed64( - ::hash64_seeded(0, &input), - ); - for r in 0..self.rows { - let col = ::col_for_row( - &hashed, r, self.cols, - ); - let sign = - ::sign_for_row( - &hashed, r, - ) as f64; - self.matrix[r][col] += sign * value; - } - } - - /// Estimate the frequency of `key` via the standard median-of-rows - /// CountSketch query. Returns 0 for an empty sketch. Inverts the - /// signed-counter projection that [`Self::update`] applied. - pub fn estimate(&self, key: &str) -> f64 { - if self.rows == 0 || self.cols == 0 { - return 0.0; - } - let input = crate::common::DataInput::String(key.to_owned()); - let hashed = crate::common::MatrixHashType::Packed64( - ::hash64_seeded(0, &input), - ); - let mut estimates: Vec = Vec::with_capacity(self.rows); - for r in 0..self.rows { - let col = ::col_for_row( - &hashed, r, self.cols, - ); - let sign = - ::sign_for_row( - &hashed, r, - ) as f64; - estimates.push(sign * self.matrix[r][col]); - } - estimates.sort_by(|a, b| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal)); - let mid = estimates.len() / 2; - if estimates.len() % 2 == 1 { - estimates[mid] - } else { - (estimates[mid - 1] + estimates[mid]) / 2.0 - } - } - - /// Merge one other sketch into self via element-wise addition. Both - /// operands must have identical dimensions. - pub fn merge( - &mut self, - other: &CountSketch, - ) -> Result<(), Box> { - if self.rows != other.rows || self.cols != other.cols { - return Err(format!( - "CountSketch dimension mismatch: self={}x{}, other={}x{}", - self.rows, self.cols, other.rows, other.cols - ) - .into()); - } - for r in 0..self.rows { - for c in 0..self.cols { - self.matrix[r][c] += other.matrix[r][c]; - } - } - Ok(()) - } - - /// Apply a sparse delta in place. Matches the `ApplyDelta` - /// semantics in `sketchlib-go/sketches/CountSketch/delta.go`: - /// * each `(row, col, d_count)` triple updates the count matrix - /// additively (`matrix[r][c] += d_count`); - /// * each `hh_key` is re-queried against the post-update matrix - /// and pushed into the receiver's Top-K with the merged-estimate - /// count (mirrors Go's `Delta.HHKeys` heavy-hitter rebuild). - /// - /// Returns `Err` if any `(row, col)` is out of range — indicating - /// a dimension mismatch between the snapshot this sketch was - /// built from and the delta sender. - pub fn apply_delta( - &mut self, - delta: &CountSketchDelta, - ) -> Result<(), Box> { - // 1. Cell additions. - for (row, col, d_count) in &delta.cells { - let r = *row as usize; - let c = *col as usize; - if r >= self.rows || c >= self.cols { - return Err(format!( - "CountSketchDelta cell ({r},{c}) out of range (matrix={}x{})", - self.rows, self.cols - ) - .into()); - } - // `d_count` is signed on the wire; CS counts are signed - // too (can go negative under adversarial keys). - self.matrix[r][c] += *d_count as f64; - } - // 2. Heavy-hitter rebuild from `hh_keys`. Re-estimate against - // the freshly-updated matrix and push into Top-K with the - // merged count. Mirrors sketchlib-go's `Delta.HHKeys` path. - for key in &delta.hh_keys { - let est = self.estimate(key); - self.topk_update(key, est); - } - Ok(()) - } - - /// Compute a sparse, proto-marshalled `CountSketchDelta` of `self` - /// against a `snapshot`. A cell is included when its signed count - /// delta `Δcount` (self − snapshot) is non-zero and its magnitude is - /// `>= threshold` (the threshold is truncated to an integer, matching - /// the Go reference). The full per-row L2 norm deltas are always - /// carried (one entry per row). - /// - /// This is the Rust twin of the Go reference implementation's - /// `ComputeDelta` + `SerializeDelta`: it iterates the matrix in - /// row-major order, subtracts the snapshot's value for each cell, and - /// emits the surviving signed cell deltas using the packed-array proto - /// encoding (`cell_rows`/`cell_cols`/`d_counts`). L2 row deltas are - /// derived from the matrices — `l2[r] = Σ_c count[r][c]^2` — matching - /// the Go producer's incrementally-maintained norm. Heavy-hitter - /// candidate keys (`hh_keys`) are sourced from an upstream Space-Saving - /// tracker that this minimal wrapper does not maintain, so the field is - /// left empty (an empty repeated field encodes to nothing on the wire). - /// The returned bytes are a `prost`-encoded - /// [`crate::proto::sketchlib::CountSketchDelta`], byte-identical to the - /// Go `proto.Marshal(CountSketchDelta)` output for the same inputs when - /// no heavy-hitter keys are forwarded (cross-language byte parity). - /// - /// Delta-against-empty: when `snapshot` is the all-zero sketch, every - /// surviving cell delta equals this window's own (signed) cell count, - /// so the result is this window's full state encoded as a delta (no - /// cross-window subtraction). CountSketch deltas carry only - /// sketch-internal cells/norms — there are no DataPoint-level metric - /// scalars to drop. - pub fn compute_delta( - &self, - snapshot: &CountSketch, - threshold: f64, - ) -> Result, Box> { - use crate::proto::sketchlib::CountSketchDelta as ProtoDelta; - use prost::Message; - - if self.rows != snapshot.rows || self.cols != snapshot.cols { - return Err(format!( - "CountSketch dimension mismatch: self={}x{}, snapshot={}x{}", - self.rows, self.cols, snapshot.rows, snapshot.cols - ) - .into()); - } - - // Threshold is compared against |Δ| after truncation to an - // integer, mirroring the Go reference's `int64(threshold)` cast. - let thresh = threshold as i64; - - let mut cell_rows: Vec = Vec::new(); - let mut cell_cols: Vec = Vec::new(); - let mut d_counts: Vec = Vec::new(); - let mut l2: Vec = Vec::with_capacity(self.rows); - - for r in 0..self.rows { - let mut cur_l2 = 0.0f64; - let mut snap_l2 = 0.0f64; - for c in 0..self.cols { - let cv = self.matrix[r][c]; - let sv = snapshot.matrix[r][c]; - cur_l2 += cv * cv; - snap_l2 += sv * sv; - let dc = (cv - sv) as i64; - if dc != 0 && (dc <= -thresh || dc >= thresh) { - cell_rows.push(r as u32); - cell_cols.push(c as u32); - d_counts.push(dc); - } - } - l2.push(cur_l2 - snap_l2); - } - - let delta = ProtoDelta { - rows: self.rows as u32, - cols: self.cols as u32, - cells_legacy: Vec::new(), - l2_legacy: Vec::new(), - topk: None, - hh_keys: Vec::new(), - cell_rows, - cell_cols, - d_counts, - l2, - }; - Ok(delta.encode_to_vec()) - } - - /// Apply a `prost`-encoded [`crate::proto::sketchlib::CountSketchDelta`] - /// to this sketch in place (additive signed-cell merge + heavy-hitter - /// rebuild). The Rust twin of the Go reference implementation's - /// `DeserializeDelta` + `ApplyDelta`. Reads the packed cell arrays - /// (`cell_rows`/`cell_cols`/`d_counts`) and the heavy-hitter keys - /// (`hh_keys`), falling back to the legacy per-cell records and TopK - /// entries for payloads from older producers. - /// - /// Returns `Err` if `bytes` is not a valid `CountSketchDelta` proto or - /// a cell is out of range for this sketch's dimensions. - pub fn apply_delta_bytes( - &mut self, - bytes: &[u8], - ) -> Result<(), Box> { - use crate::proto::sketchlib::CountSketchDelta as ProtoDelta; - use prost::Message; - - let proto = ProtoDelta::decode(bytes)?; - let cells: Vec<(u32, u32, i64)> = if !proto.cell_rows.is_empty() { - proto - .cell_rows - .iter() - .zip(proto.cell_cols.iter()) - .zip(proto.d_counts.iter()) - .map(|((&r, &c), &d)| (r, c, d)) - .collect() - } else { - proto - .cells_legacy - .iter() - .map(|c| (c.row, c.col, c.d_count as i64)) - .collect() - }; - let l2 = if !proto.l2.is_empty() { - proto.l2 - } else { - proto.l2_legacy - }; - let hh_keys = if !proto.hh_keys.is_empty() { - proto.hh_keys - } else if let Some(topk) = proto.topk { - topk.entries.into_iter().map(|e| e.key).collect() - } else { - Vec::new() - }; - let delta = CountSketchDelta { - rows: proto.rows, - cols: proto.cols, - cells, - l2, - hh_keys, - }; - self.apply_delta(&delta) - } - - /// MessagePack twin of [`Self::compute_delta`]. Computes the same - /// sparse signed-cell delta of `self` against `snapshot` (a cell is - /// carried when `Δcount` is non-zero and `|Δcount| >= threshold`, the - /// threshold truncated to an integer), but serializes it as the - /// cells-only parallel-array MessagePack layout via - /// [`CountSketchDelta`]'s [`MessagePackCodec`] instead of proto. The - /// per-row `l2` norms and `hh_keys` the proto delta carries are dropped - /// — the plain-sketch estimate path does not consume them. - /// Delta-against-empty carries this window's own signed cell counts. - pub fn compute_delta_msgpack( - &self, - snapshot: &CountSketch, - threshold: f64, - ) -> Result, Box> { - if self.rows != snapshot.rows || self.cols != snapshot.cols { - return Err(format!( - "CountSketch dimension mismatch: self={}x{}, snapshot={}x{}", - self.rows, self.cols, snapshot.rows, snapshot.cols - ) - .into()); - } - let thresh = threshold as i64; - let mut delta = CountSketchDelta { - rows: self.rows as u32, - cols: self.cols as u32, - ..CountSketchDelta::default() - }; - for r in 0..self.rows { - for c in 0..self.cols { - let dc = (self.matrix[r][c] - snapshot.matrix[r][c]) as i64; - if dc != 0 && (dc <= -thresh || dc >= thresh) { - delta.cells.push((r as u32, c as u32, dc)); - } - } - } - Ok(delta.to_msgpack()?) - } - - /// MessagePack twin of [`Self::apply_delta_bytes`]. Decodes the - /// cells-only parallel-array MessagePack [`CountSketchDelta`] and - /// applies it in place (additive signed-cell merge). - /// - /// Returns `Err` if `bytes` is not a valid MessagePack - /// `CountSketchDelta` or a cell is out of range for this sketch's - /// dimensions. - pub fn apply_delta_msgpack_bytes( - &mut self, - bytes: &[u8], - ) -> Result<(), Box> { - let delta = CountSketchDelta::from_msgpack(bytes)?; - self.apply_delta(&delta) - } - - /// Merge a slice of references into a single new sketch. All inputs - /// must share the same dimensions; returns `Err` on mismatch or an - /// empty input. - pub fn merge_refs( - inputs: &[&CountSketch], - ) -> Result> { - let first = inputs - .first() - .ok_or("CountSketch::merge_refs called with empty input")?; - let mut merged = CountSketch::new(first.rows, first.cols); - for cs in inputs { - merged.merge(cs)?; - } - Ok(merged) - } -} - -impl MessagePackCodec for CountSketch { - fn to_msgpack(&self) -> Result, MsgPackError> { - Ok(rmp_serde::to_vec(self)?) - } - - fn from_msgpack(bytes: &[u8]) -> Result { - Ok(rmp_serde::from_slice(bytes)?) - } -} - -impl MessagePackCodec for CountSketchDelta { - /// Encodes the sparse signed-cell delta as the 5-element MessagePack - /// array - /// `[ rows:u64, cols:u64, row_idx:[]u32, col_idx:[]u32, d_count:[]i64 ]` - /// — byte-identical to Rust - /// `rmp_serde::to_vec(&(u64, u64, Vec, Vec, Vec))` - /// (compact mode), the layout the Go `count_sketch_delta_sparse.go` - /// established (with signed `d_count` in place of the F2-broadcast's - /// `f64` vals). Only the matrix cells cross the wire; the per-row `l2` - /// norms and `hh_keys` heavy-hitter surface are not carried. - fn to_msgpack(&self) -> Result, MsgPackError> { - let mut row_idx: Vec = Vec::with_capacity(self.cells.len()); - let mut col_idx: Vec = Vec::with_capacity(self.cells.len()); - let mut d_count: Vec = Vec::with_capacity(self.cells.len()); - for (r, c, d) in &self.cells { - row_idx.push(*r); - col_idx.push(*c); - d_count.push(*d); - } - Ok(rmp_serde::to_vec(&( - self.rows as u64, - self.cols as u64, - row_idx, - col_idx, - d_count, - ))?) - } - - fn from_msgpack(bytes: &[u8]) -> Result { - let (rows, cols, row_idx, col_idx, d_count): (u64, u64, Vec, Vec, Vec) = - rmp_serde::from_slice(bytes)?; - let cells = row_idx - .into_iter() - .zip(col_idx) - .zip(d_count) - .map(|((r, c), d)| (r, c, d)) - .collect(); - Ok(CountSketchDelta { - rows: rows as u32, - cols: cols as u32, - cells, - ..CountSketchDelta::default() - }) - } -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn msgpack_delta_against_empty_round_trips() { - let (rows, cols) = (5, 1024); - let mut w = CountSketch::new(rows, cols); - for _ in 0..200 { - w.update("hot", 1.0); - } - let empty = CountSketch::new(rows, cols); - let bytes = w.compute_delta_msgpack(&empty, 1.0).unwrap(); - let mut recon = CountSketch::new(rows, cols); - recon.apply_delta_msgpack_bytes(&bytes).unwrap(); - // CountSketch round-trips its cells exactly under delta-against-empty. - assert_eq!(recon.sketch(), w.sketch()); - } - - #[test] - fn msgpack_delta_wire_layout_is_parallel_arrays() { - use crate::message_pack_format::MessagePackCodec; - let (rows, cols) = (3, 8); - let mut w = CountSketch::new(rows, cols); - w.update("x", 5.0); - let empty = CountSketch::new(rows, cols); - let bytes = w.compute_delta_msgpack(&empty, 1.0).unwrap(); - let cur = w.sketch(); - let (mut ri, mut ci, mut dc): (Vec, Vec, Vec) = - (Vec::new(), Vec::new(), Vec::new()); - for (r, row) in cur.iter().enumerate() { - for (c, &cell) in row.iter().enumerate() { - let d = cell as i64; - if d != 0 { - ri.push(r as u32); - ci.push(c as u32); - dc.push(d); - } - } - } - let expected = rmp_serde::to_vec(&(rows as u64, cols as u64, ri, ci, dc)).unwrap(); - assert_eq!(bytes, expected); - let d = CountSketchDelta::from_msgpack(&bytes).unwrap(); - assert_eq!(d.rows, rows as u32); - assert_eq!(d.cols, cols as u32); - } - - #[test] - fn test_new_empty() { - let cs = CountSketch::new(2, 3); - assert_eq!(cs.rows, 2); - assert_eq!(cs.cols, 3); - assert_eq!(cs.sketch(), &vec![vec![0.0, 0.0, 0.0], vec![0.0, 0.0, 0.0]]); - } - - #[test] - fn test_from_legacy_matrix() { - let m = vec![vec![1.0, -2.0, 3.0], vec![-4.0, 5.0, -6.0]]; - let cs = CountSketch::from_legacy_matrix(m.clone(), 2, 3); - assert_eq!(cs.sketch(), &m); - } - - #[test] - fn test_merge_element_wise() { - let mut a = CountSketch::from_legacy_matrix(vec![vec![1.0, 2.0], vec![3.0, 4.0]], 2, 2); - let b = CountSketch::from_legacy_matrix(vec![vec![-1.0, -2.0], vec![-3.0, -4.0]], 2, 2); - a.merge(&b).unwrap(); - assert_eq!(a.sketch(), &vec![vec![0.0, 0.0], vec![0.0, 0.0]]); - } - - #[test] - fn test_merge_dimension_mismatch() { - let mut a = CountSketch::new(2, 3); - let b = CountSketch::new(3, 3); - assert!(a.merge(&b).is_err()); - } - - #[test] - fn test_merge_refs() { - let a = CountSketch::from_legacy_matrix(vec![vec![1.0, 2.0]], 1, 2); - let b = CountSketch::from_legacy_matrix(vec![vec![3.0, 4.0]], 1, 2); - let c = CountSketch::from_legacy_matrix(vec![vec![5.0, 6.0]], 1, 2); - let merged = CountSketch::merge_refs(&[&a, &b, &c]).unwrap(); - assert_eq!(merged.sketch(), &vec![vec![9.0, 12.0]]); - } - - #[test] - fn test_apply_delta_additive() { - let mut cs = CountSketch::from_legacy_matrix( - vec![vec![1.0, -2.0, 3.0], vec![-4.0, 5.0, -6.0]], - 2, - 3, - ); - let delta = CountSketchDelta { - rows: 2, - cols: 3, - cells: vec![ - (0, 0, 10), // 1 + 10 = 11 - (0, 2, -3), // 3 - 3 = 0 - (1, 1, -15), // 5 - 15 = -10 - ], - l2: vec![], - hh_keys: vec![], - }; - cs.apply_delta(&delta).unwrap(); - assert_eq!( - cs.sketch(), - &vec![vec![11.0, -2.0, 0.0], vec![-4.0, -10.0, -6.0]] - ); - } - - #[test] - fn test_apply_delta_matches_full_merge() { - let base = CountSketch::from_legacy_matrix(vec![vec![1.0, 2.0], vec![3.0, 4.0]], 2, 2); - let addition = - CountSketch::from_legacy_matrix(vec![vec![10.0, 0.0], vec![0.0, 20.0]], 2, 2); - let mut via_merge = base.clone(); - via_merge.merge(&addition).unwrap(); - - let delta = CountSketchDelta { - rows: 2, - cols: 2, - cells: vec![(0, 0, 10), (1, 1, 20)], - l2: vec![], - hh_keys: vec![], - }; - let mut via_delta = base; - via_delta.apply_delta(&delta).unwrap(); - assert_eq!(via_delta.sketch(), via_merge.sketch()); - } - - #[test] - fn test_apply_delta_out_of_range() { - let mut cs = CountSketch::new(2, 3); - let delta = CountSketchDelta { - rows: 2, - cols: 3, - cells: vec![(2, 0, 1)], // row 2 out of range for 2-row matrix - l2: vec![], - hh_keys: vec![], - }; - assert!(cs.apply_delta(&delta).is_err()); - } - - #[test] - fn test_apply_delta_rebuilds_topk_from_hh_keys() { - // Build a sketch with two known keys via the in-process - // `update` path so the matrix has a coherent shape, then - // send a delta that only carries `hh_keys` entries. The - // receiver should re-query the merged matrix and populate - // `topk` with the resulting estimates. Mirrors sketchlib-go's - // `Delta.HHKeys` heavy-hitter rebuild path. - let mut cs = CountSketch::new(3, 16); - cs.update("alpha", 5.0); - cs.update("beta", 3.0); - let delta = CountSketchDelta { - rows: 3, - cols: 16, - cells: vec![], - l2: vec![], - hh_keys: vec!["alpha".to_string(), "beta".to_string()], - }; - cs.apply_delta(&delta).unwrap(); - assert_eq!(cs.topk.len(), 2); - let alpha_count = cs - .topk - .iter() - .find(|(k, _)| k == "alpha") - .map(|(_, v)| *v) - .expect("alpha should be in topk"); - let beta_count = cs - .topk - .iter() - .find(|(k, _)| k == "beta") - .map(|(_, v)| *v) - .expect("beta should be in topk"); - // Alpha was inserted with weight 5; the median estimate - // should exceed beta's (weight 3) modulo signed-counter - // cancellation in this small 3x16 matrix. - assert!( - alpha_count > beta_count, - "alpha={alpha_count} beta={beta_count}" - ); - } - - #[test] - fn test_apply_delta_hh_keys_topk_capacity() { - // Verify the Top-K heap is bounded by COUNT_SKETCH_TOPK_CAPACITY - // and that on overflow, the smallest-count entry is evicted in - // favor of a larger-count newcomer. - let mut cs = CountSketch::new(3, 1024); - let n = COUNT_SKETCH_TOPK_CAPACITY + 5; - let keys: Vec = (0..n).map(|i| format!("k{i:04}")).collect(); - // Fill all keys into the matrix so estimates are non-zero. - for (i, k) in keys.iter().enumerate() { - cs.update(k, (i + 1) as f64); - } - let delta = CountSketchDelta { - rows: 3, - cols: 1024, - cells: vec![], - l2: vec![], - hh_keys: keys.clone(), - }; - cs.apply_delta(&delta).unwrap(); - assert_eq!(cs.topk.len(), COUNT_SKETCH_TOPK_CAPACITY); - } - - #[test] - fn test_msgpack_round_trip() { - let original = - CountSketch::from_legacy_matrix(vec![vec![1.5, -2.5], vec![3.5, -4.5]], 2, 2); - let bytes = original.to_msgpack().unwrap(); - let decoded = CountSketch::from_msgpack(&bytes).unwrap(); - assert_eq!(decoded.sketch(), original.sketch()); - assert_eq!(decoded.rows, original.rows); - assert_eq!(decoded.cols, original.cols); - } - - /// Cross-language byte-parity guard against `sketchlib-go`'s - /// `CountSketch.SerializeProtoBytes` output for the deterministic - /// input `goldenCsKeys()` (25 keys "k-a"..."k-e", each repeated 5×) - /// at dimensions `(rows=3, cols=512)`. The hex blob below was - /// captured from a `proto.Marshal` of the Go envelope with - /// `Producer` and `HashSpec` cleared (matching the - /// `integration/parity/golden_test.go::TestGenerateGoldenFixtures` - /// recipe used for DDSketch and KLL). - /// - /// Any drift in [`CountSketch::update`]'s hash → (col, sign) - /// derivation breaks this test cell-for-cell; that is the contract - /// `cross_language_parity::countsketch_byte_parity_with_go` in - /// ASAPCollector relies on. - #[test] - fn test_update_then_envelope_matches_sketchlib_go_bytes() { - use crate::proto::sketchlib::{ - CountSketchState, CounterType, SketchEnvelope, sketch_envelope::SketchState, - }; - use prost::Message; - - let rows = 3usize; - let cols = 512usize; - let mut sk = CountSketch::new(rows, cols); - for i in 0..25 { - let key = format!("k-{}", (b'a' + (i % 5) as u8) as char); - sk.update(&key, 1.0); - } - - // Build envelope mirroring sketchlib-go's CountSketch.SerializePortable: - // counter_type = INT64, counts packed sint64 in row-major - // order, l2 = per-row sum of squared cells, no TopK. - let mut counts_int: Vec = Vec::with_capacity(rows * cols); - let mut l2: Vec = Vec::with_capacity(rows); - for row in sk.matrix.iter().take(rows) { - let mut row_l2 = 0.0f64; - for &cell in row.iter().take(cols) { - counts_int.push(cell as i64); - row_l2 += cell * cell; - } - l2.push(row_l2); - } - - let state = CountSketchState { - rows: rows as u32, - cols: cols as u32, - counter_type: CounterType::Int64 as i32, - counts_int, - counts_float: Vec::new(), - l2, - topk: None, - }; - let envelope = SketchEnvelope { - format_version: 1, - producer: None, - hash_spec: None, - sample_p: 0.0, - sketch_state: Some(SketchState::CountSketch(state)), - }; - let mut got = Vec::with_capacity(envelope.encoded_len()); - envelope.encode(&mut got).expect("prost encode"); - - // Byte string captured from sketchlib-go for the same - // (3,512) × `goldenCsKeys()` input — see - // `integration/parity/golden_test.go` and - // `cross_language_parity::countsketch_byte_parity_with_go` - // in ASAPCollector. 1577 bytes total: a `SketchEnvelope` proto - // wrapping a `CountSketchState` whose `counts_int` is a packed - // sint64 (zigzag) row-major matrix and `l2` is `[125.0;3]` - // (each row's 5 hot cells hold ±5 → l2 = 5·25 = 125). - const GOLDEN_HEX: &str = "08015aa40c0803108004180222800c000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000900000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000a0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000090000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000009000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000900000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000a000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000900000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000a0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000090000000000000a000000000000000900000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000900000000000000000000000000000000000000000000000000000000000000000a0000000000000000000000090000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000009000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000032180000000000405f400000000000405f400000000000405f40"; - let want = decode_hex(GOLDEN_HEX); - assert_eq!( - got, - want, - "CountSketch envelope bytes diverge from sketchlib-go golden \ - ({} bytes got vs {} bytes want)", - got.len(), - want.len(), - ); - } - - /// Cross-language byte-parity guard for CountSketchDelta hh_keys: encoding - /// the canonical packed delta with heavy-hitter keys at tag 6 must be - /// byte-identical to the Go reference's `SerializeDelta` output for the same - /// state and keys. The empty-`hh_keys` path is already covered by - /// `test_compute_delta_matches_go_golden_bytes`. - #[test] - fn test_hh_keys_matches_go_golden_bytes() { - use crate::proto::sketchlib::CountSketchDelta as ProtoDelta; - use prost::Message; - - let rows = 3usize; - let cols = 512usize; - let mut current = CountSketch::new(rows, cols); - for i in 0..25 { - let key = format!("k-{}", (b'a' + (i % 5) as u8) as char); - current.update(&key, 1.0); - } - let empty = CountSketch::new(rows, cols); - - // Inject heavy-hitter keys onto the canonical packed delta at tag 6. - let no_hh = current.compute_delta(&empty, 1.0).unwrap(); - let mut proto = ProtoDelta::decode(no_hh.as_slice()).unwrap(); - proto.hh_keys = vec!["flow-0".into(), "flow-3".into(), "flow-7".into()]; - let got = proto.encode_to_vec(); - - // Captured from the Go reference implementation's SerializeDelta for the - // same input with HHKeys = ["flow-0","flow-3","flow-7"]. - const GOLDEN_HEX: &str = "08031080043206666c6f772d303206666c6f772d333206666c6f772d374a0f000000000001010101010202020202521a3d7fd9019402d3034eb9019602f703fe0306a501c601d201ba025a0f090a0909090a090a090a09090a090962180000000000405f400000000000405f400000000000405f40"; - let want = decode_hex(GOLDEN_HEX); - assert_eq!( - got, - want, - "CountSketch delta hh_keys bytes diverge from the Go reference golden \ - ({} bytes got vs {} bytes want)", - got.len(), - want.len(), - ); - } - - fn decode_hex(s: &str) -> Vec { - s.as_bytes() - .chunks(2) - .map(|pair| { - let high = hex_nibble(pair[0]); - let low = hex_nibble(pair[1]); - (high << 4) | low - }) - .collect() - } - - fn hex_nibble(c: u8) -> u8 { - match c { - b'0'..=b'9' => c - b'0', - b'a'..=b'f' => c - b'a' + 10, - b'A'..=b'F' => c - b'A' + 10, - _ => panic!("non-hex byte {}", c as char), - } - } - - /// `compute_delta` against an EMPTY snapshot reconstructs the window's - /// full (signed) state when its bytes are applied to a fresh empty - /// sketch (round-trip). With `threshold = 1.0` every changed cell - /// survives. - #[test] - fn test_compute_delta_against_empty_round_trips() { - let rows = 3usize; - let cols = 512usize; - let mut window = CountSketch::new(rows, cols); - for i in 0..200u64 { - window.update(&format!("k-{}", i % 23), 1.0); - } - let empty = CountSketch::new(rows, cols); - - let delta_bytes = window.compute_delta(&empty, 1.0).unwrap(); - - let mut reconstructed = CountSketch::new(rows, cols); - reconstructed.apply_delta_bytes(&delta_bytes).unwrap(); - - assert_eq!(reconstructed.sketch(), window.sketch()); - } - - /// A delta computed between two non-empty snapshots reconstructs the - /// current sketch when applied to the base. - #[test] - fn test_compute_delta_then_apply_matches_current() { - let mut base = CountSketch::new(3, 64); - for i in 0..40u64 { - base.update(&format!("x{}", i % 8), 1.0); - } - let mut current = base.clone(); - for i in 0..30u64 { - current.update(&format!("x{}", i % 8), 1.0); - } - - let delta_bytes = current.compute_delta(&base, 1.0).unwrap(); - let mut reconstructed = base.clone(); - reconstructed.apply_delta_bytes(&delta_bytes).unwrap(); - assert_eq!(reconstructed.sketch(), current.sketch()); - } - - /// Cross-language byte-parity guard: `compute_delta` against an empty - /// snapshot must emit bytes identical to the Go reference - /// implementation's `SerializeDelta(ComputeDelta(empty, current, 1.0))` - /// for the same `(rows=3, cols=512)` × "k-a".."k-e" (each 5×, 25 - /// unweighted updates) input. The golden hex was captured from a - /// `proto.Marshal` of the Go reference's `CountSketchDelta` - /// (packed-array encoding). No heavy-hitter keys are forwarded by this - /// minimal wrapper, so `hh_keys` is empty in both producers; the golden - /// pins the packed cell arrays and the per-row L2 (=125) norm deltas. - #[test] - fn test_compute_delta_matches_go_golden_bytes() { - let rows = 3usize; - let cols = 512usize; - let mut current = CountSketch::new(rows, cols); - for i in 0..25 { - let key = format!("k-{}", (b'a' + (i % 5) as u8) as char); - current.update(&key, 1.0); - } - let empty = CountSketch::new(rows, cols); - let got = current.compute_delta(&empty, 1.0).unwrap(); - - // Captured from the Go reference implementation's - // SerializeDelta(ComputeDelta(empty, current, 1.0)) for the same input. - const GOLDEN_HEX: &str = "08031080044a0f000000000001010101010202020202521a3d7fd9019402d3034eb9019602f703fe0306a501c601d201ba025a0f090a0909090a090a090a09090a090962180000000000405f400000000000405f400000000000405f40"; - let want = decode_hex(GOLDEN_HEX); - assert_eq!( - got, - want, - "CountSketch delta bytes diverge from the Go reference golden \ - ({} bytes got vs {} bytes want)", - got.len(), - want.len(), - ); - } -} diff --git a/src/message_pack_format/portable/mod.rs b/src/message_pack_format/portable/mod.rs index 97e0d372..7b0dfc7d 100644 --- a/src/message_pack_format/portable/mod.rs +++ b/src/message_pack_format/portable/mod.rs @@ -2,14 +2,13 @@ //! //! Each per-algorithm submodule holds one sketch's pre-envelope wire type. //! These carry a cross-language contract of their own, on goldens apart -//! from ASAPv1's `asapv1_golden/`: HLL, Count-Min, Count Sketch and KLL pin +//! from ASAPv1's `asapv1_golden/`: HLL, Count-Min and KLL pin //! byte parity against Go's serializers, KLL in both directions; DDSketch's //! golden is a Rust self-pin, not yet reconciled. New code uses ASAPv1: //! prefer a sketch's own `serialize_to_bytes`; do not add callers here. pub mod countminsketch; pub mod countminsketch_topk; -pub mod countsketch; pub mod countsketch_topk; pub mod ddsketch; pub mod delta_set_aggregator; diff --git a/src/proto/generated/sketchlib.v1.rs b/src/proto/generated/sketchlib.v1.rs index 2c098d3f..f2b55c37 100644 --- a/src/proto/generated/sketchlib.v1.rs +++ b/src/proto/generated/sketchlib.v1.rs @@ -213,10 +213,9 @@ pub struct CountMinState { /// byte-identical delta frames for identical window state (cross-language byte /// parity). /// -/// CountMinDelta is structurally identical to CountSketchDelta: rows, cols, the -/// packed cell encoding, per-row norm deltas, and an optional repeated hh_keys -/// at the SAME tag number (6). Both sketches can track heavy hitters; whether -/// hh_keys is populated is a control-plane decision (it is empty/omitted when +/// CountMinDelta carries rows, cols, the packed cell encoding, per-row norm +/// deltas, and an optional repeated hh_keys (tag 6). Whether hh_keys is +/// populated is a control-plane decision (it is empty/omitted when /// heavy-hitter tracking is not enabled for this sketch). #[derive(Clone, PartialEq, ::prost::Message)] pub struct CountMinDelta { @@ -234,8 +233,7 @@ pub struct CountMinDelta { /// Per-row L2 norm deltas, length = rows. #[prost(double, repeated, tag = "5")] pub l2: ::prost::alloc::vec::Vec, - /// Heavy-hitter candidate keys from an upstream tracker, mirroring - /// CountSketchDelta.hh_keys (same tag number 6, same wire shape). Downstream + /// Heavy-hitter candidate keys from an upstream tracker. Downstream /// queries the merged Count-Min matrix for each key to (re)build its Top-K /// with globally-merged estimates. Empty/omitted when heavy-hitter tracking /// is not enabled. @@ -328,70 +326,6 @@ pub struct HeapEntry { #[prost(double, tag = "2")] pub count: f64, } -/// CountSketchDelta carries only the matrix cells that changed by at least the -/// threshold between two consecutive snapshots, plus the full per-row L2 norm -/// deltas and any heavy-hitter candidate keys. -/// -/// Cells apply additively on the receiver: matrix[row][col] += d_count. Count -/// Sketch counters are signed (±1 increments), so d_count is a signed integer -/// (sint64). Heavy-hitter candidate keys (hh_keys) are re-queried against the -/// merged matrix downstream to rebuild the receiver's Top-K with accurate, -/// globally-merged estimates. -/// -/// The packed cell encoding (cell_rows / cell_cols / d_counts, tags 9-11) plus -/// l2 (tag 12) is the canonical form; the legacy per-cell message (tag 3), -/// l2_legacy (tag 4) and topk (tag 5) are retained only so payloads from older -/// producers still decode. This message is byte-identical to the Go reference -/// implementation's CountSketchDelta so the two runtimes emit byte-identical -/// delta frames for identical window state (cross-language byte parity). -#[derive(Clone, PartialEq, ::prost::Message)] -pub struct CountSketchDelta { - #[prost(uint32, tag = "1")] - pub rows: u32, - #[prost(uint32, tag = "2")] - pub cols: u32, - /// Deprecated: use cell_rows/cell_cols/d_counts (tags 9-11). Retained for - /// backward-compatible decode of payloads from older producers. - #[prost(message, repeated, tag = "3")] - pub cells_legacy: ::prost::alloc::vec::Vec, - /// Deprecated: use l2 (tag 12) instead. - #[prost(double, repeated, tag = "4")] - pub l2_legacy: ::prost::alloc::vec::Vec, - /// Deprecated: use hh_keys (tag 6) instead. - #[prost(message, optional, tag = "5")] - pub topk: ::core::option::Option, - /// Heavy-hitter candidate keys from the upstream tracker. Downstream queries - /// the merged Count Sketch matrix for each key to build its Top-K. - #[prost(string, repeated, tag = "6")] - pub hh_keys: ::prost::alloc::vec::Vec<::prost::alloc::string::String>, - /// Packed cell encoding (canonical form): - /// - /// row index of each changed cell - #[prost(uint32, repeated, tag = "9")] - pub cell_rows: ::prost::alloc::vec::Vec, - /// col index of each changed cell - #[prost(uint32, repeated, tag = "10")] - pub cell_cols: ::prost::alloc::vec::Vec, - /// signed integer count delta - #[prost(sint64, repeated, tag = "11")] - pub d_counts: ::prost::alloc::vec::Vec, - /// per-row L2 norm deltas - #[prost(double, repeated, tag = "12")] - pub l2: ::prost::alloc::vec::Vec, -} -/// CountSketchCell is the deprecated per-cell delta record. Producers emit the -/// packed cell arrays on CountSketchDelta instead; this message exists only to -/// decode payloads from older producers. -#[derive(Clone, Copy, PartialEq, ::prost::Message)] -pub struct CountSketchCell { - #[prost(uint32, tag = "1")] - pub row: u32, - #[prost(uint32, tag = "2")] - pub col: u32, - /// deprecated: use CountSketchDelta.d_counts - #[prost(double, tag = "3")] - pub d_count: f64, -} /// HyperLogLogState is the portable state of a HyperLogLog cardinality sketch. /// /// Precision p determines the register count: num_registers = 2^precision. @@ -892,7 +826,7 @@ pub struct SketchEnvelope { /// The sketch payload. Exactly one field must be set. #[prost( oneof = "sketch_envelope::SketchState", - tags = "10, 11, 12, 13, 14, 15, 16, 17, 18" + tags = "10, 12, 13, 14, 15, 16, 17, 18" )] pub sketch_state: ::core::option::Option, } @@ -903,8 +837,6 @@ pub mod sketch_envelope { pub enum SketchState { #[prost(message, tag = "10")] CountMin(super::CountMinState), - #[prost(message, tag = "11")] - CountSketch(super::CountSketchState), #[prost(message, tag = "12")] Hll(super::HyperLogLogState), #[prost(message, tag = "13")] diff --git a/tests/PBT_COVERAGE.md b/tests/PBT_COVERAGE.md index cddee44f..07b7741a 100644 --- a/tests/PBT_COVERAGE.md +++ b/tests/PBT_COVERAGE.md @@ -97,12 +97,12 @@ Compared against: a key's `(col, sign)` per row, read off a one-key sketch; the - linearity - insert `w` then `-w` for every `w`: cells back to start; signed stream over a background - `insert(k, -1)` undoes `insert(k)`, cell for cell - - `cells(S(xs)) + cells(S(ys)) == cells(S(xs ++ ys))`; integral weights, power-of-two widths + - `cells(S(xs)) + cells(S(ys)) == cells(S(xs ++ ys))`; signed weights - negate every weight: every cell negated - merge - - portable `CountSketch`: `merge(S(xs), S(ys)) == S(xs ++ ys)` to `REL_TOL` + - `cells(merge(S(xs), S(ys))) == cells(S(xs ++ ys))`; signed weights - wire - - msgpack keeps cells and estimates + - ASAPv1 keeps every cell and estimate; signed weights - ASAPv1 keeps 64 estimates; `edge_rows` x `edge_dimension` ## ddsketch.rs — 14 diff --git a/tests/cs_envelope_golden.hex b/tests/cs_envelope_golden.hex deleted file mode 100644 index e1c665c2..00000000 --- a/tests/cs_envelope_golden.hex +++ /dev/null @@ -1 +0,0 @@ -08015aa40c0803108004180222800c000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000900000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000a0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000090000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000009000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000900000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000a000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000900000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000a0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000090000000000000a000000000000000900000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000900000000000000000000000000000000000000000000000000000000000000000a0000000000000000000000090000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000009000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000032180000000000405f400000000000405f400000000000405f40 diff --git a/tests/e2e/frequency.rs b/tests/e2e/frequency.rs index a913f20f..279fc928 100644 --- a/tests/e2e/frequency.rs +++ b/tests/e2e/frequency.rs @@ -25,7 +25,6 @@ use common::{FreqTruth, uniform_u64, zipf_u64}; use std::collections::HashMap; use asap_sketchlib::message_pack_format::portable::countminsketch::CountMinSketch; -use asap_sketchlib::message_pack_format::portable::countsketch::CountSketch; use asap_sketchlib::{ Count, CountL2HH, CountMin, DataInput, DefaultXxHasher, FastPath, FoldCMS, FoldCS, RegularPath, Vector2D, @@ -693,30 +692,26 @@ fn folded_sketches_keep_their_own_bounds_through_a_sixteen_way_merge() { ); } -// ------------------------------------------------------ Portable wire twins +// ------------------------------------------------------ Portable wire twin -/// The portable wire twins answer over string keys and must satisfy the same -/// two theorems as their core counterparts — Count-Min additive, Count Sketch -/// L2 — with no extra slack for being on the wire side. +/// The portable Count-Min answers over string keys and must satisfy the same +/// additive theorem as its core counterpart, with no extra slack for being on +/// the wire side. #[test] -fn portable_cms_and_cs_string_keys_satisfy_their_own_bounds() { +fn portable_cms_string_keys_satisfy_their_own_bound() { const CM_ROWS: usize = 3; - const CS_ROWS: usize = 5; const COLS: usize = 4096; const STREAM_SEED: u64 = 1006; let stream = zipf_u64(50_000, 2048, 1.1, STREAM_SEED); - // The portable types key on strings; `FreqTruth` keys on i64. The stream's + // The portable type keys on strings; `FreqTruth` keys on i64. The stream's // u64 values are the identity behind both, so `k -> "k{k}"` is injective // and the two truths agree key for key. let mut truth = FreqTruth::default(); let mut pcs = CountMinSketch::new(CM_ROWS, COLS); - let mut pcss = CountSketch::new(CS_ROWS, COLS); for k in &stream { truth.observe(*k as i64); - let key = format!("k{k}"); - pcs.update(&key, 1.0); - pcss.update(&key, 1.0); + pcs.update(&format!("k{k}"), 1.0); } let context = format!("zipf(1.1) domain=2048 n=50000 string keys, stream_seed={STREAM_SEED}"); @@ -726,12 +721,6 @@ fn portable_cms_and_cs_string_keys_satisfy_their_own_bounds() { |k| pcs.estimate(&format!("k{k}")), &context, ); - CountSketchSpec::new(CS_ROWS, COLS).assert_contract( - "portable CountSketch", - &truth, - |k| pcss.estimate(&format!("k{k}")), - &context, - ); } // --------------------------------------------------------------------------- diff --git a/tests/msgpack_compat.rs b/tests/msgpack_compat.rs index 267a51c6..387230fc 100644 --- a/tests/msgpack_compat.rs +++ b/tests/msgpack_compat.rs @@ -31,8 +31,8 @@ use asap_sketchlib::message_pack_format::portable::delta_set_aggregator::DeltaRe use asap_sketchlib::message_pack_format::portable::hydra_kll::HydraKllSketchWire; use asap_sketchlib::message_pack_format::portable::kll::KllSketchData; use asap_sketchlib::{ - CountMinSketch, CountMinSketchWithHeap, CountSketch, DdSketch, HllSketch, HllVariant, - HydraKllSketch, KllSketch, SetAggregator, + CountMinSketch, CountMinSketchWithHeap, DdSketch, HllSketch, HllVariant, HydraKllSketch, + KllSketch, SetAggregator, }; // ===== round-trip: every wire-format-aligned type ===== @@ -61,17 +61,6 @@ fn count_min_sketch_with_heap_round_trip() { assert_eq!(restored.heap_size, 8); } -#[test] -fn count_sketch_round_trip() { - let mut s = CountSketch::new(3, 64); - s.update("k1", 1.0); - s.update("k2", 2.0); - let bytes = s.to_msgpack().expect("encode"); - let restored = CountSketch::from_msgpack(&bytes).expect("decode"); - assert_eq!(restored.rows, 3); - assert_eq!(restored.cols, 64); -} - #[test] fn dd_sketch_round_trip() { let mut s = DdSketch::new(0.01); diff --git a/tests/pbt/count_sketch.rs b/tests/pbt/count_sketch.rs index 54710227..d3f82b01 100644 --- a/tests/pbt/count_sketch.rs +++ b/tests/pbt/count_sketch.rs @@ -1,13 +1,9 @@ -//! Property tests for Count Sketch: the native matrix sketch and the portable -//! `CountSketch` that carries it over the wire. +//! Property tests for Count Sketch. //! //! Charikar, Chen, Farach-Colton, ICALP '02. -use crate::support::{ - PROBE_KEYS, edge_dimension, edge_rows, grid, keyed_updates, keys, matrices_close, -}; -use asap_sketchlib::message_pack_format::MessagePackCodec; -use asap_sketchlib::{Count, CountSketch, DataInput, RegularPath, Vector2D}; +use crate::support::{PROBE_KEYS, edge_dimension, edge_rows, grid, keys}; +use asap_sketchlib::{Count, DataInput, RegularPath, Vector2D}; use proptest::prelude::*; type Cs = Count, RegularPath>; @@ -28,30 +24,25 @@ fn weighted(max: usize) -> impl Strategy> { prop::collection::vec((0u64..512, 1i32..8), 0..max) } -fn cs_of(rows: usize, cols: usize, updates: &[(String, f64)]) -> CountSketch { - let mut s = CountSketch::new(rows, cols); - for (k, v) in updates { - s.update(k, *v); +fn cs_of(rows: usize, cols: usize, updates: &[(u64, i32)]) -> Cs { + let mut s = Cs::with_dimensions(rows, cols); + for (k, w) in updates { + s.insert_many(&DataInput::U64(*k), *w); } s } +fn wire_cells(sketch: &CsWire) -> Vec { + grid(sketch.rows(), sketch.cols(), |r, c| { + sketch.as_storage().query_one_counter(r, c) + }) +} + /// Signed weights small enough that no `i32` cell can overflow. fn signed_weights(max: usize) -> impl Strategy> { prop::collection::vec((0u64..512, -1_000i32..1_000), 0..max) } -/// Integral weights, so every portable cell stays exact in `f64`. -fn integral_updates(max: usize) -> impl Strategy> { - prop::collection::vec(("[a-z]{1,4}", -1_000i32..1_000), 0..max) - .prop_map(|v| v.into_iter().map(|(k, w)| (k, f64::from(w))).collect()) -} - -/// Power-of-two widths the packed-64 hash carries at these row counts. -fn packed_cols() -> impl Strategy { - prop_oneof![Just(2usize), Just(4), Just(8), Just(16), Just(32)] -} - /// The cell and sign a key holds in each row, read off a sketch holding that /// key alone. fn signature(rows: usize, cols: usize, key: u64) -> Vec<(usize, i32)> { @@ -133,22 +124,22 @@ proptest! { prop_assert_eq!(cs_cells(&by_weight), cs_cells(&by_repeat)); } - // ===== Merge algebra, on the portable representation ===== + // ===== Merge algebra ===== #[test] fn count_sketch_merge_matches_streaming_the_concatenation( rows in 1usize..6, - cols in 4usize..64, - a in keyed_updates(30), - b in keyed_updates(30), + cols in 1usize..64, + a in signed_weights(30), + b in signed_weights(30), ) { let mut merged = cs_of(rows, cols, &a); - merged.merge(&cs_of(rows, cols, &b)).expect("merge"); + merged.merge(&cs_of(rows, cols, &b)); let concatenated: Vec<_> = a.iter().chain(b.iter()).cloned().collect(); let streamed = cs_of(rows, cols, &concatenated); - prop_assert!(matrices_close(merged.sketch(), streamed.sketch())); + prop_assert_eq!(cs_cells(&merged), cs_cells(&streamed)); } // ===== Wire ===== @@ -157,16 +148,22 @@ proptest! { fn count_sketch_round_trip_preserves_every_cell( rows in 1usize..8, cols in 1usize..128, - updates in keyed_updates(40), + updates in signed_weights(40), ) { - let s = cs_of(rows, cols, &updates); + let mut s = CsWire::with_dimensions(rows, cols); + for (k, w) in &updates { + s.insert_many(&DataInput::U64(*k), i64::from(*w)); + } - let bytes = s.to_msgpack().expect("encode"); - let restored = CountSketch::from_msgpack(&bytes).expect("decode"); + let bytes = s.serialize_to_bytes().expect("encode"); + let restored = CsWire::deserialize_from_bytes(&bytes).expect("decode"); - prop_assert_eq!(restored.sketch(), s.sketch()); + prop_assert_eq!(wire_cells(&restored), wire_cells(&s)); for (k, _) in &updates { - prop_assert_eq!(restored.estimate(k), s.estimate(k), "key {}", k); + prop_assert_eq!( + restored.estimate(&DataInput::U64(*k)), s.estimate(&DataInput::U64(*k)), + "key {}", k + ); } } @@ -337,42 +334,38 @@ proptest! { #[test] fn cells_of_two_streams_add_to_the_cells_of_their_concatenation( rows in 1usize..6, - cols in packed_cols(), - a in integral_updates(30), - b in integral_updates(30), + cols in 1usize..64, + a in signed_weights(30), + b in signed_weights(30), ) { - let left = cs_of(rows, cols, &a); - let right = cs_of(rows, cols, &b); + let left = cs_cells(&cs_of(rows, cols, &a)); + let right = cs_cells(&cs_of(rows, cols, &b)); let concatenated: Vec<_> = a.iter().chain(b.iter()).cloned().collect(); - let both = cs_of(rows, cols, &concatenated); - - for r in 0..rows { - for c in 0..cols { - prop_assert_eq!( - left.sketch()[r][c] + right.sketch()[r][c], both.sketch()[r][c], - "cell ({}, {})", r, c - ); - } + let both = cs_cells(&cs_of(rows, cols, &concatenated)); + + for (i, cell) in both.iter().enumerate() { + prop_assert_eq!( + left[i] + right[i], *cell, + "cell ({}, {})", i / cols, i % cols + ); } } #[test] fn negating_every_weight_negates_every_cell( rows in 1usize..6, - cols in packed_cols(), - updates in integral_updates(40), + cols in 1usize..64, + updates in signed_weights(40), ) { - let forward = cs_of(rows, cols, &updates); - let negated: Vec<_> = updates.iter().map(|(k, v)| (k.clone(), -v)).collect(); - let backward = cs_of(rows, cols, &negated); - - for r in 0..rows { - for c in 0..cols { - prop_assert_eq!( - backward.sketch()[r][c], -forward.sketch()[r][c], - "cell ({}, {})", r, c - ); - } + let forward = cs_cells(&cs_of(rows, cols, &updates)); + let negated: Vec<_> = updates.iter().map(|(k, w)| (*k, -*w)).collect(); + let backward = cs_cells(&cs_of(rows, cols, &negated)); + + for (i, cell) in backward.iter().enumerate() { + prop_assert_eq!( + *cell, -forward[i], + "cell ({}, {})", i / cols, i % cols + ); } } } diff --git a/tests/sketches_go_parity_probe.rs b/tests/sketches_go_parity_probe.rs index 2997e486..cfed6faf 100644 --- a/tests/sketches_go_parity_probe.rs +++ b/tests/sketches_go_parity_probe.rs @@ -4,7 +4,7 @@ //! //! Approach: reuse the exact same Go-golden envelopes that the //! `message_pack_format::portable` parity tests use, but build the -//! matrix via `sketches::Count` / `sketches::CountMin` with `FastPath` +//! matrix via `sketches::CountMin` with `FastPath` //! and `DefaultXxHasher`. Identical bytes confirm the shared FastPath //! math is sufficient — no hashspec bypass needed. @@ -12,22 +12,15 @@ use asap_sketchlib::common::DataInput; use asap_sketchlib::common::hash::CANONICAL_HASH_SEED; use asap_sketchlib::common::structures::HllBucketListP14; use asap_sketchlib::proto::sketchlib::{ - CountMinState, CountSketchState, CounterType, DdSketchState, HyperLogLogState, SketchEnvelope, + CountMinState, CounterType, DdSketchState, HyperLogLogState, SketchEnvelope, sketch_envelope::SketchState, }; use asap_sketchlib::sketches::countminsketch::CountMin; -use asap_sketchlib::sketches::countsketch::Count; use asap_sketchlib::sketches::ddsketch::DDSketch; use asap_sketchlib::sketches::hll::{ErtlMLE, HyperLogLogImpl}; use asap_sketchlib::{DefaultXxHasher, FastPath, Vector2D}; use prost::Message; -/// 1577-byte envelope captured from `sketchlib-go::CountSketch.SerializeProtoBytes` -/// for `(rows=3, cols=512)` × `goldenCsKeys()` (25 keys "k-a".."k-e", each -/// repeated 5×). Identical to the constant in -/// `src/message_pack_format/portable/countsketch.rs::test_update_then_envelope_matches_sketchlib_go_bytes`. -const COUNTSKETCH_GOLDEN_HEX: &str = include_str!("cs_envelope_golden.hex"); - /// 8275-byte CMS Frequency-Only golden, captured from /// `sketchlib-go::CountMinSketch.SerializeProtoBytesFO` for /// `(rows=4, cols=2048)` × `goldenCmsKeys()` (10 keys "flow-0".."flow-9", @@ -48,61 +41,6 @@ const HLL_GOLDEN_HEX: &str = include_str!("../src/sketches/testdata/hll_envelope /// cross-language byte parity. const DDSKETCH_GOLDEN_HEX: &str = "0801728e03096214ae47e17a843f128003000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000100000000000000000000000000000000000000000000000000000000000000000001000000000000000000000000000000000000000100000000000000000000000000000100000000000000000000010000000000000000010000000000000001000000000001000000000001000000000001000000010000000001000000010000010000000100000100000100000100000100010000010001000100010001000100010001000100010100010100010100010101000101010100010101010101010100000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000187f"; -#[test] -fn sketches_count_fastpath_matches_go_count_sketch_envelope() { - let rows = 3usize; - let cols = 512usize; - - let mut sk: Count, FastPath, DefaultXxHasher> = - Count::with_dimensions(rows, cols); - for i in 0..25 { - let key = format!("k-{}", (b'a' + (i % 5) as u8) as char); - sk.insert_many(&DataInput::String(key), 1i64); - } - - let storage = sk.as_storage(); - let mut counts_int: Vec = Vec::with_capacity(rows * cols); - let mut l2: Vec = Vec::with_capacity(rows); - for r in 0..rows { - let mut row_l2 = 0.0f64; - for c in 0..cols { - let cell = *storage.get(r, c).expect("cell in range"); - counts_int.push(cell); - row_l2 += (cell as f64) * (cell as f64); - } - l2.push(row_l2); - } - - let state = CountSketchState { - rows: rows as u32, - cols: cols as u32, - counter_type: CounterType::Int64 as i32, - counts_int, - counts_float: Vec::new(), - l2, - topk: None, - }; - let envelope = SketchEnvelope { - format_version: 1, - producer: None, - hash_spec: None, - sample_p: 0.0, - sketch_state: Some(SketchState::CountSketch(state)), - }; - let mut got = Vec::with_capacity(envelope.encoded_len()); - envelope.encode(&mut got).expect("prost encode"); - - let want = decode_hex(COUNTSKETCH_GOLDEN_HEX); - assert_eq!( - got, - want, - "sketches::Count, FastPath> envelope diverges from Go golden \ - ({} bytes got vs {} bytes want)", - got.len(), - want.len(), - ); -} - #[test] fn sketches_countmin_fastpath_matches_go_count_min_envelope() { let rows = 4usize; diff --git a/tests/xtest_consumer.rs b/tests/xtest_consumer.rs index 0113ff45..3c7dd943 100644 --- a/tests/xtest_consumer.rs +++ b/tests/xtest_consumer.rs @@ -9,7 +9,6 @@ //! kll.pb — KLLState with quantile items and coin RNG state //! ddsketch.pb — DDSketchState with alpha + bucket array //! hll.pb — HyperLogLogState (ErtlMLE variant) -//! countsketch.pb — CountSketchState with float64 signed counters //! coco.pb — CocoSketchState (hash+val+hasKey buckets) //! elastic.pb — ElasticState (heavy buckets + light CountMin) //! univmon.pb — UnivMonState (layered CountSketch + TopK heaps) @@ -245,44 +244,6 @@ fn cross_language_proto() { all_ok = false; } - // ----------------------------------------------------------------------- - // CountSketch - // ----------------------------------------------------------------------- - println!(); - println!("[CountSketch] Step 1/3 — Read countsketch.pb"); - let bytes = read_file(in_dir.join("countsketch.pb")); - let env = SketchEnvelope::decode(bytes.as_slice()).expect("decode countsketch envelope"); - - println!( - "[CountSketch] Step 2/3 — Validate envelope (format_version={}, producer={})", - env.format_version, - env.producer.as_ref().map_or("?", |p| &p.library) - ); - - let cs_state = match env.sketch_state { - Some(sketch_envelope::SketchState::CountSketch(ref s)) => s.clone(), - other => panic!("expected CountSketch sketch_state, got {other:?}"), - }; - - println!( - "[CountSketch] rows={} cols={} counter_type={:?}", - cs_state.rows, - cs_state.cols, - CounterType::try_from(cs_state.counter_type) - ); - - // Query "cs:hot" — inserted 200 extra times. - // Go hash: common.Hash64([]byte("cs:hot")) = xxh3_64_seeded(seedList[0], b"cs:hot") - let cs_hot_hash = xxh3_64_seeded(SEED_0, b"cs:hot"); - let cs_est = count_sketch_query_float(&cs_state, cs_hot_hash); - println!("[CountSketch] Step 3/3 — 'cs:hot' est = {cs_est:.0} (expect ≥ 200)"); - if cs_est >= 200.0 { - println!("[CountSketch] PASS"); - } else { - eprintln!("[CountSketch] FAIL: estimate {cs_est:.0} < 200"); - all_ok = false; - } - // ----------------------------------------------------------------------- // CocoSketch // ----------------------------------------------------------------------- @@ -790,38 +751,6 @@ fn hll_ertl_mle_estimate(state: &HyperLogLogState) -> u64 { (0.5 / std::f64::consts::LN_2 * m * m / z).round() as u64 } -// --------------------------------------------------------------------------- -// CountSketch median frequency query (float64 counters, packed hash) -// --------------------------------------------------------------------------- -// Mirrors Go's CountSketch.fastPacked64PosAndSign + ComputeMedianInlineF64. -// hash = xxh3_64_seeded(SEED_0, key) — Go's Hash64 - -fn count_sketch_query_float(state: &CountSketchState, hash: u64) -> f64 { - let rows = state.rows as usize; - let cols = state.cols as usize; - let bits_per_row = col_bits(cols); - let mask = (cols as u64) - 1; - // Read whichever counter field the producer populated (int vs float), - // discriminated by counter_type — same as CountMin. - let counts: Vec = if !state.counts_float.is_empty() { - state.counts_float.clone() - } else { - state.counts_int.iter().map(|&v| v as f64).collect() - }; - let counts = &counts; - - let mut estimates: Vec = Vec::with_capacity(rows); - for r in 0..rows { - let shift = (r as u64) * bits_per_row; - let col = ((hash >> shift) & mask) as usize; - // sign: bit (63-r) of hash → 1 means +1, 0 means −1 - let sign_bit = (hash >> (63 - r)) & 1; - let sign = if sign_bit == 1 { 1.0f64 } else { -1.0f64 }; - estimates.push(counts[r * cols + col] * sign); - } - median_f64(&mut estimates) -} - fn median_f64(v: &mut [f64]) -> f64 { if v.is_empty() { return 0.0; From 489912c0ccd83745bf9cdd9f52e9b22443acf75a Mon Sep 17 00:00:00 2001 From: GordonYuanyc Date: Fri, 2 Oct 2026 00:49:29 -0600 Subject: [PATCH 2/2] test(countsketch): pin fast-path string-key cells against Go's placement Count, FastPath> fed the 25 k-a..k-e inserts must hold the 15 signed cells Go's CountSketch wrote for the same stream. Also rewords the CSHeap portable module doc and the features.md proto list. Co-Authored-By: Claude Opus 5.5 --- docs/features.md | 2 +- docs/tests.md | 1 + .../portable/countsketch_topk.rs | 2 +- src/sketches/countsketch.rs | 38 +++++++++++++++++++ 4 files changed, 41 insertions(+), 2 deletions(-) diff --git a/docs/features.md b/docs/features.md index 64f296a3..232b591c 100644 --- a/docs/features.md +++ b/docs/features.md @@ -112,7 +112,7 @@ MessagePack (`rmp-serde`) support. **serde support** means the type derives `Ser | EHSketchList | Yes | Yes | | ExponentialHistogram | No | Yes | -Protobuf (prost): `.proto` definitions exist for CountMin, Count, HLL, DDSketch, KLL, Elastic, Coco, Hydra, and UnivMon. Rust conversion code is in progress. +Protobuf (prost): `.proto` definitions exist for CountMin, HLL, DDSketch, KLL, Elastic, Coco, Hydra, and UnivMon; Count Sketch has only the `CountSketchState` that Hydra and UnivMon embed. Rust conversion code is in progress. ### API Stability diff --git a/docs/tests.md b/docs/tests.md index 47732b4d..b0256e86 100644 --- a/docs/tests.md +++ b/docs/tests.md @@ -71,6 +71,7 @@ Wire tests: [`src/sketches/countsketch/wire.rs`](../src/sketches/countsketch/wir | `zipf_stream_stays_within_twenty_percent_for_most_keys` | Zipf stream keeps relative error under 20% for most keys. | On Zipf stream (`rows=5`, `cols=8192`, `domain=8192`, `exponent=1.1`, `N=200_000`), computes per-key relative error and requires at least 70% of keys with error `< 0.20`. | | `cs_regular_path_correctness` | Regular-path counter/sign mapping and estimates are exact on deterministic inserts. | Recomputes expected signed counter updates for `I32(0..9)` using regular hashing/sign logic, asserts exact matrix match after one pass, doubled counters after second pass, and estimate `== 2.0` for each inserted key. | | `cs_fast_path_correctness` | Fast-path row-hash/sign mapping matches expected counters. | Recomputes expected fast-path updates for `I32(0..9)` using matrix hash row slices and row signs, then asserts exact full-matrix equality. | +| `fast_path_string_keys_land_in_go_cells` | Fast-path string keys land in the same cells with the same signs as `sketchlib-go`'s Count Sketch. | Fills `Count, FastPath, DefaultXxHasher>` at `3x512` with 25 `insert_many(DataInput::String(..), 1)` calls over the five keys `k-a`..`k-e` and verifies the matrix is zero except 15 cells, five per row, each holding `+5` or `-5`, at the positions Go's serializer wrote for the same inserts. | | `count_sketch_round_trip_serialization` | The envelope frames a sketch under kind_id `0x04 0x00`, and the state survives a round trip. | Serializes a populated `3x8` regular-path `Vector2D` sketch and verifies the bytes open with the ASAPv1 magic, a `kind_id_len` of `2`, and `0x04 0x00`, then that the decode preserves rows, cols, and the full counter array. | | `count_sketch_negative_counters_round_trip` | Signed cells reach the wire and come back unchanged. | Round-trips a `2x4` matrix holding alternating positive and negative counters and verifies the decoded slice equals the source and still holds a negative cell. | | `count_sketch_custom_hasher_profile_round_trips_and_is_self_describing` | The metadata describes the hasher that built the sketch rather than a hardcoded profile. | For a `3x8` sketch over a hasher declaring its own `HashProfile`, verifies the bytes round-trip, differ from the standard-profile sketch's bytes over the same inserts, and are rejected by a standard-profile decode. | diff --git a/src/message_pack_format/portable/countsketch_topk.rs b/src/message_pack_format/portable/countsketch_topk.rs index 82ec8958..a3d1a7ec 100644 --- a/src/message_pack_format/portable/countsketch_topk.rs +++ b/src/message_pack_format/portable/countsketch_topk.rs @@ -1,7 +1,7 @@ //! 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 +//! the signed `rows x cols` matrix 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 diff --git a/src/sketches/countsketch.rs b/src/sketches/countsketch.rs index 6a3c7ef5..ab07610f 100644 --- a/src/sketches/countsketch.rs +++ b/src/sketches/countsketch.rs @@ -1012,4 +1012,42 @@ mod tests { hi / lo ); } + + /// Go's `CountSketch` places these 25 string inserts in these 15 cells + /// with these signs. + #[test] + fn fast_path_string_keys_land_in_go_cells() { + let (rows, cols) = (3usize, 512usize); + let mut sk: Count, FastPath, DefaultXxHasher> = + Count::with_dimensions(rows, cols); + for i in 0..25 { + let key = format!("k-{}", (b'a' + (i % 5) as u8) as char); + sk.insert_many(&DataInput::String(key), 1i64); + } + + let mut want = vec![0i64; rows * cols]; + for (r, c, v) in [ + (0, 61, -5), + (0, 127, 5), + (0, 217, -5), + (0, 276, -5), + (0, 467, -5), + (1, 78, 5), + (1, 185, -5), + (1, 278, 5), + (1, 503, -5), + (1, 510, 5), + (2, 6, -5), + (2, 165, -5), + (2, 198, 5), + (2, 210, -5), + (2, 314, -5), + ] { + want[r * cols + c] = v; + } + let got: Vec = (0..rows * cols) + .map(|i| sk.as_storage().query_one_counter(i / cols, i % cols)) + .collect(); + assert_eq!(got, want); + } }