diff --git a/docs/msgpack-magic-ids.md b/docs/msgpack-magic-ids.md new file mode 100644 index 0000000..099b5a6 --- /dev/null +++ b/docs/msgpack-magic-ids.md @@ -0,0 +1,152 @@ +# ASAPv1 Sketch Wire Format + +Every serialised sketch binary produced by this library is wrapped in the +**ASAPv1 envelope**, a self-describing header that carries the sketch's +type discriminant plus wrapper-level metadata needed by consumers that update +or query sketches outside the original process. + +``` +┌───────────────┬────────────┬──────────────────┬───────────────────────┬───────────────────┬────────────────────┬──────────────────────┐ +│ b"ASAPv1": 6B │ version: u8 │ kind_id_len: u8 │ kind_id: [kind_id_len] │ metadata_len: u32 │ metadata: msgpack │ msgpack payload … │ +└───────────────┴────────────┴──────────────────┴───────────────────────┴───────────────────┴────────────────────┴──────────────────────┘ +``` + +| Field | Value | Notes | +|-------|-------|-------| +| `b"ASAPv1"` | `0x41 0x53 0x41 0x50 0x76 0x31` | 6-byte ASCII sentinel, not a valid msgpack prefix | +| `version` | `0x02` | Increment only if the envelope layout changes | +| `kind_id_len` | 1 or 2 | Number of `kind_id` bytes (1 for portable, 2 for native) | +| `kind_id` | see tables below | Canonical big-endian, no leading zero bytes | +| `metadata_len` | big-endian `u32` | Number of bytes in the metadata MessagePack block | +| `metadata` | compact msgpack array | Wrapper metadata; currently records the hash profile | +| payload | msgpack bytes | Compact (array) or named (map) depending on sketch type | + +Current metadata fields, in compact MessagePack array order: + +| Index | Field | Standard hash value | +|-------|-------|---------------------| +| 0 | `metadata_version` | `1` | +| 1 | `hash_spec_present` | `true` for portable IDs and registered `DefaultXxHasher` native IDs | +| 2 | `hash_profile_id` | `projectasap.xxh3.seedlist.v1` | +| 3 | `hash_algorithm` | `xxh3_64_128` | +| 4 | `seed_list` | `[0xcafe3553, 0xade3415118, 0x8cc70208, 0x2f024b2b, 0x451a3df5, 0x6a09e667, 0xbb67ae85, 0x3c6ef372, 0xa54ff53a, 0x510e527f, 0x9b05688c, 0x1f83d9ab, 0x5be0cd19, 0xcbbb9d5d, 0x629a292a, 0x9159015a, 0x152fecd8, 0x67332667, 0x8eb44a87, 0xdb0c2e0d]` | +| 5 | `canonical_seed_index` | `5` | +| 6 | `matrix_seed_index` | `0` | +| 7 | `hydra_seed_index` | `6` | +| 8 | `univmon_bottom_layer_seed_index` | `19` | +| 9 | `seed_derivation` | `seed_list_index_wrap` | +| 10 | `input_encoding` | `projectasap.input.v1` | + +If a wrapper cannot declare a complete registered hash profile, it still carries +metadata version `1` with `hash_spec_present = false` and empty/default hash +fields. Readers validate known hash metadata before exposing the payload. + +**Portable** sketches use a 1-byte `kind_id` (`kind_id_len = 1`). +**Native** Rust sketches use a 2-byte `kind_id` (`kind_id_len = 2`): first byte +is the type/mode discriminant, second byte is the hasher ID (`HASHER_*`). + +The `hasher_id` second byte is `0xFF` (`HASHER_UNKNOWN`) for types without an `H` +parameter. Custom hashers that do not register an ID also store `0xFF` — the mismatch +check is skipped on both sides when either value is `0xFF`. + +Kind IDs are **stable** — once assigned, a value is never reused or +reassigned. Adding a new sketch type requires a new constant; removing or +repurposing an existing constant is a **breaking protocol change**. + +The single source of truth in code is +[`src/message_pack_format/magic_ids.rs`](../src/message_pack_format/magic_ids.rs). +The Go mirror lives in +[`sketchlib-go/wire/asapmsgpack/magic_ids.go`](https://github.com/ProjectASAP/sketchlib-go/blob/main/wire/asapmsgpack/magic_ids.go). + +--- + +## Portable IDs (0x01 – 0x09) + +These IDs identify the **cross-language wire format** shared with +`sketchlib-go`. Any byte blob with a portable ID can be decoded by either +the Rust or Go implementation. + +| ID | Rust type / entry point | Go entry point | +|--------|--------------------------------------------------|----------------------------------------| +| `0x01` | `portable::HllSketch::to_msgpack` | `HLL.SerializeMsgpack` | +| `0x02` | `portable::CountMinSketch::to_msgpack` | `CountMinSketch.SerializeMsgpack` | +| `0x03` | `portable::CountMinSketchWithHeap::to_msgpack` | `CountSketch.SerializeMsgpackWithHeap` | +| `0x04` | `portable::CountSketch::to_msgpack` | `CountSketch.SerializeMsgpack` | +| `0x05` | `portable::DdSketch::to_msgpack` | `DDSketch.SerializeMsgpack` | +| `0x06` | `portable::KllSketch::to_msgpack` | _(Rust-only path; no Go equivalent)_ | +| `0x07` | `portable::HydraKllSketch::to_msgpack` | _(Rust-only path; no Go equivalent)_ | +| `0x08` | `portable::SetAggregator::to_msgpack` | _(Rust-only path; no Go equivalent)_ | +| `0x09` | `portable::DeltaResult::to_msgpack` | _(Rust-only path; no Go equivalent)_ | + +> **Note on `0x03`:** The Go producer calls this format "CountSketchWithHeap"; +> the Rust consumer knows it as `CountMinSketchWithHeap`. The delta-heap frame +> (`SerializeMsgpackWithHeapDelta`) shares the same magic ID because the Rust +> consumer uses the same `from_msgpack` path for both the full and delta shapes. + +--- + +## Native IDs (0x81 – 0x89) + +These IDs identify the **Rust-internal** format produced by +`serialize_to_bytes` / `deserialize_from_bytes` on the generic sketch types +in `src/sketches/`. Go never reads these bytes directly. + +The native wire format uses `rmp_serde`'s **named** (map) encoding +(`to_vec_named`), whereas the portable format uses **compact** (array) +encoding. The two are not interchangeable even for logically equivalent types +(e.g., `sketches::DDSketch` vs `portable::DdSketch`). + +| ID | Rust type / method | Notes | +|--------|--------------------------------------------------|-------| +| `0x81` | `CountMin<_, RegularPath, _>::serialize_to_bytes` | Named map format | +| `0x82` | `CountMin<_, FastPath, _>::serialize_to_bytes` | Named map format | +| `0x83` | `Count<_, RegularPath, _>::serialize_to_bytes` | Named map format | +| `0x84` | `Count<_, FastPath, _>::serialize_to_bytes` | Named map format | +| `0x85` | `CountL2HH::serialize_to_bytes` | Named map format (CMSHeap / heavy-hitter) | +| `0x86` | `HyperLogLogImpl::serialize_to_bytes` | Named map format | +| `0x87` | `HyperLogLogImpl::serialize_to_bytes` | Named map format | +| `0x88` | `HyperLogLogHIPImpl<_>::serialize_to_bytes` | Named map; always `HASHER_DEFAULT_XX` as second byte | +| `0x89` | `sketches::DDSketch::serialize_to_bytes` | Named map; distinct from portable `0x05` | +| `0x8a` | `sketches::KLL::serialize_to_bytes` | Compact array format | +| `0x8b` | `sketches::KLLDynamic::serialize_to_bytes` | Compact array format | +| `0x8c` | `sketches::KMV::serialize_to_bytes` | Named map format (experimental feature) | +| `0x8d` | `sketch_framework::Hydra::serialize_to_bytes` | Named map format | +| `0x8e` | `sketch_framework::UnivMon::serialize_to_bytes` | Named map format | + +### Relationship between native and portable KLL + +`portable::KllSketch::to_msgpack` (`0x06`) embeds the raw KLL cell bytes +inside a `KllSketchData { k, sketch_bytes }` msgpack struct. Those embedded +bytes are produced by `KLL::serialize_to_bytes` and therefore carry the native +`0x8a` prefix. The portable round-trip is: + +``` +KllSketch::to_msgpack() → [ ASAPv1 | v=1 | len=1 | 0x06 | metadata | msgpack([k, [ASAPv1|v=1|len=2|0x8a|0xff|metadata|raw_kll]]) ] +KllSketch::from_msgpack() → decode_wrapper → kind_id=[0x06], payload=msgpack struct + → decode struct: k + sketch_bytes + → KLL::deserialize_from_bytes(sketch_bytes) + → decode_wrapper → kind_id=[0x8a, 0xff], payload=raw_kll +``` + +The same pattern applies to `HydraKllSketch` (`0x07`), which contains a grid +of KLL cells. + +--- + +## Adding a new sketch type + +1. Add a constant to `src/message_pack_format/magic_ids.rs` (choose the next + available value in the appropriate range). +2. If the type has a **portable** wire format shared with Go, also add the + constant to `sketchlib-go/wire/asapmsgpack/magic_ids.go` with the same + value, and update `SerializeMsgpack` / `DeserializeMsgpack` in the + corresponding Go sketch package to use `asapmsgpack.EncodeWrapper` / + `asapmsgpack.DecodeWrapper`. +3. Implement `MessagePackCodec::to_msgpack` / `from_msgpack` (or + `serialize_to_bytes` / `deserialize_from_bytes` for native types) using + `magic_ids::encode_wrapper` / `magic_ids::decode_wrapper`. + - Portable: `kind_id = &[MAGIC_CONSTANT]` (1 byte) + - Native with hasher: `kind_id = &[TYPE_BYTE, H::hasher_magic_id()]` (2 bytes) + - Native without hasher: `kind_id = &[TYPE_BYTE, HASHER_UNKNOWN]` (2 bytes) +4. Add or update round-trip tests. +5. Update this document. diff --git a/src/common/hash.rs b/src/common/hash.rs index ad8cfbc..d2df1ac 100644 --- a/src/common/hash.rs +++ b/src/common/hash.rs @@ -65,6 +65,17 @@ pub trait SketchHasher: Clone + Debug { cols: usize, key: &DataInput, ) -> Self::HashType; + + /// Returns a stable byte discriminant for this hasher, embedded as the + /// second byte of every native serialized header so readers can verify + /// the hash function matches. + /// + /// The default is [`crate::message_pack_format::magic_ids::HASHER_UNKNOWN`] + /// (`0xff`), which disables the mismatch check on both sides. Override + /// this for any hasher that should be verified across process boundaries. + fn hasher_magic_id() -> u8 { + crate::message_pack_format::magic_ids::HASHER_UNKNOWN + } } /// Default hasher using twox_hash (XxHash3). This is the built-in implementation @@ -75,6 +86,10 @@ pub struct DefaultXxHasher; impl SketchHasher for DefaultXxHasher { type HashType = MatrixHashType; + fn hasher_magic_id() -> u8 { + crate::message_pack_format::magic_ids::HASHER_DEFAULT_XX + } + #[inline(always)] fn hash64_seeded(d: usize, key: &DataInput) -> u64 { let seed = SEEDLIST[normalized_seed_idx(d)]; diff --git a/src/common/input.rs b/src/common/input.rs index 0fbd337..2eb3e19 100644 --- a/src/common/input.rs +++ b/src/common/input.rs @@ -483,7 +483,7 @@ impl HydraCounter { (HydraCounter::UNIVERSAL(um), HydraQuery::L1Norm) => Ok(um.calc_l1()), (HydraCounter::UNIVERSAL(um), HydraQuery::L2Norm) => Ok(um.calc_l2()), (HydraCounter::UNIVERSAL(um), HydraQuery::Entropy) => Ok(um.calc_entropy()), - (c, q) => Err(format!("{} does not support {}", c, q)), + (c, q) => Err(format!("{c} does not support {q}")), } } diff --git a/src/common/structure_utils.rs b/src/common/structure_utils.rs index 29e9165..a17e461 100644 --- a/src/common/structure_utils.rs +++ b/src/common/structure_utils.rs @@ -365,8 +365,7 @@ mod tests { let sort_median = median_three_sort(v); assert_eq!( fast_median, sort_median, - "median for sort is {sort_median} but fast gives {fast_median}, input is {:?}", - v + "median for sort is {sort_median} but fast gives {fast_median}, input is {v:?}" ); } for v in &mut four_vec { @@ -374,8 +373,7 @@ mod tests { let sort_median = median_four_sort(v); assert_eq!( fast_median, sort_median, - "median for sort is {sort_median} but fast gives {fast_median}, input is {:?}", - v + "median for sort is {sort_median} but fast gives {fast_median}, input is {v:?}" ); } for v in &mut five_vec { @@ -383,8 +381,7 @@ mod tests { let sort_median = median_five_sort(v); assert_eq!( fast_median, sort_median, - "median for sort is {sort_median} but fast gives {fast_median}, input is {:?}", - v + "median for sort is {sort_median} but fast gives {fast_median}, input is {v:?}" ); } } diff --git a/src/message_pack_format/error.rs b/src/message_pack_format/error.rs index a69452b..ef033cf 100644 --- a/src/message_pack_format/error.rs +++ b/src/message_pack_format/error.rs @@ -12,6 +12,9 @@ pub enum Error { Encode(rmp_serde::encode::Error), /// MessagePack decoding failed. Decode(rmp_serde::decode::Error), + /// The leading magic-ID byte was missing or did not match the expected value. + /// `got` is the byte that was found (or `None` if the buffer was empty). + BadMagicId { expected: u8, got: Option }, } impl fmt::Display for Error { @@ -19,6 +22,16 @@ impl fmt::Display for Error { match self { Error::Encode(e) => write!(f, "MessagePack encode failed: {e}"), Error::Decode(e) => write!(f, "MessagePack decode failed: {e}"), + Error::BadMagicId { expected, got } => match got { + Some(b) => write!( + f, + "MessagePack magic-ID mismatch: expected 0x{expected:02x}, got 0x{b:02x}" + ), + None => write!( + f, + "MessagePack magic-ID missing: expected 0x{expected:02x} but buffer is empty" + ), + }, } } } @@ -28,6 +41,7 @@ impl StdError for Error { match self { Error::Encode(e) => Some(e), Error::Decode(e) => Some(e), + Error::BadMagicId { .. } => None, } } } diff --git a/src/message_pack_format/magic_ids.rs b/src/message_pack_format/magic_ids.rs new file mode 100644 index 0000000..65fee54 --- /dev/null +++ b/src/message_pack_format/magic_ids.rs @@ -0,0 +1,420 @@ +//! Magic-ID constants and wrapper encoding for the ASAP sketch wire format. +//! +//! Every serialized binary produced by this library is wrapped in the **ASAPv1 +//! envelope**, a self-describing header that identifies the sketch type and +//! reserves space for future metadata without a fixed-size ceiling: +//! +//! ```text +//! [ b"ASAPv1" | version: u8 | kind_id_len: u8 | kind_id: [u8; kind_id_len] | metadata_len: u32 | metadata: bytes | ] +//! ``` +//! +//! * `b"ASAPv1"` — 6-byte ASCII sentinel; unambiguously not a msgpack value. +//! * `version` — envelope layout version; currently `0x01`. +//! * `kind_id_len + kind_id` — variable-length sketch discriminant encoded as +//! canonical unsigned big-endian with no leading zero bytes. For current +//! sketch types this is 1–2 bytes; future additions can use more without a +//! protocol change. +//! * `metadata_len + metadata` — MessagePack metadata describing the hash +//! profile needed for cross-process updates and queries. +//! +//! **Portable** sketches (cross-language, shared with `sketchlib-go`) use a +//! 1-byte `kind_id` drawn from the `0x01–0x09` range. +//! +//! **Native** Rust sketches use a 2-byte `kind_id`: first byte is the family / +//! mode discriminant (`0x81–0x8e`); second byte is the hasher ID (`HASHER_*`). +//! +//! Magic IDs are **stable** — once assigned, a value is never reused. Adding +//! a new sketch type requires a new constant here; removing or repurposing an +//! existing constant is a **breaking protocol change**. The Go mirror of this +//! table lives in `sketchlib-go/wire/asapmsgpack/magic_ids.go`. + +// ── Envelope constants ──────────────────────────────────────────────────────── + +/// 6-byte ASCII sentinel that opens every ASAP sketch binary. +pub const WRAPPER_MAGIC: &[u8; 6] = b"ASAPv1"; + +/// Envelope layout version stored immediately after `WRAPPER_MAGIC`. Increment +/// if the header structure (field order, field semantics) ever changes. +pub const WRAPPER_VERSION: u8 = 0x02; + +use serde::{Deserialize, Serialize}; + +pub const HASH_PROFILE_PROJECTASAP_XXH3_V1: &str = "projectasap.xxh3.seedlist.v1"; +pub const HASH_ALGORITHM_XXH3_64_128: &str = "xxh3_64_128"; +pub const HASH_SEED_DERIVATION_INDEX_WRAP: &str = "seed_list_index_wrap"; +pub const HASH_INPUT_ENCODING_PROJECTASAP_V1: &str = "projectasap.input.v1"; +pub const MATRIX_BASE_SEED_INDEX: u32 = 0; + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct WrapperMetadata { + pub metadata_version: u8, + pub hash_spec_present: bool, + pub hash_profile_id: String, + pub hash_algorithm: String, + pub seed_list: Vec, + pub canonical_seed_index: u32, + pub matrix_seed_index: u32, + pub hydra_seed_index: u32, + pub univmon_bottom_layer_seed_index: u32, + pub seed_derivation: String, + pub input_encoding: String, +} + +impl WrapperMetadata { + pub fn standard_hash() -> Self { + Self { + metadata_version: 1, + hash_spec_present: true, + hash_profile_id: HASH_PROFILE_PROJECTASAP_XXH3_V1.to_string(), + hash_algorithm: HASH_ALGORITHM_XXH3_64_128.to_string(), + seed_list: crate::SEEDLIST.to_vec(), + canonical_seed_index: crate::CANONICAL_HASH_SEED as u32, + matrix_seed_index: MATRIX_BASE_SEED_INDEX, + hydra_seed_index: crate::HYDRA_SEED as u32, + univmon_bottom_layer_seed_index: crate::BOTTOM_LAYER_FINDER as u32, + seed_derivation: HASH_SEED_DERIVATION_INDEX_WRAP.to_string(), + input_encoding: HASH_INPUT_ENCODING_PROJECTASAP_V1.to_string(), + } + } + + pub fn no_hash_spec() -> Self { + Self { + metadata_version: 1, + hash_spec_present: false, + hash_profile_id: String::new(), + hash_algorithm: String::new(), + seed_list: Vec::new(), + canonical_seed_index: 0, + matrix_seed_index: 0, + hydra_seed_index: 0, + univmon_bottom_layer_seed_index: 0, + seed_derivation: String::new(), + input_encoding: String::new(), + } + } + + fn validate(&self) -> Result<(), String> { + if self.metadata_version != 1 { + return Err(format!( + "ASAPv1 wrapper: unsupported metadata_version {}", + self.metadata_version + )); + } + if !self.hash_spec_present { + let expected = Self::no_hash_spec(); + if self != &expected { + return Err("ASAPv1 wrapper: no-hash metadata mismatch".to_string()); + } + return Ok(()); + } + let expected = Self::standard_hash(); + if self != &expected { + return Err("ASAPv1 wrapper: hash metadata mismatch".to_string()); + } + Ok(()) + } +} + +fn metadata_for_kind_id(kind_id: &[u8]) -> WrapperMetadata { + match kind_id { + // Portable wire formats are cross-language and always use the standard + // ProjectASAP hash profile when they need key hashing. Keep the profile + // present for every portable kind so consumers can validate uniformly. + [0x01..=0x09] => WrapperMetadata::standard_hash(), + // Native Rust sketches with an explicit DefaultXxHasher byte. + [_, HASHER_DEFAULT_XX] => WrapperMetadata::standard_hash(), + // Native framework sketches with no generic hasher but standard hash use. + [NATIVE_HYDRA | NATIVE_UNIVMON, HASHER_UNKNOWN] => WrapperMetadata::standard_hash(), + // DDSketch/KLL and custom/unknown hashers do not have a complete, + // registered hash profile to record. + _ => WrapperMetadata::no_hash_spec(), + } +} + +/// Prepend the ASAPv1 envelope to `payload` and return the complete binary. +/// +/// `kind_id` identifies the sketch type (1–2 bytes for all current types). +pub fn encode_wrapper(kind_id: &[u8], payload: &[u8]) -> Vec { + let metadata = metadata_for_kind_id(kind_id); + let metadata_bytes = + rmp_serde::to_vec(&metadata).expect("WrapperMetadata serialization cannot fail"); + let metadata_len = u32::try_from(metadata_bytes.len()).expect("wrapper metadata too large"); + let mut out = Vec::with_capacity( + WRAPPER_MAGIC.len() + 1 + 1 + kind_id.len() + 4 + metadata_bytes.len() + payload.len(), + ); + out.extend_from_slice(WRAPPER_MAGIC); + out.push(WRAPPER_VERSION); + out.push(kind_id.len() as u8); + out.extend_from_slice(kind_id); + out.extend_from_slice(&metadata_len.to_be_bytes()); + out.extend_from_slice(&metadata_bytes); + out.extend_from_slice(payload); + out +} + +/// Strip the ASAPv1 envelope from `bytes` and return `(kind_id, payload)`. +/// +/// Returns `Err(String)` on any structural mismatch so callers can convert to +/// their own error type. +pub fn decode_wrapper(bytes: &[u8]) -> Result<(&[u8], &[u8]), String> { + let (kind_id, _metadata, payload) = decode_wrapper_with_metadata(bytes)?; + Ok((kind_id, payload)) +} + +/// Strip the ASAPv1 envelope from `bytes` and return `(kind_id, metadata, payload)`. +/// +/// Returns `Err(String)` on any structural mismatch or unsupported metadata. +pub fn decode_wrapper_with_metadata( + bytes: &[u8], +) -> Result<(&[u8], WrapperMetadata, &[u8]), String> { + let magic_len = WRAPPER_MAGIC.len(); + let version_offset = magic_len; + let kind_id_len_offset = magic_len + 1; + let kind_id_offset = magic_len + 2; + let min_len = kind_id_offset + 1 + 4; + + if bytes.len() < min_len { + return Err(format!( + "ASAPv1 wrapper: too short ({} bytes, need ≥{min_len})", + bytes.len() + )); + } + if &bytes[..magic_len] != WRAPPER_MAGIC { + return Err(format!( + "ASAPv1 wrapper: bad magic {:?}, expected b\"ASAPv1\"", + &bytes[..magic_len] + )); + } + let version = bytes[version_offset]; + if version != WRAPPER_VERSION { + return Err(format!( + "ASAPv1 wrapper: unsupported version 0x{version:02x}" + )); + } + let kind_id_len = bytes[kind_id_len_offset] as usize; + let metadata_len_offset = kind_id_offset + kind_id_len; + if bytes.len() < metadata_len_offset + 4 { + return Err(format!( + "ASAPv1 wrapper: kind_id_len={kind_id_len} but only {} bytes available after offset {kind_id_offset}", + bytes.len().saturating_sub(kind_id_offset) + )); + } + let metadata_len = u32::from_be_bytes( + bytes[metadata_len_offset..metadata_len_offset + 4] + .try_into() + .expect("metadata length slice has four bytes"), + ) as usize; + let metadata_start = metadata_len_offset + 4; + let metadata_end = metadata_start + metadata_len; + if bytes.len() < metadata_end { + return Err(format!( + "ASAPv1 wrapper: metadata_len={metadata_len} but only {} bytes available after offset {metadata_start}", + bytes.len().saturating_sub(metadata_start) + )); + } + let metadata: WrapperMetadata = rmp_serde::from_slice(&bytes[metadata_start..metadata_end]) + .map_err(|err| format!("ASAPv1 wrapper: invalid metadata: {err}"))?; + metadata.validate()?; + let kind_id = &bytes[kind_id_offset..metadata_len_offset]; + let expected_metadata = metadata_for_kind_id(kind_id); + if metadata != expected_metadata { + return Err("ASAPv1 wrapper: metadata does not match kind_id".to_string()); + } + Ok((kind_id, metadata, &bytes[metadata_end..])) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn encode_wrapper_with_metadata( + kind_id: &[u8], + metadata: &WrapperMetadata, + payload: &[u8], + ) -> Vec { + let metadata_bytes = rmp_serde::to_vec(metadata).expect("metadata serialization"); + let metadata_len = u32::try_from(metadata_bytes.len()).expect("metadata too large"); + let mut out = Vec::new(); + out.extend_from_slice(WRAPPER_MAGIC); + out.push(WRAPPER_VERSION); + out.push(kind_id.len() as u8); + out.extend_from_slice(kind_id); + out.extend_from_slice(&metadata_len.to_be_bytes()); + out.extend_from_slice(&metadata_bytes); + out.extend_from_slice(payload); + out + } + + #[test] + fn portable_wrapper_records_standard_hash_metadata() { + let payload = [0x91, 0x2a]; + let encoded = encode_wrapper(&[COUNT_MIN_SKETCH], &payload); + + let (kind_id, metadata, decoded_payload) = + decode_wrapper_with_metadata(&encoded).expect("decode wrapper"); + + assert_eq!(kind_id, &[COUNT_MIN_SKETCH]); + assert_eq!(decoded_payload, payload); + assert_eq!(metadata, WrapperMetadata::standard_hash()); + assert_eq!(metadata.seed_list, crate::SEEDLIST); + assert_eq!( + metadata.canonical_seed_index, + crate::CANONICAL_HASH_SEED as u32 + ); + assert_eq!(metadata.hydra_seed_index, crate::HYDRA_SEED as u32); + assert_eq!( + metadata.univmon_bottom_layer_seed_index, + crate::BOTTOM_LAYER_FINDER as u32 + ); + } + + #[test] + fn unknown_native_hasher_does_not_claim_hash_metadata() { + let payload = [0x90]; + let kind_id = [NATIVE_COUNT_MIN_REGULAR, HASHER_UNKNOWN]; + let encoded = encode_wrapper(&kind_id, &payload); + + let (decoded_kind_id, metadata, decoded_payload) = + decode_wrapper_with_metadata(&encoded).expect("decode wrapper"); + + assert_eq!(decoded_kind_id, kind_id); + assert_eq!(decoded_payload, payload); + assert_eq!(metadata, WrapperMetadata::no_hash_spec()); + } + + #[test] + fn wrapper_rejects_hash_metadata_mismatch() { + let payload = [0x90]; + let mut metadata = WrapperMetadata::standard_hash(); + metadata.canonical_seed_index += 1; + let encoded = encode_wrapper_with_metadata(&[COUNT_SKETCH], &metadata, &payload); + + let err = decode_wrapper(&encoded).expect_err("metadata mismatch should fail"); + + assert!(err.contains("hash metadata mismatch"), "{err}"); + } + + #[test] + fn wrapper_rejects_metadata_kind_id_mismatch() { + let payload = [0x90]; + let metadata = WrapperMetadata::no_hash_spec(); + let encoded = encode_wrapper_with_metadata(&[COUNT_SKETCH], &metadata, &payload); + + let err = decode_wrapper(&encoded).expect_err("metadata kind mismatch should fail"); + + assert!(err.contains("metadata does not match kind_id"), "{err}"); + } +} + +/// HLL sketch (all variants: Regular, Datafusion, Hip). +pub const HLL: u8 = 0x01; + +/// Count-Min sketch (f64 counters, no top-k heap). +pub const COUNT_MIN_SKETCH: u8 = 0x02; + +/// Count-Min sketch with top-k heap (f64 counters + heap). +pub const COUNT_MIN_SKETCH_WITH_HEAP: u8 = 0x03; + +/// Count Sketch (signed f64 counters). +pub const COUNT_SKETCH: u8 = 0x04; + +/// DDSketch (quantile sketch, alpha-parameterised bucket array). +pub const DD_SKETCH: u8 = 0x05; + +/// KLL quantile sketch. +pub const KLL_SKETCH: u8 = 0x06; + +/// Hydra-KLL sketch (grid of KLL cells). +pub const HYDRA_KLL_SKETCH: u8 = 0x07; + +/// Set aggregator (distinct string set). +pub const SET_AGGREGATOR: u8 = 0x08; + +/// Delta-set aggregator result (added / removed string sets). +pub const DELTA_RESULT: u8 = 0x09; + +// ── Hasher discriminants ───────────────────────────────────────────────────── +// +// Embedded as the second header byte in every native sketch binary so the +// reader can verify that it uses the same hash function as the writer. +// Custom hashers that do not register an ID are stored as HASHER_UNKNOWN, +// which suppresses the mismatch check on both sides. + +/// Default XxHash3-64 hasher (`DefaultXxHasher`). +pub const HASHER_DEFAULT_XX: u8 = 0x01; + +/// Sentinel for custom hashers with no registered ID — mismatch check is +/// skipped when either the stored value or the expected value is this byte. +pub const HASHER_UNKNOWN: u8 = 0xff; + +/// Validates the stored hasher byte against the expected hasher `H`. +/// +/// The check is skipped (returns `Ok`) when either side is `HASHER_UNKNOWN`, +/// allowing custom hashers to interoperate without requiring a registered ID. +pub(crate) fn check_hasher_id( + stored: u8, +) -> Result<(), rmp_serde::decode::Error> { + let expected = H::hasher_magic_id(); + if stored == HASHER_UNKNOWN || expected == HASHER_UNKNOWN || stored == expected { + return Ok(()); + } + Err(rmp_serde::decode::Error::Uncategorized(format!( + "hasher mismatch: stored 0x{stored:02x}, expected 0x{expected:02x}" + ))) +} + +// ── Native (Rust-internal) sketch types ───────────────────────────────────── +// +// These are the generic sketch types in `crate::sketches`. Their wire format +// is produced by `serialize_to_bytes` / `deserialize_from_bytes` and is +// internal to Rust — Go (`sketchlib-go`) never reads these bytes directly. +// +// The first header byte encodes both the sketch family AND the phantom-type +// parameters (Mode, Variant) that are invisible in the msgpack payload. +// The second header byte encodes the hasher (see HASHER_* above). +// +// Layout: [ family+mode byte | hasher byte | ] +// +// ID range 0x81–0x8f is reserved for native types. + +/// Count-Min sketch with `RegularPath` hashing mode. +pub const NATIVE_COUNT_MIN_REGULAR: u8 = 0x81; + +/// Count-Min sketch with `FastPath` hashing mode. +pub const NATIVE_COUNT_MIN_FAST: u8 = 0x82; + +/// Count Sketch with `RegularPath` hashing mode. +pub const NATIVE_COUNT_SKETCH_REGULAR: u8 = 0x83; + +/// Count Sketch with `FastPath` hashing mode. +pub const NATIVE_COUNT_SKETCH_FAST: u8 = 0x84; + +/// Count-Min + heavy-hitter heap (`CountL2HH`). +pub const NATIVE_CMS_HEAP: u8 = 0x85; + +/// HyperLogLog Classic (HLL++) estimator (`HyperLogLogImpl`). +pub const NATIVE_HLL_CLASSIC: u8 = 0x86; + +/// HyperLogLog ErtlMLE estimator (`HyperLogLogImpl`). +pub const NATIVE_HLL_ERTL_MLE: u8 = 0x87; + +/// HyperLogLog HIP variant (`HyperLogLogHIPImpl<_>`). +pub const NATIVE_HLL_HIP: u8 = 0x88; + +/// DDSketch (`sketches::DDSketch`). +pub const NATIVE_DD_SKETCH: u8 = 0x89; + +/// KLL quantile sketch (`sketches::kll::KLL`). +pub const NATIVE_KLL: u8 = 0x8a; + +/// Dynamic KLL quantile sketch (`sketches::kll_dynamic::KLLDynamic`). +pub const NATIVE_KLL_DYNAMIC: u8 = 0x8b; + +/// KMV (K-Minimum Values) sketch (`sketches::KMV`). +pub const NATIVE_KMV: u8 = 0x8c; + +/// Hydra composite sketch (`sketch_framework::hydra`). +pub const NATIVE_HYDRA: u8 = 0x8d; + +/// UnivMon sketch (`sketch_framework::univmon`). +pub const NATIVE_UNIVMON: u8 = 0x8e; diff --git a/src/message_pack_format/mod.rs b/src/message_pack_format/mod.rs index 7734463..3f16d9f 100644 --- a/src/message_pack_format/mod.rs +++ b/src/message_pack_format/mod.rs @@ -20,6 +20,7 @@ pub mod codec; pub mod error; +pub mod magic_ids; pub mod native; pub mod portable; diff --git a/src/message_pack_format/native/countminsketch.rs b/src/message_pack_format/native/countminsketch.rs index bed39e2..4471d24 100644 --- a/src/message_pack_format/native/countminsketch.rs +++ b/src/message_pack_format/native/countminsketch.rs @@ -4,9 +4,22 @@ use serde::{Deserialize, Serialize}; use crate::message_pack_format::{Error, MessagePackCodec}; use crate::sketches::countminsketch::CountMin; -use crate::{MatrixStorage, SketchHasher}; +use crate::{FastPath, MatrixStorage, RegularPath, SketchHasher}; -impl MessagePackCodec for CountMin +impl MessagePackCodec for CountMin +where + S: MatrixStorage + Serialize + for<'de> Deserialize<'de>, +{ + fn to_msgpack(&self) -> Result, Error> { + Ok(self.serialize_to_bytes()?) + } + + fn from_msgpack(bytes: &[u8]) -> Result { + Ok(Self::deserialize_from_bytes(bytes)?) + } +} + +impl MessagePackCodec for CountMin where S: MatrixStorage + Serialize + for<'de> Deserialize<'de>, { diff --git a/src/message_pack_format/native/countsketch.rs b/src/message_pack_format/native/countsketch.rs index 03a2662..453d106 100644 --- a/src/message_pack_format/native/countsketch.rs +++ b/src/message_pack_format/native/countsketch.rs @@ -4,9 +4,23 @@ use serde::{Deserialize, Serialize}; use crate::message_pack_format::{Error, MessagePackCodec}; use crate::sketches::countsketch::{Count, CountSketchCounter}; -use crate::{MatrixStorage, SketchHasher}; +use crate::{FastPath, MatrixStorage, RegularPath, SketchHasher}; -impl MessagePackCodec for Count +impl MessagePackCodec for Count +where + S: MatrixStorage + Serialize + for<'de> Deserialize<'de>, + C: CountSketchCounter, +{ + fn to_msgpack(&self) -> Result, Error> { + Ok(self.serialize_to_bytes()?) + } + + fn from_msgpack(bytes: &[u8]) -> Result { + Ok(Self::deserialize_from_bytes(bytes)?) + } +} + +impl MessagePackCodec for Count where S: MatrixStorage + Serialize + for<'de> Deserialize<'de>, C: CountSketchCounter, diff --git a/src/message_pack_format/native/hll.rs b/src/message_pack_format/native/hll.rs index 36f76e9..5c624f7 100644 --- a/src/message_pack_format/native/hll.rs +++ b/src/message_pack_format/native/hll.rs @@ -1,11 +1,27 @@ //! Native MessagePack codec impls for [`crate::sketches::hll`]. +use serde::{Deserialize, Serialize}; + use crate::message_pack_format::{Error, MessagePackCodec}; -use crate::sketches::hll::{HyperLogLogHIPImpl, HyperLogLogImpl}; +use crate::sketches::hll::{Classic, ErtlMLE, HyperLogLogHIPImpl, HyperLogLogImpl}; use crate::{HllRegisterStorage, SketchHasher}; -impl MessagePackCodec - for HyperLogLogImpl +impl MessagePackCodec for HyperLogLogImpl +where + Registers: HllRegisterStorage + Serialize + for<'de> Deserialize<'de>, +{ + fn to_msgpack(&self) -> Result, Error> { + Ok(self.serialize_to_bytes()?) + } + + fn from_msgpack(bytes: &[u8]) -> Result { + Ok(Self::deserialize_from_bytes(bytes)?) + } +} + +impl MessagePackCodec for HyperLogLogImpl +where + Registers: HllRegisterStorage + Serialize + for<'de> Deserialize<'de>, { fn to_msgpack(&self) -> Result, Error> { Ok(self.serialize_to_bytes()?) diff --git a/src/message_pack_format/portable/countminsketch.rs b/src/message_pack_format/portable/countminsketch.rs index 41fd91a..b99264c 100644 --- a/src/message_pack_format/portable/countminsketch.rs +++ b/src/message_pack_format/portable/countminsketch.rs @@ -3,7 +3,7 @@ use serde::{Deserialize, Serialize}; -use crate::message_pack_format::{Error as MsgPackError, MessagePackCodec}; +use crate::message_pack_format::{Error as MsgPackError, MessagePackCodec, magic_ids}; use crate::sketches::countminsketch::CountMin; use crate::{DataInput, FastPath, Vector2D}; @@ -442,11 +442,23 @@ impl MessagePackCodec for CountMinSketch { rows: self.rows, cols: self.cols, }; - Ok(rmp_serde::to_vec(&wire)?) + let payload = rmp_serde::to_vec(&wire)?; + Ok(magic_ids::encode_wrapper( + &[magic_ids::COUNT_MIN_SKETCH], + &payload, + )) } fn from_msgpack(bytes: &[u8]) -> Result { - let wire: CountMinSketchWire = rmp_serde::from_slice(bytes)?; + let (kind_id, payload) = magic_ids::decode_wrapper(bytes) + .map_err(|msg| MsgPackError::Decode(rmp_serde::decode::Error::Uncategorized(msg)))?; + if kind_id != [magic_ids::COUNT_MIN_SKETCH] { + return Err(MsgPackError::BadMagicId { + expected: magic_ids::COUNT_MIN_SKETCH, + got: kind_id.first().copied(), + }); + } + let wire: CountMinSketchWire = rmp_serde::from_slice(payload)?; let backend = sketchlib_cms_from_matrix(wire.rows, wire.cols, &wire.sketch); Ok(Self { rows: wire.rows, diff --git a/src/message_pack_format/portable/countminsketch_topk.rs b/src/message_pack_format/portable/countminsketch_topk.rs index fa850f9..35d962b 100644 --- a/src/message_pack_format/portable/countminsketch_topk.rs +++ b/src/message_pack_format/portable/countminsketch_topk.rs @@ -6,7 +6,7 @@ use serde::{Deserialize, Serialize}; -use crate::message_pack_format::{Error as MsgPackError, MessagePackCodec}; +use crate::message_pack_format::{Error as MsgPackError, MessagePackCodec, magic_ids}; use crate::sketches::countminsketch_topk::CMSHeap; use crate::{DataInput, RegularPath, Vector2D}; @@ -86,7 +86,7 @@ pub fn heap_to_wire(cms_heap: &SketchlibCMSHeap) -> Vec { .map(|hh_item| { let key = match &hh_item.key { crate::HeapItem::String(s) => s.clone(), - other => format!("{:?}", other), + other => format!("{other:?}"), }; WireHeapItem { key, @@ -297,11 +297,23 @@ impl MessagePackCodec for CountMinSketchWithHeap { topk_heap: self.topk_heap_items(), heap_size: self.heap_size, }; - Ok(rmp_serde::to_vec(&wire)?) + let payload = rmp_serde::to_vec(&wire)?; + Ok(magic_ids::encode_wrapper( + &[magic_ids::COUNT_MIN_SKETCH_WITH_HEAP], + &payload, + )) } fn from_msgpack(bytes: &[u8]) -> Result { - let wire: CountMinSketchWithHeapWire = rmp_serde::from_slice(bytes)?; + let (kind_id, payload) = magic_ids::decode_wrapper(bytes) + .map_err(|msg| MsgPackError::Decode(rmp_serde::decode::Error::Uncategorized(msg)))?; + if kind_id != [magic_ids::COUNT_MIN_SKETCH_WITH_HEAP] { + return Err(MsgPackError::BadMagicId { + expected: magic_ids::COUNT_MIN_SKETCH_WITH_HEAP, + got: kind_id.first().copied(), + }); + } + let wire: CountMinSketchWithHeapWire = rmp_serde::from_slice(payload)?; let mut sorted_topk_heap = wire.topk_heap; sorted_topk_heap.sort_by(|a, b| b.value.partial_cmp(&a.value).unwrap()); diff --git a/src/message_pack_format/portable/countsketch.rs b/src/message_pack_format/portable/countsketch.rs index 408c693..2d8c3a7 100644 --- a/src/message_pack_format/portable/countsketch.rs +++ b/src/message_pack_format/portable/countsketch.rs @@ -3,7 +3,7 @@ use serde::{Deserialize, Serialize}; -use crate::message_pack_format::{Error as MsgPackError, MessagePackCodec}; +use crate::message_pack_format::{Error as MsgPackError, MessagePackCodec, magic_ids}; // ===================================================================== // ASAP runtime wire-format-aligned variant . @@ -434,11 +434,23 @@ impl CountSketch { impl MessagePackCodec for CountSketch { fn to_msgpack(&self) -> Result, MsgPackError> { - Ok(rmp_serde::to_vec(self)?) + let payload = rmp_serde::to_vec(self)?; + Ok(magic_ids::encode_wrapper( + &[magic_ids::COUNT_SKETCH], + &payload, + )) } fn from_msgpack(bytes: &[u8]) -> Result { - Ok(rmp_serde::from_slice(bytes)?) + let (kind_id, payload) = magic_ids::decode_wrapper(bytes) + .map_err(|msg| MsgPackError::Decode(rmp_serde::decode::Error::Uncategorized(msg)))?; + if kind_id != [magic_ids::COUNT_SKETCH] { + return Err(MsgPackError::BadMagicId { + expected: magic_ids::COUNT_SKETCH, + got: kind_id.first().copied(), + }); + } + Ok(rmp_serde::from_slice(payload)?) } } diff --git a/src/message_pack_format/portable/ddsketch.rs b/src/message_pack_format/portable/ddsketch.rs index 49414f3..c5cbc02 100644 --- a/src/message_pack_format/portable/ddsketch.rs +++ b/src/message_pack_format/portable/ddsketch.rs @@ -3,7 +3,7 @@ use serde::{Deserialize, Serialize}; -use crate::message_pack_format::{Error as MsgPackError, MessagePackCodec}; +use crate::message_pack_format::{Error as MsgPackError, MessagePackCodec, magic_ids}; /// Bucket-store growth chunk for the wire-format-aligned [`DdSketch`] /// variant. Matches the Go reference implementation's `DDSketch.GrowChunk` @@ -403,11 +403,20 @@ impl DdSketch { impl MessagePackCodec for DdSketch { fn to_msgpack(&self) -> Result, MsgPackError> { - Ok(rmp_serde::to_vec(self)?) + let payload = rmp_serde::to_vec(self)?; + Ok(magic_ids::encode_wrapper(&[magic_ids::DD_SKETCH], &payload)) } fn from_msgpack(bytes: &[u8]) -> Result { - Ok(rmp_serde::from_slice(bytes)?) + let (kind_id, payload) = magic_ids::decode_wrapper(bytes) + .map_err(|msg| MsgPackError::Decode(rmp_serde::decode::Error::Uncategorized(msg)))?; + if kind_id != [magic_ids::DD_SKETCH] { + return Err(MsgPackError::BadMagicId { + expected: magic_ids::DD_SKETCH, + got: kind_id.first().copied(), + }); + } + Ok(rmp_serde::from_slice(payload)?) } } @@ -539,16 +548,26 @@ mod tests { /// DataPoint-level METRIC scalars (count/sum/min/max). This pins the /// element count so the bytes stay parity-aligned with the Go /// reference implementation. + /// + /// The binary is wrapped in the ASAPv1 envelope; the payload starts after + /// the header (b"ASAPv1" + version + kind_id_len + kind_id). #[test] fn test_msgpack_is_three_element_array() { + use crate::message_pack_format::magic_ids; let sk = DdSketch::from_raw(0.01, vec![1, 2, 3], -2); let bytes = sk.to_msgpack().unwrap(); - // rmp compact encoding leads with an array marker. A fixarray of - // length 3 is the single byte 0x93 (0b1001_0011). + let (kind_id, payload) = magic_ids::decode_wrapper(&bytes).expect("ASAPv1 header"); + assert_eq!( + kind_id, + [magic_ids::DD_SKETCH], + "expected DD_SKETCH kind_id" + ); + // rmp compact encoding of the payload leads with an array marker. + // A fixarray of length 3 is the single byte 0x93 (0b1001_0011). assert_eq!( - bytes[0], 0x93, - "expected a 3-element msgpack fixarray (0x93), got {:#04x}", - bytes[0] + payload[0], 0x93, + "expected a 3-element msgpack fixarray (0x93) at payload[0], got {:#04x}", + payload[0] ); } @@ -581,13 +600,11 @@ mod tests { // P99 ≈ exp(mu + sigma * Φ⁻¹(0.99)) = e^(3 + 0.7×2.326) ≈ 102.4. assert!( (p50 / 20.09).ln().abs() < 0.05, - "P50 {} not close to 20.09", - p50 + "P50 {p50} not close to 20.09" ); assert!( (p99 / 102.4).ln().abs() < 0.05, - "P99 {} not close to 102.4", - p99 + "P99 {p99} not close to 102.4" ); } @@ -661,12 +678,7 @@ mod tests { let rel_err = (got / want - 1.0).abs(); assert!( rel_err <= alpha, - "q={} rel_err={:.4} exceeds α={}: reconstituted={}, full={}", - q, - rel_err, - alpha, - got, - want, + "q={q} rel_err={rel_err:.4} exceeds α={alpha}: reconstituted={got}, full={want}", ); } } diff --git a/src/message_pack_format/portable/delta_set_aggregator.rs b/src/message_pack_format/portable/delta_set_aggregator.rs index d26dbeb..46228f7 100644 --- a/src/message_pack_format/portable/delta_set_aggregator.rs +++ b/src/message_pack_format/portable/delta_set_aggregator.rs @@ -8,7 +8,7 @@ use std::collections::HashSet; use serde::{Deserialize, Serialize}; -use crate::message_pack_format::{Error as MsgPackError, MessagePackCodec}; +use crate::message_pack_format::{Error as MsgPackError, MessagePackCodec, magic_ids}; /// Wire DTO for the delta set aggregator: a snapshot of added/removed /// string keys between two consecutive observations. @@ -20,11 +20,23 @@ pub struct DeltaResult { impl MessagePackCodec for DeltaResult { fn to_msgpack(&self) -> Result, MsgPackError> { - Ok(rmp_serde::to_vec(self)?) + let payload = rmp_serde::to_vec(self)?; + Ok(magic_ids::encode_wrapper( + &[magic_ids::DELTA_RESULT], + &payload, + )) } fn from_msgpack(bytes: &[u8]) -> Result { - Ok(rmp_serde::from_slice(bytes)?) + let (kind_id, payload) = magic_ids::decode_wrapper(bytes) + .map_err(|msg| MsgPackError::Decode(rmp_serde::decode::Error::Uncategorized(msg)))?; + if kind_id != [magic_ids::DELTA_RESULT] { + return Err(MsgPackError::BadMagicId { + expected: magic_ids::DELTA_RESULT, + got: kind_id.first().copied(), + }); + } + Ok(rmp_serde::from_slice(payload)?) } } diff --git a/src/message_pack_format/portable/hll.rs b/src/message_pack_format/portable/hll.rs index 17b8686..949a7e6 100644 --- a/src/message_pack_format/portable/hll.rs +++ b/src/message_pack_format/portable/hll.rs @@ -8,7 +8,7 @@ use serde::{Deserialize, Serialize}; -use crate::message_pack_format::{Error as MsgPackError, MessagePackCodec}; +use crate::message_pack_format::{Error as MsgPackError, MessagePackCodec, magic_ids}; use crate::{CANONICAL_HASH_SEED, DataInput, hash64_seeded}; /// HLL estimator variant. Mirrors `asap_sketchlib::proto::sketchlib::HllVariant`. @@ -430,11 +430,20 @@ fn read_uvarint(buf: &[u8]) -> Option<(u64, usize)> { impl MessagePackCodec for HllSketch { fn to_msgpack(&self) -> Result, MsgPackError> { - Ok(rmp_serde::to_vec(self)?) + let payload = rmp_serde::to_vec(self)?; + Ok(magic_ids::encode_wrapper(&[magic_ids::HLL], &payload)) } fn from_msgpack(bytes: &[u8]) -> Result { - Ok(rmp_serde::from_slice(bytes)?) + let (kind_id, payload) = magic_ids::decode_wrapper(bytes) + .map_err(|msg| MsgPackError::Decode(rmp_serde::decode::Error::Uncategorized(msg)))?; + if kind_id != [magic_ids::HLL] { + return Err(MsgPackError::BadMagicId { + expected: magic_ids::HLL, + got: kind_id.first().copied(), + }); + } + Ok(rmp_serde::from_slice(payload)?) } } diff --git a/src/message_pack_format/portable/hydra_kll.rs b/src/message_pack_format/portable/hydra_kll.rs index b2d57e8..ee612f9 100644 --- a/src/message_pack_format/portable/hydra_kll.rs +++ b/src/message_pack_format/portable/hydra_kll.rs @@ -9,7 +9,7 @@ use serde::{Deserialize, Serialize}; use xxhash_rust::xxh32::xxh32; use crate::message_pack_format::portable::kll::{KllSketch, KllSketchData}; -use crate::message_pack_format::{Error as MsgPackError, MessagePackCodec}; +use crate::message_pack_format::{Error as MsgPackError, MessagePackCodec, magic_ids}; #[derive(Debug, Clone)] pub struct HydraKllSketch { @@ -157,14 +157,26 @@ impl MessagePackCodec for HydraKllSketch { cols: self.cols, sketches, }; - Ok(rmp_serde::to_vec(&wire)?) + let payload = rmp_serde::to_vec(&wire)?; + Ok(magic_ids::encode_wrapper( + &[magic_ids::HYDRA_KLL_SKETCH], + &payload, + )) } fn from_msgpack(bytes: &[u8]) -> Result { use crate::sketches::kll::KLL; use rmp_serde::decode::Error as RmpDecodeError; - let wire: HydraKllSketchWire = rmp_serde::from_slice(bytes)?; + let (kind_id, payload) = magic_ids::decode_wrapper(bytes) + .map_err(|msg| MsgPackError::Decode(rmp_serde::decode::Error::Uncategorized(msg)))?; + if kind_id != [magic_ids::HYDRA_KLL_SKETCH] { + return Err(MsgPackError::BadMagicId { + expected: magic_ids::HYDRA_KLL_SKETCH, + got: kind_id.first().copied(), + }); + } + let wire: HydraKllSketchWire = rmp_serde::from_slice(payload)?; if wire.sketches.len() != wire.rows { return Err(MsgPackError::Decode(RmpDecodeError::Uncategorized( diff --git a/src/message_pack_format/portable/kll.rs b/src/message_pack_format/portable/kll.rs index 55f5cf9..959e732 100644 --- a/src/message_pack_format/portable/kll.rs +++ b/src/message_pack_format/portable/kll.rs @@ -9,7 +9,7 @@ use serde::{Deserialize, Serialize}; -use crate::message_pack_format::{Error as MsgPackError, MessagePackCodec}; +use crate::message_pack_format::{Error as MsgPackError, MessagePackCodec, magic_ids}; use crate::sketches::kll::KLL; /// Concrete KLL type backing the wire-format `KllSketch`. @@ -347,11 +347,23 @@ impl MessagePackCodec for KllSketch { k: self.k, sketch_bytes: self.sketch_bytes(), }; - Ok(rmp_serde::to_vec(&wire)?) + let payload = rmp_serde::to_vec(&wire)?; + Ok(magic_ids::encode_wrapper( + &[magic_ids::KLL_SKETCH], + &payload, + )) } fn from_msgpack(bytes: &[u8]) -> Result { - let wire: KllSketchData = rmp_serde::from_slice(bytes)?; + let (kind_id, payload) = magic_ids::decode_wrapper(bytes) + .map_err(|msg| MsgPackError::Decode(rmp_serde::decode::Error::Uncategorized(msg)))?; + if kind_id != [magic_ids::KLL_SKETCH] { + return Err(MsgPackError::BadMagicId { + expected: magic_ids::KLL_SKETCH, + got: kind_id.first().copied(), + }); + } + let wire: KllSketchData = rmp_serde::from_slice(payload)?; let backend = KLL::deserialize_from_bytes(&wire.sketch_bytes)?; Ok(Self { k: wire.k, backend }) } diff --git a/src/message_pack_format/portable/set_aggregator.rs b/src/message_pack_format/portable/set_aggregator.rs index d346a9e..99a8cde 100644 --- a/src/message_pack_format/portable/set_aggregator.rs +++ b/src/message_pack_format/portable/set_aggregator.rs @@ -9,7 +9,7 @@ use std::collections::HashSet; use serde::{Deserialize, Serialize}; -use crate::message_pack_format::{Error as MsgPackError, MessagePackCodec}; +use crate::message_pack_format::{Error as MsgPackError, MessagePackCodec, magic_ids}; /// Set aggregator for tracking a set of unique string keys. #[derive(Debug, Clone)] @@ -81,11 +81,23 @@ impl MessagePackCodec for SetAggregator { let wrapper = StringSetRef { values: &self.values, }; - Ok(rmp_serde::to_vec(&wrapper)?) + let payload = rmp_serde::to_vec(&wrapper)?; + Ok(magic_ids::encode_wrapper( + &[magic_ids::SET_AGGREGATOR], + &payload, + )) } fn from_msgpack(bytes: &[u8]) -> Result { - let wrapper: StringSetOwned = rmp_serde::from_slice(bytes)?; + let (kind_id, payload) = magic_ids::decode_wrapper(bytes) + .map_err(|msg| MsgPackError::Decode(rmp_serde::decode::Error::Uncategorized(msg)))?; + if kind_id != [magic_ids::SET_AGGREGATOR] { + return Err(MsgPackError::BadMagicId { + expected: magic_ids::SET_AGGREGATOR, + got: kind_id.first().copied(), + }); + } + let wrapper: StringSetOwned = rmp_serde::from_slice(payload)?; Ok(Self { values: wrapper.values, }) @@ -154,8 +166,9 @@ mod tests { let mut sa = SetAggregator::new(); sa.update("a"); let bytes = sa.to_msgpack().unwrap(); + let (_, payload) = magic_ids::decode_wrapper(&bytes).expect("ASAPv1 header"); let decoded: StringSet = - rmp_serde::from_slice(&bytes).expect("should decode as StringSet { values: ... }"); + rmp_serde::from_slice(payload).expect("should decode as StringSet { values: ... }"); assert!(decoded.values.contains("a")); } } diff --git a/src/sketch_framework/hydra.rs b/src/sketch_framework/hydra.rs index 31f61aa..f90ee0a 100644 --- a/src/sketch_framework/hydra.rs +++ b/src/sketch_framework/hydra.rs @@ -160,14 +160,30 @@ impl Hydra { self.query_key(key, &HydraQuery::Cdf(threshold)) } - /// Serializes the Hydra sketch (including all counters) into MessagePack bytes. + /// Serializes the Hydra sketch into ASAPv1-wrapped MessagePack bytes. + /// kind_id: `[NATIVE_HYDRA, HASHER_UNKNOWN]`. pub fn serialize_to_bytes(&self) -> Result, RmpEncodeError> { - to_vec_named(self) + use crate::message_pack_format::magic_ids; + let payload = to_vec_named(self)?; + Ok(magic_ids::encode_wrapper( + &[magic_ids::NATIVE_HYDRA, magic_ids::HASHER_UNKNOWN], + &payload, + )) } - /// Deserializes a Hydra sketch from MessagePack bytes. + /// Deserializes a Hydra sketch from MessagePack bytes produced by [`Self::serialize_to_bytes`]. pub fn deserialize_from_bytes(bytes: &[u8]) -> Result { - from_slice(bytes) + use crate::message_pack_format::magic_ids; + let (kind_id, payload) = + magic_ids::decode_wrapper(bytes).map_err(RmpDecodeError::Uncategorized)?; + match kind_id { + [id, _hasher] if *id == magic_ids::NATIVE_HYDRA => from_slice(payload), + _ => Err(RmpDecodeError::Uncategorized(format!( + "Hydra kind_id mismatch: expected [0x{:02x}, hasher], got {:?}", + magic_ids::NATIVE_HYDRA, + kind_id + ))), + } } } @@ -291,8 +307,7 @@ impl MultiHeadHydra { } if std::mem::discriminant(counter) != std::mem::discriminant(other_counter) { return Err(format!( - "MultiHeadHydra counter type mismatch for dimension '{}'", - name + "MultiHeadHydra counter type mismatch for dimension '{name}'" )); } } @@ -715,8 +730,7 @@ mod tests { let quantile = hydra.query_key(vec!["metrics", "latency"], &HydraQuery::Cdf(30.0)); assert!( (quantile - 0.6).abs() < 1e-9, - "expected CDF near 0.6, got {}", - quantile + "expected CDF near 0.6, got {quantile}" ); let empty_bucket = hydra.query_key(vec!["other", "key"], &HydraQuery::Cdf(50.0)); @@ -835,8 +849,7 @@ mod tests { let card = result.unwrap(); assert!( card > 90.0 && card < 110.0, - "Expected approx 100, got {}", - card + "Expected approx 100, got {card}" ); } @@ -858,8 +871,7 @@ mod tests { let median = result.unwrap(); assert!( (median - 50.0).abs() < 5.0, - "Expected approx 50, got {}", - median + "Expected approx 50, got {median}" ); } diff --git a/src/sketch_framework/univmon.rs b/src/sketch_framework/univmon.rs index 9d22f23..d52cd5e 100644 --- a/src/sketch_framework/univmon.rs +++ b/src/sketch_framework/univmon.rs @@ -146,7 +146,7 @@ impl UnivMon { pub fn print_hh_layer(&self) { print!("Print HH_Layer: "); for i in 0..self.layer_size { - println!("layer {}: ", i); + println!("layer {i}: "); self.hh_layers[i].print_heap(); } } @@ -265,14 +265,30 @@ impl UnivMon { &mut self.hh_layers[layer] } - /// Serializes the UnivMon sketch into MessagePack bytes. + /// Serializes the UnivMon sketch into ASAPv1-wrapped MessagePack bytes. + /// kind_id: `[NATIVE_UNIVMON, HASHER_UNKNOWN]`. pub fn serialize_to_bytes(&self) -> Result, RmpEncodeError> { - to_vec_named(self) + use crate::message_pack_format::magic_ids; + let payload = to_vec_named(self)?; + Ok(magic_ids::encode_wrapper( + &[magic_ids::NATIVE_UNIVMON, magic_ids::HASHER_UNKNOWN], + &payload, + )) } - /// Deserializes a UnivMon sketch from MessagePack bytes. + /// Deserializes a UnivMon sketch from MessagePack bytes produced by [`Self::serialize_to_bytes`]. pub fn deserialize_from_bytes(bytes: &[u8]) -> Result { - from_slice(bytes) + use crate::message_pack_format::magic_ids; + let (kind_id, payload) = + magic_ids::decode_wrapper(bytes).map_err(RmpDecodeError::Uncategorized)?; + match kind_id { + [id, _hasher] if *id == magic_ids::NATIVE_UNIVMON => from_slice(payload), + _ => Err(RmpDecodeError::Uncategorized(format!( + "UnivMon kind_id mismatch: expected [0x{:02x}, hasher], got {:?}", + magic_ids::NATIVE_UNIVMON, + kind_id + ))), + } } } @@ -635,7 +651,7 @@ mod tests { for (prefix, count, repeat) in scenarios { for i in 0..repeat { - let key = format!("{}_{}", prefix, i); + let key = format!("{prefix}_{i}"); let val = count as i64; let val_f = val as f64; diff --git a/src/sketches/countminsketch.rs b/src/sketches/countminsketch.rs index 43f296f..888625d 100644 --- a/src/sketches/countminsketch.rs +++ b/src/sketches/countminsketch.rs @@ -260,18 +260,74 @@ impl CountMin { } } -// Serialization helpers for CountMin. -impl CountMin { - /// Serializes the sketch into MessagePack bytes. +// Serialization helpers for CountMin — one impl per Mode so the magic byte +// encodes both the sketch family and the hashing strategy. +// +// Wire layout: [ mode_id: u8 | hasher_id: u8 | ] + +impl CountMin { + /// Serializes the sketch into ASAPv1-wrapped MessagePack bytes. + /// kind_id: `[NATIVE_COUNT_MIN_REGULAR, hasher_id]`. pub fn serialize_to_bytes(&self) -> Result, RmpEncodeError> { - to_vec_named(self) + use crate::message_pack_format::magic_ids; + let payload = to_vec_named(self)?; + Ok(magic_ids::encode_wrapper( + &[magic_ids::NATIVE_COUNT_MIN_REGULAR, H::hasher_magic_id()], + &payload, + )) } } -impl Deserialize<'de>, Mode, H: SketchHasher> CountMin { - /// Deserializes a sketch from MessagePack bytes. +impl Deserialize<'de>, H: SketchHasher> CountMin { + /// Deserializes a sketch from bytes produced by [`Self::serialize_to_bytes`]. pub fn deserialize_from_bytes(bytes: &[u8]) -> Result { - from_slice(bytes) + use crate::message_pack_format::magic_ids; + let (kind_id, payload) = + magic_ids::decode_wrapper(bytes).map_err(RmpDecodeError::Uncategorized)?; + match kind_id { + [id, hasher] if *id == magic_ids::NATIVE_COUNT_MIN_REGULAR => { + magic_ids::check_hasher_id::(*hasher)?; + from_slice(payload) + } + _ => Err(RmpDecodeError::Uncategorized(format!( + "CountMin kind_id mismatch: expected [0x{:02x}, hasher], got {:?}", + magic_ids::NATIVE_COUNT_MIN_REGULAR, + kind_id + ))), + } + } +} + +impl CountMin { + /// Serializes the sketch into ASAPv1-wrapped MessagePack bytes. + /// kind_id: `[NATIVE_COUNT_MIN_FAST, hasher_id]`. + pub fn serialize_to_bytes(&self) -> Result, RmpEncodeError> { + use crate::message_pack_format::magic_ids; + let payload = to_vec_named(self)?; + Ok(magic_ids::encode_wrapper( + &[magic_ids::NATIVE_COUNT_MIN_FAST, H::hasher_magic_id()], + &payload, + )) + } +} + +impl Deserialize<'de>, H: SketchHasher> CountMin { + /// Deserializes a sketch from bytes produced by [`Self::serialize_to_bytes`]. + pub fn deserialize_from_bytes(bytes: &[u8]) -> Result { + use crate::message_pack_format::magic_ids; + let (kind_id, payload) = + magic_ids::decode_wrapper(bytes).map_err(RmpDecodeError::Uncategorized)?; + match kind_id { + [id, hasher] if *id == magic_ids::NATIVE_COUNT_MIN_FAST => { + magic_ids::check_hasher_id::(*hasher)?; + from_slice(payload) + } + _ => Err(RmpDecodeError::Uncategorized(format!( + "CountMin kind_id mismatch: expected [0x{:02x}, hasher], got {:?}", + magic_ids::NATIVE_COUNT_MIN_FAST, + kind_id + ))), + } } } @@ -959,4 +1015,50 @@ mod tests { decoded.as_storage().as_slice() ); } + + #[test] + fn count_min_fast_path_round_trip_serialization() { + use crate::message_pack_format::magic_ids; + + let mut sketch = CountMin::, FastPath>::with_dimensions(3, 8); + sketch.insert(&DataInput::U64(42)); + sketch.insert(&DataInput::U64(7)); + + let encoded = sketch + .serialize_to_bytes() + .expect("serialize CountMin FastPath"); + let (kind_id, _) = magic_ids::decode_wrapper(&encoded).expect("ASAPv1 header"); + assert_eq!( + kind_id, + [ + magic_ids::NATIVE_COUNT_MIN_FAST, + magic_ids::HASHER_DEFAULT_XX + ], + "wrong kind_id" + ); + + let decoded = CountMin::, FastPath>::deserialize_from_bytes(&encoded) + .expect("deserialize CountMin FastPath"); + + assert_eq!(sketch.rows(), decoded.rows()); + assert_eq!(sketch.cols(), decoded.cols()); + assert_eq!( + sketch.as_storage().as_slice(), + decoded.as_storage().as_slice() + ); + } + + #[test] + fn count_min_mode_mismatch_is_rejected() { + let mut sketch = CountMin::, FastPath>::with_dimensions(3, 8); + sketch.insert(&DataInput::U64(1)); + + let fast_bytes = sketch.serialize_to_bytes().unwrap(); + // Feeding FastPath bytes to a RegularPath decoder must error. + let result = CountMin::, RegularPath>::deserialize_from_bytes(&fast_bytes); + assert!( + result.is_err(), + "expected error decoding FastPath bytes as RegularPath" + ); + } } diff --git a/src/sketches/countsketch.rs b/src/sketches/countsketch.rs index 4c5cd53..162b262 100644 --- a/src/sketches/countsketch.rs +++ b/src/sketches/countsketch.rs @@ -274,27 +274,90 @@ where } } -// Serialization helpers for Count. -impl Count +// Serialization helpers for Count — one impl per Mode so the magic byte +// encodes both the sketch family and the hashing strategy. +// +// Wire layout: [ mode_id: u8 | hasher_id: u8 | ] + +impl Count where S: MatrixStorage + Serialize, C: CountSketchCounter, { - /// Serializes the sketch into MessagePack bytes. + /// Serializes the sketch into ASAPv1-wrapped MessagePack bytes. + /// kind_id: `[NATIVE_COUNT_SKETCH_REGULAR, hasher_id]`. pub fn serialize_to_bytes(&self) -> Result, RmpEncodeError> { - to_vec_named(self) + use crate::message_pack_format::magic_ids; + let payload = to_vec_named(self)?; + Ok(magic_ids::encode_wrapper( + &[magic_ids::NATIVE_COUNT_SKETCH_REGULAR, H::hasher_magic_id()], + &payload, + )) } } -// Deserialization helpers for Count. -impl Count +impl Count +where + S: MatrixStorage + for<'de> Deserialize<'de>, + C: CountSketchCounter, +{ + /// Deserializes a sketch from bytes produced by [`Self::serialize_to_bytes`]. + pub fn deserialize_from_bytes(bytes: &[u8]) -> Result { + use crate::message_pack_format::magic_ids; + let (kind_id, payload) = + magic_ids::decode_wrapper(bytes).map_err(RmpDecodeError::Uncategorized)?; + match kind_id { + [id, hasher] if *id == magic_ids::NATIVE_COUNT_SKETCH_REGULAR => { + magic_ids::check_hasher_id::(*hasher)?; + from_slice(payload) + } + _ => Err(RmpDecodeError::Uncategorized(format!( + "Count kind_id mismatch: expected [0x{:02x}, hasher], got {:?}", + magic_ids::NATIVE_COUNT_SKETCH_REGULAR, + kind_id + ))), + } + } +} + +impl Count +where + S: MatrixStorage + Serialize, + C: CountSketchCounter, +{ + /// Serializes the sketch into ASAPv1-wrapped MessagePack bytes. + /// kind_id: `[NATIVE_COUNT_SKETCH_FAST, hasher_id]`. + pub fn serialize_to_bytes(&self) -> Result, RmpEncodeError> { + use crate::message_pack_format::magic_ids; + let payload = to_vec_named(self)?; + Ok(magic_ids::encode_wrapper( + &[magic_ids::NATIVE_COUNT_SKETCH_FAST, H::hasher_magic_id()], + &payload, + )) + } +} + +impl Count where S: MatrixStorage + for<'de> Deserialize<'de>, C: CountSketchCounter, { - /// Deserializes a sketch from MessagePack bytes. + /// Deserializes a sketch from bytes produced by [`Self::serialize_to_bytes`]. pub fn deserialize_from_bytes(bytes: &[u8]) -> Result { - from_slice(bytes) + use crate::message_pack_format::magic_ids; + let (kind_id, payload) = + magic_ids::decode_wrapper(bytes).map_err(RmpDecodeError::Uncategorized)?; + match kind_id { + [id, hasher] if *id == magic_ids::NATIVE_COUNT_SKETCH_FAST => { + magic_ids::check_hasher_id::(*hasher)?; + from_slice(payload) + } + _ => Err(RmpDecodeError::Uncategorized(format!( + "Count kind_id mismatch: expected [0x{:02x}, hasher], got {:?}", + magic_ids::NATIVE_COUNT_SKETCH_FAST, + kind_id + ))), + } } } @@ -1100,4 +1163,49 @@ mod tests { decoded.as_storage().as_slice() ); } + + #[test] + fn count_sketch_fast_path_round_trip_serialization() { + use crate::message_pack_format::magic_ids; + + let mut sketch = Count::, FastPath>::with_dimensions(3, 8); + sketch.insert(&DataInput::U64(42)); + sketch.insert(&DataInput::U64(7)); + + let encoded = sketch + .serialize_to_bytes() + .expect("serialize Count FastPath"); + let (kind_id, _) = magic_ids::decode_wrapper(&encoded).expect("ASAPv1 header"); + assert_eq!( + kind_id, + [ + magic_ids::NATIVE_COUNT_SKETCH_FAST, + magic_ids::HASHER_DEFAULT_XX + ], + "wrong kind_id" + ); + + let decoded = Count::, FastPath>::deserialize_from_bytes(&encoded) + .expect("deserialize Count FastPath"); + + assert_eq!(sketch.rows(), decoded.rows()); + assert_eq!(sketch.cols(), decoded.cols()); + assert_eq!( + sketch.as_storage().as_slice(), + decoded.as_storage().as_slice() + ); + } + + #[test] + fn count_sketch_mode_mismatch_is_rejected() { + let mut sketch = Count::, FastPath>::with_dimensions(3, 8); + sketch.insert(&DataInput::U64(1)); + + let fast_bytes = sketch.serialize_to_bytes().unwrap(); + let result = Count::, RegularPath>::deserialize_from_bytes(&fast_bytes); + assert!( + result.is_err(), + "expected error decoding FastPath bytes as RegularPath" + ); + } } diff --git a/src/sketches/countsketch_topk.rs b/src/sketches/countsketch_topk.rs index c37e913..0928bea 100644 --- a/src/sketches/countsketch_topk.rs +++ b/src/sketches/countsketch_topk.rs @@ -1106,14 +1106,33 @@ impl CountL2HH { compute_median_inline_f64(&mut lst[..]) } - /// Serializes the CountL2HH sketch into MessagePack bytes. + /// Serializes the CountL2HH sketch into ASAPv1-wrapped MessagePack bytes. + /// kind_id: `[NATIVE_CMS_HEAP, hasher_id]`. pub fn serialize_to_bytes(&self) -> Result, RmpEncodeError> { - to_vec_named(self) + use crate::message_pack_format::magic_ids; + let payload = to_vec_named(self)?; + Ok(magic_ids::encode_wrapper( + &[magic_ids::NATIVE_CMS_HEAP, H::hasher_magic_id()], + &payload, + )) } - /// Deserializes a CountL2HH sketch from MessagePack bytes. + /// Deserializes a CountL2HH sketch from MessagePack bytes produced by [`Self::serialize_to_bytes`]. pub fn deserialize_from_bytes(bytes: &[u8]) -> Result { - from_slice(bytes) + use crate::message_pack_format::magic_ids; + let (kind_id, payload) = + magic_ids::decode_wrapper(bytes).map_err(RmpDecodeError::Uncategorized)?; + match kind_id { + [id, hasher] if *id == magic_ids::NATIVE_CMS_HEAP => { + magic_ids::check_hasher_id::(*hasher)?; + from_slice(payload) + } + _ => Err(RmpDecodeError::Uncategorized(format!( + "CountL2HH kind_id mismatch: expected [0x{:02x}, hasher], got {:?}", + magic_ids::NATIVE_CMS_HEAP, + kind_id + ))), + } } } diff --git a/src/sketches/ddsketch.rs b/src/sketches/ddsketch.rs index c213275..347421e 100644 --- a/src/sketches/ddsketch.rs +++ b/src/sketches/ddsketch.rs @@ -165,12 +165,19 @@ impl DDSketch { } } - /// Serializes the sketch to a MessagePack byte vector. + /// Serializes the sketch to ASAPv1-wrapped MessagePack bytes. + /// kind_id: `[NATIVE_DD_SKETCH, HASHER_UNKNOWN]`. pub fn serialize_to_bytes(&self) -> Result, RmpEncodeError> { - to_vec_named(self) + use crate::message_pack_format::magic_ids; + let payload = to_vec_named(self)?; + Ok(magic_ids::encode_wrapper( + &[magic_ids::NATIVE_DD_SKETCH, magic_ids::HASHER_UNKNOWN], + &payload, + )) } - /// Deserializes a DDSketch from a MessagePack byte slice. + /// Deserializes a DDSketch from a MessagePack byte slice produced by + /// [`Self::serialize_to_bytes`]. /// /// The `count`/`sum`/`min`/`max` scalars are `#[serde(skip)]` (dropped /// from the wire, ProjectASAP/sketchlib-go#243), so they default to @@ -179,9 +186,21 @@ impl DDSketch { /// reconstructed from the per-bucket representative values, accurate /// to within the sketch's α relative-accuracy bound. pub fn deserialize_from_bytes(bytes: &[u8]) -> Result { - let mut sk: Self = from_slice(bytes)?; - sk.recompute_scalars_from_store(); - Ok(sk) + use crate::message_pack_format::magic_ids; + let (kind_id, payload) = + magic_ids::decode_wrapper(bytes).map_err(RmpDecodeError::Uncategorized)?; + match kind_id { + [id, _hasher] if *id == magic_ids::NATIVE_DD_SKETCH => { + let mut sk: Self = from_slice(payload)?; + sk.recompute_scalars_from_store(); + Ok(sk) + } + _ => Err(RmpDecodeError::Uncategorized(format!( + "DDSketch kind_id mismatch: expected [0x{:02x}, hasher], got {:?}", + magic_ids::NATIVE_DD_SKETCH, + kind_id + ))), + } } /// Rebuild the in-memory `count`/`sum`/`min`/`max` aggregates from the @@ -523,13 +542,7 @@ mod tests { let err = rel_err(got, want); assert!( err <= tol, - "quantile {} (p={:.2}) relerr={:.4} got={} want={} tol={}", - name, - p, - err, - got, - want, - tol + "quantile {name} (p={p:.2}) relerr={err:.4} got={got} want={want} tol={tol}" ); } } @@ -590,13 +603,7 @@ mod tests { let err = rel_err(got, want); assert!( err <= tol, - "quantile {} (p={:.2}) relerr={:.4} got={} want={} tol={}", - name, - p, - err, - got, - want, - tol + "quantile {name} (p={p:.2}) relerr={err:.4} got={got} want={want} tol={tol}" ); } } @@ -660,13 +667,7 @@ mod tests { let err = rel_err(got, want); assert!( err <= tol, - "quantile {} (p={:.2}) relerr={:.4} got={} want={} tol={}", - name, - p, - err, - got, - want, - tol + "quantile {name} (p={p:.2}) relerr={err:.4} got={got} want={want} tol={tol}" ); } } @@ -727,13 +728,7 @@ mod tests { let err = rel_err(got, want); assert!( err <= tol + 1e-9, - "quantile {} (p={:.2}) relerr={:.4} got={} want={} tol={}", - name, - p, - err, - got, - want, - tol + "quantile {name} (p={p:.2}) relerr={err:.4} got={got} want={want} tol={tol}" ); } } diff --git a/src/sketches/hll.rs b/src/sketches/hll.rs index dcdbae3..7c4fad7 100644 --- a/src/sketches/hll.rs +++ b/src/sketches/hll.rs @@ -116,16 +116,6 @@ impl } } - /// Serializes the sketch into MessagePack bytes. - pub fn serialize_to_bytes(&self) -> Result, RmpEncodeError> { - to_vec_named(self) - } - - /// Deserializes a sketch from MessagePack bytes. - pub fn deserialize_from_bytes(bytes: &[u8]) -> Result { - from_slice(bytes) - } - /// Borrow the raw register byte slice (one byte per register). pub fn registers_as_slice(&self) -> &[u8] { self.registers.as_slice() @@ -170,6 +160,83 @@ impl } } +// Serialization — split by Variant so the magic byte encodes the estimator. +// Wire layout: [ variant_id: u8 | hasher_id: u8 | ] + +impl + HyperLogLogImpl +{ + /// Serializes the sketch into ASAPv1-wrapped MessagePack bytes. + /// kind_id: `[NATIVE_HLL_CLASSIC, hasher_id]`. + pub fn serialize_to_bytes(&self) -> Result, RmpEncodeError> { + use crate::message_pack_format::magic_ids; + let payload = to_vec_named(self)?; + Ok(magic_ids::encode_wrapper( + &[magic_ids::NATIVE_HLL_CLASSIC, H::hasher_magic_id()], + &payload, + )) + } +} + +impl Deserialize<'de>, H: SketchHasher> + HyperLogLogImpl +{ + /// Deserializes a sketch from bytes produced by [`Self::serialize_to_bytes`]. + pub fn deserialize_from_bytes(bytes: &[u8]) -> Result { + use crate::message_pack_format::magic_ids; + let (kind_id, payload) = + magic_ids::decode_wrapper(bytes).map_err(RmpDecodeError::Uncategorized)?; + match kind_id { + [id, hasher] if *id == magic_ids::NATIVE_HLL_CLASSIC => { + magic_ids::check_hasher_id::(*hasher)?; + from_slice(payload) + } + _ => Err(RmpDecodeError::Uncategorized(format!( + "HyperLogLogImpl kind_id mismatch: expected [0x{:02x}, hasher], got {:?}", + magic_ids::NATIVE_HLL_CLASSIC, + kind_id + ))), + } + } +} + +impl + HyperLogLogImpl +{ + /// Serializes the sketch into ASAPv1-wrapped MessagePack bytes. + /// kind_id: `[NATIVE_HLL_ERTL_MLE, hasher_id]`. + pub fn serialize_to_bytes(&self) -> Result, RmpEncodeError> { + use crate::message_pack_format::magic_ids; + let payload = to_vec_named(self)?; + Ok(magic_ids::encode_wrapper( + &[magic_ids::NATIVE_HLL_ERTL_MLE, H::hasher_magic_id()], + &payload, + )) + } +} + +impl Deserialize<'de>, H: SketchHasher> + HyperLogLogImpl +{ + /// Deserializes a sketch from bytes produced by [`Self::serialize_to_bytes`]. + pub fn deserialize_from_bytes(bytes: &[u8]) -> Result { + use crate::message_pack_format::magic_ids; + let (kind_id, payload) = + magic_ids::decode_wrapper(bytes).map_err(RmpDecodeError::Uncategorized)?; + match kind_id { + [id, hasher] if *id == magic_ids::NATIVE_HLL_ERTL_MLE => { + magic_ids::check_hasher_id::(*hasher)?; + from_slice(payload) + } + _ => Err(RmpDecodeError::Uncategorized(format!( + "HyperLogLogImpl kind_id mismatch: expected [0x{:02x}, hasher], got {:?}", + magic_ids::NATIVE_HLL_ERTL_MLE, + kind_id + ))), + } + } +} + // DataInput adapters (hashing + batch helpers). impl HyperLogLogImpl @@ -369,14 +436,39 @@ impl HyperLogLogHIPImpl { self.est as usize } - /// Serializes the sketch into MessagePack bytes. + /// Serializes the sketch into ASAPv1-wrapped MessagePack bytes. + /// kind_id: `[NATIVE_HLL_HIP, HASHER_DEFAULT_XX]` — HIP always uses DefaultXxHasher. pub fn serialize_to_bytes(&self) -> Result, RmpEncodeError> { - to_vec_named(self) + use crate::message_pack_format::magic_ids; + let payload = to_vec_named(self)?; + Ok(magic_ids::encode_wrapper( + &[magic_ids::NATIVE_HLL_HIP, magic_ids::HASHER_DEFAULT_XX], + &payload, + )) } - /// Deserializes a sketch from MessagePack bytes. + /// Deserializes a sketch from bytes produced by [`Self::serialize_to_bytes`]. pub fn deserialize_from_bytes(bytes: &[u8]) -> Result { - from_slice(bytes) + use crate::message_pack_format::magic_ids; + let (kind_id, payload) = + magic_ids::decode_wrapper(bytes).map_err(RmpDecodeError::Uncategorized)?; + match kind_id { + [id, hasher] if *id == magic_ids::NATIVE_HLL_HIP => { + if *hasher != magic_ids::HASHER_DEFAULT_XX && *hasher != magic_ids::HASHER_UNKNOWN { + return Err(RmpDecodeError::Uncategorized(format!( + "HyperLogLogHIPImpl hasher mismatch: expected 0x{:02x}, got 0x{:02x}", + magic_ids::HASHER_DEFAULT_XX, + hasher + ))); + } + from_slice(payload) + } + _ => Err(RmpDecodeError::Uncategorized(format!( + "HyperLogLogHIPImpl kind_id mismatch: expected [0x{:02x}, hasher], got {:?}", + magic_ids::NATIVE_HLL_HIP, + kind_id + ))), + } } } @@ -675,6 +767,40 @@ mod tests { assert_serialization_round_trip::("HllDsP12"); } + #[test] + fn hll_magic_bytes_are_variant_specific() { + use crate::message_pack_format::magic_ids; + + let classic_bytes = HyperLogLog::::default() + .serialize_to_bytes() + .unwrap(); + let ertl_bytes = HyperLogLog::::default() + .serialize_to_bytes() + .unwrap(); + + let (classic_kind_id, _) = + magic_ids::decode_wrapper(&classic_bytes).expect("ASAPv1 header"); + let (ertl_kind_id, _) = magic_ids::decode_wrapper(&ertl_bytes).expect("ASAPv1 header"); + + assert_eq!(classic_kind_id[0], magic_ids::NATIVE_HLL_CLASSIC); + assert_eq!(ertl_kind_id[0], magic_ids::NATIVE_HLL_ERTL_MLE); + // Sanity: they are different. + assert_ne!(classic_kind_id[0], ertl_kind_id[0]); + } + + #[test] + fn hll_variant_mismatch_is_rejected() { + let ertl_bytes = HyperLogLog::::default() + .serialize_to_bytes() + .unwrap(); + // Feeding ErtlMLE bytes to a Classic decoder must error. + let result = HyperLogLog::::deserialize_from_bytes(&ertl_bytes); + assert!( + result.is_err(), + "expected error decoding ErtlMLE bytes as Classic" + ); + } + // insert 10 values and check corresponding counter is updated #[test] fn hll_correctness_test() { diff --git a/src/sketches/kll.rs b/src/sketches/kll.rs index ee11b3e..9f5d6fa 100644 --- a/src/sketches/kll.rs +++ b/src/sketches/kll.rs @@ -662,20 +662,37 @@ impl KLL { // -- Serialization ------------------------------------------------------- - /// Serializes the sketch to a MessagePack byte vector. + /// Serializes the sketch into ASAPv1-wrapped MessagePack bytes. + /// kind_id: `[NATIVE_KLL, HASHER_UNKNOWN]`. pub fn serialize_to_bytes(&self) -> Result, RmpEncodeError> where T: Serialize, { - rmp_serde::to_vec(self) + use crate::message_pack_format::magic_ids; + let payload = rmp_serde::to_vec(self)?; + Ok(magic_ids::encode_wrapper( + &[magic_ids::NATIVE_KLL, magic_ids::HASHER_UNKNOWN], + &payload, + )) } - /// Deserializes a KLL sketch from a MessagePack byte slice. + /// Deserializes a KLL sketch from a MessagePack byte slice produced by + /// [`Self::serialize_to_bytes`]. pub fn deserialize_from_bytes(bytes: &[u8]) -> Result where T: for<'de> Deserialize<'de>, { - rmp_serde::from_slice(bytes) + use crate::message_pack_format::magic_ids; + let (kind_id, payload) = + magic_ids::decode_wrapper(bytes).map_err(rmp_serde::decode::Error::Uncategorized)?; + match kind_id { + [id, _hasher] if *id == magic_ids::NATIVE_KLL => rmp_serde::from_slice(payload), + _ => Err(rmp_serde::decode::Error::Uncategorized(format!( + "KLL kind_id mismatch: expected [0x{:02x}, hasher], got {:?}", + magic_ids::NATIVE_KLL, + kind_id + ))), + } } fn ensure_levels_sorted(&mut self) { @@ -1171,7 +1188,7 @@ mod tests { let median = cdf.query(0.5); // Median should be 30.5 - assert!(median > 20.0 && median < 40.2, "Median = {}", median); + assert!(median > 20.0 && median < 40.2, "Median = {median}"); // Test error handling for non-numeric input let result = kll.update_data_input(&DataInput::String("not a number".to_string())); @@ -1201,7 +1218,7 @@ mod tests { // cdf.print_entries(); let median = cdf.query(0.5); // only 30 and 40 is possible - assert!(median == 30.0 || median == 40.0, "Median = {}", median); + assert!(median == 30.0 || median == 40.0, "Median = {median}"); } #[test] @@ -1226,7 +1243,7 @@ mod tests { // kll.print_compactors(); let median = cdf.query(0.5); // Median should be 30 - assert!(median == 30.0, "Median = {}", median); + assert!(median == 30.0, "Median = {median}"); } #[test] diff --git a/src/sketches/kll_dynamic.rs b/src/sketches/kll_dynamic.rs index e48f2eb..c7a4885 100644 --- a/src/sketches/kll_dynamic.rs +++ b/src/sketches/kll_dynamic.rs @@ -353,23 +353,40 @@ impl KLLDynamic { self.items.len() } - /// Serialize the sketch into MessagePack bytes. + /// Serialize the sketch into ASAPv1-wrapped MessagePack bytes. + /// kind_id: `[NATIVE_KLL_DYNAMIC, HASHER_UNKNOWN]`. pub fn serialize_to_bytes(&self) -> Result, RmpEncodeError> where T: Serialize, { - rmp_serde::to_vec(self) + use crate::message_pack_format::magic_ids; + let payload = rmp_serde::to_vec(self)?; + Ok(magic_ids::encode_wrapper( + &[magic_ids::NATIVE_KLL_DYNAMIC, magic_ids::HASHER_UNKNOWN], + &payload, + )) } - /// Deserialize a sketch from MessagePack bytes. + /// Deserialize a sketch from MessagePack bytes produced by [`Self::serialize_to_bytes`]. pub fn deserialize_from_bytes(bytes: &[u8]) -> Result where T: for<'de> Deserialize<'de>, { - rmp_serde::from_slice(bytes).map(|mut sketch: KLLDynamic| { - sketch.rebuild_capacity_cache(); - sketch - }) + use crate::message_pack_format::magic_ids; + let (kind_id, payload) = + magic_ids::decode_wrapper(bytes).map_err(rmp_serde::decode::Error::Uncategorized)?; + match kind_id { + [id, _hasher] if *id == magic_ids::NATIVE_KLL_DYNAMIC => rmp_serde::from_slice(payload) + .map(|mut sketch: KLLDynamic| { + sketch.rebuild_capacity_cache(); + sketch + }), + _ => Err(rmp_serde::decode::Error::Uncategorized(format!( + "KLLDynamic kind_id mismatch: expected [0x{:02x}, hasher], got {:?}", + magic_ids::NATIVE_KLL_DYNAMIC, + kind_id + ))), + } } } @@ -629,7 +646,7 @@ mod tests { let median = cdf.query(0.5); // Median should be 30.5 - assert!(median > 20.0 && median < 40.2, "Median = {}", median); + assert!(median > 20.0 && median < 40.2, "Median = {median}"); // Test error handling for non-numeric input let result = kll.update_data_input(&DataInput::String("not a number".to_string())); @@ -652,7 +669,7 @@ mod tests { let cdf = kll.cdf(); let median = cdf.query(0.5); // only 30 and 40 is possible - assert!(median == 30.0 || median == 40.0, "Median = {}", median); + assert!(median == 30.0 || median == 40.0, "Median = {median}"); } #[test] @@ -669,7 +686,7 @@ mod tests { let cdf = kll.cdf(); let median = cdf.query(0.5); // Median should be 30 - assert!(median == 30.0, "Median = {}", median); + assert!(median == 30.0, "Median = {median}"); } #[test] diff --git a/src/sketches/kmv.rs b/src/sketches/kmv.rs index 1517a14..57f792c 100644 --- a/src/sketches/kmv.rs +++ b/src/sketches/kmv.rs @@ -77,14 +77,33 @@ impl KMV { } } - /// Serializes the sketch into MessagePack bytes. + /// Serializes the sketch into ASAPv1-wrapped MessagePack bytes. + /// kind_id: `[NATIVE_KMV, H::hasher_magic_id()]`. pub fn serialize_to_bytes(&self) -> Result, RmpEncodeError> { - to_vec_named(self) + use crate::message_pack_format::magic_ids; + let payload = to_vec_named(self)?; + Ok(magic_ids::encode_wrapper( + &[magic_ids::NATIVE_KMV, H::hasher_magic_id()], + &payload, + )) } - /// Deserializes a sketch from MessagePack bytes. + /// Deserializes a sketch from MessagePack bytes produced by [`Self::serialize_to_bytes`]. pub fn deserialize_from_bytes(bytes: &[u8]) -> Result { - from_slice(bytes) + use crate::message_pack_format::magic_ids; + let (kind_id, payload) = + magic_ids::decode_wrapper(bytes).map_err(RmpDecodeError::Uncategorized)?; + match kind_id { + [id, hasher] if *id == magic_ids::NATIVE_KMV => { + magic_ids::check_hasher_id::(*hasher)?; + from_slice(payload) + } + _ => Err(RmpDecodeError::Uncategorized(format!( + "KMV kind_id mismatch: expected [0x{:02x}, hasher], got {:?}", + magic_ids::NATIVE_KMV, + kind_id + ))), + } } } diff --git a/tests/xtest_consumer.rs b/tests/xtest_consumer.rs index ce9364e..ea9332a 100644 --- a/tests/xtest_consumer.rs +++ b/tests/xtest_consumer.rs @@ -62,7 +62,7 @@ fn cross_language_proto() { let cm_state = match env.sketch_state { Some(sketch_envelope::SketchState::CountMin(ref s)) => s.clone(), - other => panic!("expected CountMin sketch_state, got {:?}", other), + other => panic!("expected CountMin sketch_state, got {other:?}"), }; println!( @@ -96,10 +96,7 @@ fn cross_language_proto() { let hot_key = b"item:42" as &[u8]; let hot_hash = xxh3_64_seeded(SEED_0, hot_key); - println!( - "[CountMin] Step 3/3 — Point query 'item:42' (hash=0x{:016x})", - hot_hash - ); + println!("[CountMin] Step 3/3 — Point query 'item:42' (hash=0x{hot_hash:016x})"); let bits_per_row = col_bits(cols); let mask = (cols as u64) - 1; @@ -112,12 +109,9 @@ fn cross_language_proto() { min_freq = v; } } - println!( - "[CountMin] min frequency estimate = {:.0} (expect ≥ 101)", - min_freq - ); + println!("[CountMin] min frequency estimate = {min_freq:.0} (expect ≥ 101)"); if min_freq < 101.0 { - eprintln!("[CountMin] FAIL: frequency {:.0} < 101", min_freq); + eprintln!("[CountMin] FAIL: frequency {min_freq:.0} < 101"); all_ok = false; } else { println!("[CountMin] PASS"); @@ -139,7 +133,7 @@ fn cross_language_proto() { let kll_state = match env.sketch_state { Some(sketch_envelope::SketchState::Kll(ref s)) => s.clone(), - other => panic!("expected KLL sketch_state, got {:?}", other), + other => panic!("expected KLL sketch_state, got {other:?}"), }; println!( @@ -154,17 +148,14 @@ fn cross_language_proto() { let kll = KllFromProto::from_state(&kll_state); let p50 = kll.quantile(0.50); let p99 = kll.quantile(0.99); - println!( - "[KLL] Step 3/3 — p50 ≈ {:.1} (expect ~5000) p99 ≈ {:.1} (expect ~9900)", - p50, p99 - ); + println!("[KLL] Step 3/3 — p50 ≈ {p50:.1} (expect ~5000) p99 ≈ {p99:.1} (expect ~9900)"); let p50_ok = (p50 - 5000.0).abs() / 5000.0 < 0.05; let p99_ok = (p99 - 9900.0).abs() / 9900.0 < 0.05; if p50_ok && p99_ok { println!("[KLL] PASS"); } else { - eprintln!("[KLL] FAIL: p50={:.1} p99={:.1}", p50, p99); + eprintln!("[KLL] FAIL: p50={p50:.1} p99={p99:.1}"); all_ok = false; } @@ -184,7 +175,7 @@ fn cross_language_proto() { let dd_state = match env.sketch_state { Some(sketch_envelope::SketchState::Ddsketch(ref s)) => s.clone(), - other => panic!("expected DDSketch sketch_state, got {:?}", other), + other => panic!("expected DDSketch sketch_state, got {other:?}"), }; // `count` was dropped from the wire (ProjectASAP/sketchlib-go#243); @@ -216,7 +207,7 @@ fn cross_language_proto() { if p50_ok && p99_ok { println!("[DDSketch] PASS"); } else { - eprintln!("[DDSketch] FAIL: p50={:?} p99={:?}", p50_dd, p99_dd); + eprintln!("[DDSketch] FAIL: p50={p50_dd:?} p99={p99_dd:?}"); all_ok = false; } @@ -236,7 +227,7 @@ fn cross_language_proto() { let hll_state = match env.sketch_state { Some(sketch_envelope::SketchState::Hll(ref s)) => s.clone(), - other => panic!("expected HLL sketch_state, got {:?}", other), + other => panic!("expected HLL sketch_state, got {other:?}"), }; println!( @@ -247,14 +238,11 @@ fn cross_language_proto() { ); let hll_card = hll_ertl_mle_estimate(&hll_state); - println!( - "[HLL] Step 3/3 — cardinality ≈ {} (expect ~50000)", - hll_card - ); + println!("[HLL] Step 3/3 — cardinality ≈ {hll_card} (expect ~50000)"); if (40_000..=65_000).contains(&hll_card) { println!("[HLL] PASS"); } else { - eprintln!("[HLL] FAIL: cardinality {} not in [40000, 65000]", hll_card); + eprintln!("[HLL] FAIL: cardinality {hll_card} not in [40000, 65000]"); all_ok = false; } @@ -274,7 +262,7 @@ fn cross_language_proto() { let cs_state = match env.sketch_state { Some(sketch_envelope::SketchState::CountSketch(ref s)) => s.clone(), - other => panic!("expected CountSketch sketch_state, got {:?}", other), + other => panic!("expected CountSketch sketch_state, got {other:?}"), }; println!( @@ -288,14 +276,11 @@ fn cross_language_proto() { // 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 = {:.0} (expect ≥ 200)", - cs_est - ); + 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 {:.0} < 200", cs_est); + eprintln!("[CountSketch] FAIL: estimate {cs_est:.0} < 200"); all_ok = false; } @@ -315,7 +300,7 @@ fn cross_language_proto() { let coco_state = match env.sketch_state { Some(sketch_envelope::SketchState::Coco(ref s)) => s.clone(), - other => panic!("expected CocoSketch sketch_state, got {:?}", other), + other => panic!("expected CocoSketch sketch_state, got {other:?}"), }; println!( @@ -330,14 +315,11 @@ fn cross_language_proto() { // DeriveIndex(hash, row, width): col = (hash >> (row * maskBitsForWidth(width))) & (width-1) let coco_hash = xxh3_64_seeded(SEED_0, b"coco:hot"); let coco_est = coco_estimate(&coco_state, coco_hash); - println!( - "[CocoSketch] Step 3/3 — 'coco:hot' est = {} (expect ≥ 500)", - coco_est - ); + println!("[CocoSketch] Step 3/3 — 'coco:hot' est = {coco_est} (expect ≥ 500)"); if coco_est >= 500 { println!("[CocoSketch] PASS"); } else { - eprintln!("[CocoSketch] FAIL: estimate {} < 500", coco_est); + eprintln!("[CocoSketch] FAIL: estimate {coco_est} < 500"); all_ok = false; } @@ -357,7 +339,7 @@ fn cross_language_proto() { let elastic_state = match env.sketch_state { Some(sketch_envelope::SketchState::Elastic(ref s)) => s.clone(), - other => panic!("expected ElasticSketch sketch_state, got {:?}", other), + other => panic!("expected ElasticSketch sketch_state, got {other:?}"), }; println!( @@ -371,14 +353,11 @@ fn cross_language_proto() { // Go uses CanonicalHashSeed = seedList[5] = 0x6a09e667 let elephant_hash = xxh3_64_seeded(SEED_5, b"elephant"); let elephant_est = elastic_query(&elastic_state, "elephant", elephant_hash); - println!( - "[ElasticSketch] Step 3/3 — 'elephant' est = {} (expect ≥ 900)", - elephant_est - ); + println!("[ElasticSketch] Step 3/3 — 'elephant' est = {elephant_est} (expect ≥ 900)"); if elephant_est >= 900 { println!("[ElasticSketch] PASS"); } else { - eprintln!("[ElasticSketch] FAIL: estimate {} < 900", elephant_est); + eprintln!("[ElasticSketch] FAIL: estimate {elephant_est} < 900"); all_ok = false; } @@ -398,7 +377,7 @@ fn cross_language_proto() { let um_state = match env.sketch_state { Some(sketch_envelope::SketchState::Univmon(ref s)) => s.clone(), - other => panic!("expected UnivMon sketch_state, got {:?}", other), + other => panic!("expected UnivMon sketch_state, got {other:?}"), }; println!( @@ -409,17 +388,11 @@ fn cross_language_proto() { let um_card = univmon_cardinality(&um_state); // Note: the g-sum heuristic typically underestimates (Go itself reports ~4250 for 10k inserts). // We verify the Rust result matches Go's algorithm rather than the true cardinality. - println!( - "[UnivMon] Step 3/3 — cardinality ≈ {:.0} (g-sum heuristic, Go also ~4250)", - um_card - ); + println!("[UnivMon] Step 3/3 — cardinality ≈ {um_card:.0} (g-sum heuristic, Go also ~4250)"); if (1_000.0..=15_000.0).contains(&um_card) { println!("[UnivMon] PASS"); } else { - eprintln!( - "[UnivMon] FAIL: cardinality {:.0} not in [1000, 15000]", - um_card - ); + eprintln!("[UnivMon] FAIL: cardinality {um_card:.0} not in [1000, 15000]"); all_ok = false; } @@ -439,7 +412,7 @@ fn cross_language_proto() { let hydra_state = match env.sketch_state { Some(sketch_envelope::SketchState::Hydra(ref s)) => s.clone(), - other => panic!("expected Hydra sketch_state, got {:?}", other), + other => panic!("expected Hydra sketch_state, got {other:?}"), }; println!( @@ -456,14 +429,11 @@ fn cross_language_proto() { let hydra_subkey_hash = xxh3_64_seeded(SEED_6, b"hydra:42"); let hydra_value_hash = xxh3_64_seeded(SEED_0, b"hydra:42"); let hydra_est = hydra_query_cm(&hydra_state, hydra_subkey_hash, hydra_value_hash); - println!( - "[Hydra] Step 3/3 — 'hydra:42' est = {:.0} (expect ≥ 51)", - hydra_est - ); + println!("[Hydra] Step 3/3 — 'hydra:42' est = {hydra_est:.0} (expect ≥ 51)"); if hydra_est >= 51.0 { println!("[Hydra] PASS"); } else { - eprintln!("[Hydra] FAIL: estimate {:.0} < 51", hydra_est); + eprintln!("[Hydra] FAIL: estimate {hydra_est:.0} < 51"); all_ok = false; } @@ -488,7 +458,7 @@ fn cross_language_proto() { let cm_state = match env.sketch_state { Some(sketch_envelope::SketchState::CountMin(ref s)) => s.clone(), - other => panic!("expected CountMin sketch_state, got {:?}", other), + other => panic!("expected CountMin sketch_state, got {other:?}"), }; let rows = cm_state.rows as usize; let cols = cm_state.cols as usize; @@ -541,7 +511,7 @@ fn cross_language_proto() { let mut hll_state = match env.sketch_state { Some(sketch_envelope::SketchState::Hll(ref s)) => s.clone(), - other => panic!("expected HLL sketch_state, got {:?}", other), + other => panic!("expected HLL sketch_state, got {other:?}"), }; // Dual-read the registers (sampling thins them → may be sparse-encoded). let regs =