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/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/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..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. | @@ -1254,31 +1255,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 +1582,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 +2119,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 +2166,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 +2176,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/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/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/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); + } } 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;