diff --git a/CHANGELOG.md b/CHANGELOG.md index 545914c5..9f68d7e3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,7 @@ All significant changes to this project will be documented in this file. ### New features +* Use serial version 2 for new `CountMinSketch` instances, enabling cross-language interchange with the documented input encodings. New sketches support `1..=127` rows and require a version 2 reader. Existing version 1 data remains readable and updatable, but must be rebuilt from source items to merge with new sketches. * Add `CpcUnion::serialize`, which produces the same bytes as `to_sketch().serialize()` and can be faster. * Add `HllUnion::serialize` to serialize union results directly. * Add `XorFilter` and `XorFilterBuilder` behind the `xor` feature for immutable membership queries on values or precomputed hashes. Filters support 8-, 16-, and 32-bit fingerprints and can be serialized for later queries. diff --git a/datasketches/src/countmin/mod.rs b/datasketches/src/countmin/mod.rs index d8e4f5de..03c92838 100644 --- a/datasketches/src/countmin/mod.rs +++ b/datasketches/src/countmin/mod.rs @@ -20,15 +20,19 @@ //! The CountMin sketch provides approximate frequency counts for streaming data //! with configurable relative error and confidence bounds. //! +//! Serial version 2 sketches with `i64` counters support cross-language binary interchange when +//! items use compatible hashing; see [`CountMinSketch`] and [`hash::value`](crate::hash::value). +//! //! # Usage //! //! ``` //! use datasketches::countmin::CountMinSketch; +//! use datasketches::hash::value::raw_bytes; //! //! let mut sketch = CountMinSketch::::new(5, 256).unwrap(); -//! sketch.update("apple"); -//! sketch.update_with_weight("banana", 3); -//! assert!(sketch.estimate("banana") >= 3); +//! sketch.update(raw_bytes::from_str("apple")); +//! sketch.update_with_weight(raw_bytes::from_str("banana"), 3); +//! assert!(sketch.estimate(raw_bytes::from_str("banana")) >= 3); //! ``` //! //! # Configuration Helpers @@ -39,11 +43,9 @@ //! let buckets = CountMinSketch::::suggest_num_buckets(0.01).unwrap(); //! let hashes = CountMinSketch::::suggest_num_hashes(0.99).unwrap(); //! let sketch = CountMinSketch::::new(hashes, buckets).unwrap(); -//! assert_eq!(sketch.estimate("apple"), 0); +//! assert_eq!(sketch.estimate(1i64), 0); //! ``` -mod serialization; - mod sketch; pub use self::sketch::CountMinSketch; diff --git a/datasketches/src/countmin/serialization.rs b/datasketches/src/countmin/serialization.rs deleted file mode 100644 index 7a930e01..00000000 --- a/datasketches/src/countmin/serialization.rs +++ /dev/null @@ -1,21 +0,0 @@ -// Licensed to the Apache Software Foundation (ASF) under one -// or more contributor license agreements. See the NOTICE file -// distributed with this work for additional information -// regarding copyright ownership. The ASF licenses this file -// to you under the Apache License, Version 2.0 (the -// "License"); you may not use this file except in compliance -// with the License. You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, -// software distributed under the License is distributed on an -// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY -// KIND, either express or implied. See the License for the -// specific language governing permissions and limitations -// under the License. - -pub const PREAMBLE_LONGS_SHORT: u8 = 2; -pub const SERIAL_VERSION: u8 = 1; -pub const FLAGS_IS_EMPTY: u8 = 1 << 0; -pub const LONG_SIZE_BYTES: usize = 8; diff --git a/datasketches/src/countmin/sketch.rs b/datasketches/src/countmin/sketch.rs index b81576fa..f36b2fee 100644 --- a/datasketches/src/countmin/sketch.rs +++ b/datasketches/src/countmin/sketch.rs @@ -21,34 +21,45 @@ use std::hash::Hasher; use crate::codec::SketchBytes; use crate::codec::SketchSlice; use crate::codec::assert::ensure_preamble_longs_in; -use crate::codec::assert::ensure_serial_version_is; use crate::codec::assert::insufficient_data; use crate::codec::family::Family; use crate::countmin::CountMinValue; use crate::countmin::UnsignedCountMinValue; -use crate::countmin::serialization::FLAGS_IS_EMPTY; -use crate::countmin::serialization::LONG_SIZE_BYTES; -use crate::countmin::serialization::PREAMBLE_LONGS_SHORT; -use crate::countmin::serialization::SERIAL_VERSION; use crate::error::Error; use crate::error::ErrorKind; use crate::hash::DEFAULT_UPDATE_SEED; use crate::hash::MurmurHash3X64128; +use crate::hash::XxHash64; use crate::hash::check_seed_hash; use crate::hash::compute_seed_hash; +const PREAMBLE_LONGS_SHORT: u8 = 2; +const SERIAL_VERSION_1: u8 = 1; +const SERIAL_VERSION_2: u8 = 2; +const FLAGS_IS_EMPTY: u8 = 1 << 0; +const LONG_SIZE_BYTES: usize = 8; const MAX_TABLE_ENTRIES: usize = 1 << 30; /// CountMin sketch for estimating item frequencies. /// /// The sketch provides upper and lower bounds on estimated item frequencies /// with configurable relative error and confidence. +/// +/// New sketches use serial version 2. Deserialized version 1 sketches retain their legacy +/// hashing and serialization format, and sketches of different versions cannot be merged. +/// +/// For cross-language version 2 interchange, use `i64` counters and encode inputs consistently: +/// * `i64` and `u64` values are hashed as eight little-endian bytes; use +/// [`crate::hash::value::natural_extend`] for narrower integers. +/// * Use [`crate::hash::value::raw_bytes`] for UTF-8 strings and byte slices, and skip empty inputs +/// when interoperating with implementations that ignore them. #[derive(Debug, Clone, PartialEq)] pub struct CountMinSketch { num_hashes: u8, buckets: BucketCount, seed: u64, seed_hash: u16, + serial_version: SerialVersion, // Every bucket satisfies |count| <= total_weight, so a checked total also bounds bucket // additions during update/merge: |a + b| <= |a| + |b|. Deserialization validates this // invariant; unsigned halving and decay preserve it. @@ -57,6 +68,91 @@ pub struct CountMinSketch { hash_seeds: Vec, } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[repr(u8)] +enum SerialVersion { + V1 = SERIAL_VERSION_1, + V2 = SERIAL_VERSION_2, +} + +impl SerialVersion { + fn from_u8(byte: u8) -> Result { + match byte { + SERIAL_VERSION_1 => Ok(Self::V1), + SERIAL_VERSION_2 => Ok(Self::V2), + _ => Err(Error::deserial(format!( + "unsupported serial version: expected {SERIAL_VERSION_1} or {SERIAL_VERSION_2}, got {byte}" + ))), + } + } + + fn max_hashes(self) -> u8 { + match self { + Self::V1 => u8::MAX, + Self::V2 => 127, + } + } + + fn entries_for_config( + self, + num_hashes: u8, + num_buckets: u32, + kind: ErrorKind, + ) -> Result { + if !(1..=self.max_hashes()).contains(&num_hashes) { + return Err(Error::new( + kind, + match self { + Self::V1 => "num_hashes must be at least 1", + Self::V2 => "num_hashes must be within 1..=127", + }, + )); + } + if num_buckets < 3 { + return Err(Error::new(kind, "num_buckets must be at least 3")); + } + let entries = (num_hashes as usize) + .checked_mul(num_buckets as usize) + .ok_or_else(|| Error::new(kind, "num_hashes * num_buckets overflows usize"))?; + if entries >= MAX_TABLE_ENTRIES { + return Err(Error::new( + kind, + format!("num_hashes * num_buckets must be < {MAX_TABLE_ENTRIES}"), + )); + } + Ok(entries) + } + + fn make_hash_seeds(self, seed: u64, num_hashes: u8) -> Vec { + (0..num_hashes) + .map(|i| match self { + Self::V1 => { + let mut hasher = MurmurHash3X64128::with_seed(seed); + hasher.write(&u64::from(i).to_le_bytes()); + hasher.finish128().0 + } + Self::V2 => XxHash64::hash_u64(u64::from(i), seed), + }) + .collect() + } + + #[inline] + fn hash_item(self, item: &I, seed: u64) -> u64 { + match self { + Self::V1 => { + let mut hasher = MurmurHash3X64128::with_seed(seed); + item.hash(&mut hasher); + hasher.finish128().0 + } + Self::V2 => { + let mut hasher = XxHash64::with_seed(seed); + item.hash(&mut hasher); + hasher.finish64() >> 1 + } + } + } +} + #[derive(Debug, Clone, PartialEq)] struct BucketCount { count: u32, @@ -101,8 +197,8 @@ impl CountMinSketch { /// /// # Errors /// - /// Returns an error if `num_hashes` is `0`, `num_buckets` is less than `3`, or the total table - /// size exceeds the supported limit. + /// Returns an error if `num_hashes` is outside `1..=127`, `num_buckets` is less than `3`, or + /// the total table size exceeds the supported limit. /// /// # Examples /// @@ -121,7 +217,7 @@ impl CountMinSketch { /// # Errors /// /// Returns an error if any of: - /// * `num_hashes` is `0`. + /// * `num_hashes` is outside `1..=127`. /// * `num_buckets` is less than `3`. /// * The total table size exceeds the supported limit. /// * The computed seed hash is zero. @@ -135,13 +231,18 @@ impl CountMinSketch { /// assert_eq!(sketch.seed(), 42); /// ``` pub fn with_seed(num_hashes: u8, num_buckets: u32, seed: u64) -> Result { - let entries = entries_for_config(num_hashes, num_buckets)?; + let entries = SerialVersion::V2.entries_for_config( + num_hashes, + num_buckets, + ErrorKind::InvalidArgument, + )?; let seed_hash = compute_seed_hash(seed, ErrorKind::InvalidArgument)?; Ok(Self::make( num_hashes, num_buckets, seed, seed_hash, + SerialVersion::V2, entries, )) } @@ -219,6 +320,9 @@ impl CountMinSketch { /// Updates the sketch with a single occurrence of the item. /// + /// Use [`hash::value`](crate::hash::value) wrappers to select an input encoding for + /// cross-language hashing. + /// /// # Panics /// /// Panics without modifying the sketch if the total absolute weight would exceed `T::MAX`. @@ -229,8 +333,8 @@ impl CountMinSketch { /// use datasketches::countmin::CountMinSketch; /// /// let mut sketch = CountMinSketch::::new(4, 128).unwrap(); - /// sketch.update("apple"); - /// assert!(sketch.estimate("apple") >= 1); + /// sketch.update(1i64); + /// assert!(sketch.estimate(1i64) >= 1); /// ``` pub fn update(&mut self, item: I) { self.update_with_weight(item, T::ONE); @@ -238,6 +342,9 @@ impl CountMinSketch { /// Updates the sketch with the given item and weight. /// + /// Use [`hash::value`](crate::hash::value) wrappers to select an input encoding for + /// cross-language hashing. + /// /// # Panics /// /// Panics without modifying the sketch if the absolute weight or the total absolute weight @@ -249,8 +356,8 @@ impl CountMinSketch { /// use datasketches::countmin::CountMinSketch; /// /// let mut sketch = CountMinSketch::::new(4, 128).unwrap(); - /// sketch.update_with_weight("banana", 3); - /// assert!(sketch.estimate("banana") >= 3); + /// sketch.update_with_weight(2i64, 3); + /// assert!(sketch.estimate(2i64) >= 3); /// ``` pub fn update_with_weight(&mut self, item: I, weight: T) { if weight == T::ZERO { @@ -276,19 +383,15 @@ impl CountMinSketch { /// use datasketches::countmin::CountMinSketch; /// /// let mut sketch = CountMinSketch::::new(4, 128).unwrap(); - /// sketch.update_with_weight("pear", 2); - /// assert!(sketch.estimate("pear") >= 2); + /// sketch.update_with_weight(3i64, 2); + /// assert!(sketch.estimate(3i64) >= 2); /// ``` pub fn estimate(&self, item: I) -> T { let num_buckets = self.buckets.count as usize; let mut min = T::MAX; for (row, seed) in self.hash_seeds.iter().enumerate() { let bucket = self.bucket_index(&item, *seed); - let index = row * num_buckets + bucket; - let value = self.counts[index]; - if value < min { - min = value; - } + min = min.min(self.counts[row * num_buckets + bucket]); } min } @@ -310,7 +413,8 @@ impl CountMinSketch { /// # Errors /// /// Returns an error without modifying the sketch if the sketches have different numbers of - /// hashes, bucket counts, or seeds, or their combined total absolute weight exceeds `T::MAX`. + /// hashes, bucket counts, seeds, or serial versions, or their combined total absolute weight + /// exceeds `T::MAX`. /// /// # Examples /// @@ -320,19 +424,20 @@ impl CountMinSketch { /// let mut left = CountMinSketch::::new(4, 128).unwrap(); /// let mut right = CountMinSketch::::new(4, 128).unwrap(); /// - /// left.update("apple"); - /// right.update_with_weight("banana", 2); + /// left.update(1i64); + /// right.update_with_weight(2i64, 2); /// /// left.merge(&right).unwrap(); - /// assert!(left.estimate("banana") >= 2); + /// assert!(left.estimate(2i64) >= 2); /// ``` pub fn merge(&mut self, other: &CountMinSketch) -> Result<(), Error> { if self.num_hashes != other.num_hashes || self.buckets.count != other.buckets.count || self.seed != other.seed + || self.serial_version != other.serial_version { return Err(Error::invalid_argument( - "Count-Min sketches must have matching numbers of hashes, bucket counts, and seeds", + "Count-Min sketches must have matching numbers of hashes, bucket counts, seeds, and serial versions", )); } self.total_weight = self @@ -347,8 +452,11 @@ impl CountMinSketch { /// Serializes this sketch into the Count-Min binary format. /// - /// DataSketches C++ can read the serialized data, but item queries and updates there are - /// unreliable because the two implementations map items to counters differently. + /// New sketches are written as version 2. A sketch deserialized from a version 1 image + /// retains its legacy hashing for queries and updates and is written back as version 1. + /// + /// Version 1 images are not portable across implementations. Version 2 supports cross-language + /// interchange with `i64` counters and the input encodings documented on [`CountMinSketch`]. /// /// # Examples /// @@ -356,10 +464,10 @@ impl CountMinSketch { /// use datasketches::countmin::CountMinSketch; /// /// let mut sketch = CountMinSketch::::new(4, 128).unwrap(); - /// sketch.update("apple"); + /// sketch.update(1i64); /// let bytes = sketch.serialize(); /// let decoded = CountMinSketch::::deserialize(&bytes).unwrap(); - /// assert!(decoded.estimate("apple") >= 1); + /// assert!(decoded.estimate(1i64) >= 1); /// ``` pub fn serialize(&self) -> Vec { let header_size = PREAMBLE_LONGS_SHORT as usize * LONG_SIZE_BYTES; @@ -372,7 +480,7 @@ impl CountMinSketch { let mut bytes = SketchBytes::with_capacity(header_size + payload_size); bytes.write_u8(PREAMBLE_LONGS_SHORT); - bytes.write_u8(SERIAL_VERSION); + bytes.write_u8(self.serial_version as u8); bytes.write_u8(Family::COUNTMIN.id); bytes.write_u8(if self.is_empty() { FLAGS_IS_EMPTY } else { 0 }); bytes.write_u32_le(0); // unused @@ -399,9 +507,8 @@ impl CountMinSketch { /// Deserializes a sketch from bytes using the default seed. /// - /// Can read sketches serialized by DataSketches C++. - /// Item queries and updates on those sketches are unreliable because the two - /// implementations map items to counters differently. + /// Accepts versions 1 and 2, preserving the image's hashing and serialization version. + /// Counter values must fit `T`. See [`CountMinSketch`] for input encodings and compatibility. /// /// # Errors /// @@ -414,10 +521,10 @@ impl CountMinSketch { /// use datasketches::countmin::CountMinSketch; /// /// let mut sketch = CountMinSketch::::new(4, 64).unwrap(); - /// sketch.update("apple"); + /// sketch.update(1i64); /// let bytes = sketch.serialize(); /// let decoded = CountMinSketch::::deserialize(&bytes).unwrap(); - /// assert!(decoded.estimate("apple") >= 1); + /// assert!(decoded.estimate(1i64) >= 1); /// ``` pub fn deserialize(bytes: &[u8]) -> Result { Self::deserialize_with_seed(bytes, DEFAULT_UPDATE_SEED) @@ -425,9 +532,8 @@ impl CountMinSketch { /// Deserializes a sketch from bytes using the provided seed. /// - /// Can read sketches serialized by DataSketches C++. - /// Item queries and updates on those sketches are unreliable because the two - /// implementations map items to counters differently. + /// Accepts versions 1 and 2, preserving the image's hashing and serialization version. + /// Counter values must fit `T`. See [`CountMinSketch`] for input encodings and compatibility. /// /// # Errors /// @@ -440,10 +546,10 @@ impl CountMinSketch { /// use datasketches::countmin::CountMinSketch; /// /// let mut sketch = CountMinSketch::::with_seed(4, 64, 7).unwrap(); - /// sketch.update("apple"); + /// sketch.update(1i64); /// let bytes = sketch.serialize(); /// let decoded = CountMinSketch::::deserialize_with_seed(&bytes, 7).unwrap(); - /// assert!(decoded.estimate("apple") >= 1); + /// assert!(decoded.estimate(1i64) >= 1); /// ``` pub fn deserialize_with_seed(bytes: &[u8], seed: u64) -> Result { fn read_value( @@ -469,7 +575,7 @@ impl CountMinSketch { .map_err(insufficient_data(""))?; Family::COUNTMIN.validate_id(family_id)?; - ensure_serial_version_is(SERIAL_VERSION, serial_version)?; + let serial_version = SerialVersion::from_u8(serial_version)?; ensure_preamble_longs_in(&[PREAMBLE_LONGS_SHORT], preamble_longs)?; let num_buckets = cursor @@ -489,7 +595,8 @@ impl CountMinSketch { ErrorKind::InvalidData, )?; - let entries = entries_for_config_checked(num_hashes, num_buckets)?; + let entries = + serial_version.entries_for_config(num_hashes, num_buckets, ErrorKind::InvalidData)?; let is_empty = (flags & FLAGS_IS_EMPTY) != 0; if !is_empty { let payload_values = entries @@ -507,7 +614,14 @@ impl CountMinSketch { } } - let mut sketch = Self::make(num_hashes, num_buckets, seed, expected_seed_hash, entries); + let mut sketch = Self::make( + num_hashes, + num_buckets, + seed, + expected_seed_hash, + serial_version, + entries, + ); if is_empty { return Ok(sketch); } @@ -537,14 +651,22 @@ impl CountMinSketch { self.counts.capacity() * size_of::() + self.hash_seeds.capacity() * size_of::() } - fn make(num_hashes: u8, num_buckets: u32, seed: u64, seed_hash: u16, entries: usize) -> Self { + fn make( + num_hashes: u8, + num_buckets: u32, + seed: u64, + seed_hash: u16, + serial_version: SerialVersion, + entries: usize, + ) -> Self { let counts = vec![T::ZERO; entries]; - let hash_seeds = make_hash_seeds(seed, num_hashes); + let hash_seeds = serial_version.make_hash_seeds(seed, num_hashes); CountMinSketch { num_hashes, buckets: BucketCount::new(num_buckets), seed, seed_hash, + serial_version, total_weight: T::ZERO, counts, hash_seeds, @@ -552,10 +674,8 @@ impl CountMinSketch { } fn bucket_index(&self, item: &I, seed: u64) -> usize { - let mut hasher = MurmurHash3X64128::with_seed(seed); - item.hash(&mut hasher); - let (h1, _) = hasher.finish128(); - self.buckets.index(h1) + self.buckets + .index(self.serial_version.hash_item(item, seed)) } } @@ -568,9 +688,9 @@ impl CountMinSketch { /// use datasketches::countmin::CountMinSketch; /// /// let mut sketch = CountMinSketch::::new(4, 128).unwrap(); - /// sketch.update_with_weight("apple", 3); + /// sketch.update_with_weight(1i64, 3); /// sketch.halve(); - /// assert!(sketch.estimate("apple") >= 1); + /// assert!(sketch.estimate(1i64) >= 1); /// ``` pub fn halve(&mut self) { for c in &mut self.counts { @@ -596,9 +716,9 @@ impl CountMinSketch { /// use datasketches::countmin::CountMinSketch; /// /// let mut sketch = CountMinSketch::::new(4, 128).unwrap(); - /// sketch.update_with_weight("apple", 3); + /// sketch.update_with_weight(1i64, 3); /// sketch.decay(0.5); - /// assert!(sketch.estimate("apple") >= 1); + /// assert!(sketch.estimate(1i64) >= 1); /// ``` pub fn decay(&mut self, decay: f64) { assert!(decay > 0.0 && decay <= 1.0, "decay must be within (0, 1]"); @@ -616,54 +736,6 @@ impl CountMinSketch { } } -fn entries_for_config(num_hashes: u8, num_buckets: u32) -> Result { - if num_hashes == 0 { - return Err(Error::invalid_argument("num_hashes must be at least 1")); - } - if num_buckets < 3 { - return Err(Error::invalid_argument("num_buckets must be at least 3")); - } - let entries = (num_hashes as usize) - .checked_mul(num_buckets as usize) - .ok_or_else(|| Error::invalid_argument("num_hashes * num_buckets overflows usize"))?; - if entries >= MAX_TABLE_ENTRIES { - return Err(Error::invalid_argument(format!( - "num_hashes * num_buckets must be < {MAX_TABLE_ENTRIES}" - ))); - } - Ok(entries) -} - -fn entries_for_config_checked(num_hashes: u8, num_buckets: u32) -> Result { - if num_hashes == 0 { - return Err(Error::deserial("num_hashes must be at least 1")); - } - if num_buckets < 3 { - return Err(Error::deserial("num_buckets must be at least 3")); - } - let entries = (num_hashes as usize) - .checked_mul(num_buckets as usize) - .ok_or_else(|| Error::deserial("num_hashes * num_buckets overflows usize"))?; - if entries >= MAX_TABLE_ENTRIES { - return Err(Error::deserial(format!( - "num_hashes * num_buckets must be < {MAX_TABLE_ENTRIES}", - ))); - } - Ok(entries) -} - -fn make_hash_seeds(seed: u64, num_hashes: u8) -> Vec { - let mut seeds = Vec::with_capacity(num_hashes as usize); - for i in 0..num_hashes { - // Derive per-row hash seeds deterministically from the sketch seed. - let mut hasher = MurmurHash3X64128::with_seed(seed); - hasher.write(&u64::from(i).to_le_bytes()); - let (h1, _) = hasher.finish128(); - seeds.push(h1); - } - seeds -} - #[cfg(test)] mod tests { use super::*; diff --git a/datasketches/src/hash/mod.rs b/datasketches/src/hash/mod.rs index f2dde4f4..69fff2eb 100644 --- a/datasketches/src/hash/mod.rs +++ b/datasketches/src/hash/mod.rs @@ -38,9 +38,9 @@ mod murmurhash; ))] pub(crate) use self::murmurhash::*; -#[cfg(any(feature = "bloom", feature = "xor"))] +#[cfg(any(feature = "bloom", feature = "countmin", feature = "xor"))] mod xxhash; -#[cfg(any(feature = "bloom", feature = "xor"))] +#[cfg(any(feature = "bloom", feature = "countmin", feature = "xor"))] pub(crate) use self::xxhash::*; #[cfg(any( @@ -82,7 +82,7 @@ pub(crate) use self::seed::*; ))] pub(crate) const DEFAULT_UPDATE_SEED: u64 = 9001; -#[cfg(any(feature = "bloom", feature = "xor"))] +#[cfg(any(feature = "bloom", feature = "countmin", feature = "xor"))] #[inline(always)] fn read_u32_le(bytes: &[u8]) -> u32 { u32::from_le_bytes(bytes.try_into().expect("four-byte hash input")) diff --git a/datasketches/src/hash/value/mod.rs b/datasketches/src/hash/value/mod.rs index a4867a3d..803aba5d 100644 --- a/datasketches/src/hash/value/mod.rs +++ b/datasketches/src/hash/value/mod.rs @@ -48,7 +48,7 @@ //! //! [`natural_extend::NaturalExtend`] first widens signed values to `i64` and unsigned values to //! `u64`, and then hashes the resulting integers. Use this encoding to match short-integer hashing -//! in DataSketches C++'s Bloom filters. +//! in DataSketches C++'s Bloom filters and Count-Min sketches. //! //! The concrete wrapper documentation provides more details and examples. //! diff --git a/datasketches/src/hash/value/natural_extend.rs b/datasketches/src/hash/value/natural_extend.rs index 29d7fcff..780116f6 100644 --- a/datasketches/src/hash/value/natural_extend.rs +++ b/datasketches/src/hash/value/natural_extend.rs @@ -20,7 +20,8 @@ //! [`NaturalExtend`] first widens signed values to `i64` and unsigned values to `u64`, and then //! hashes the resulting integers. //! -//! Use this encoding to match short-integer hashing in DataSketches C++'s Bloom filters. +//! Use this encoding to match short-integer hashing in DataSketches C++'s Bloom filters and +//! Count-Min sketches. use std::hash::Hash; use std::hash::Hasher; diff --git a/datasketches/src/lib.rs b/datasketches/src/lib.rs index 18cd2bc4..bc77ea81 100644 --- a/datasketches/src/lib.rs +++ b/datasketches/src/lib.rs @@ -54,7 +54,7 @@ //! * `raw_bytes` for byte and string contents; //! * `canonical_float` for floating-point values; //! * `sign_extend` for short integers used with HLL and CPC; -//! * `natural_extend` for short integers used with Bloom filters. +//! * `natural_extend` for short integers used with Bloom filters and Count-Min sketches. //! //! When interoperating with an implementation that skips empty strings, check for empty input //! before updating here as well. diff --git a/tests-integration/tests/countmin_test/sketch.rs b/tests-integration/tests/countmin_test/sketch.rs index 8508bee6..34372936 100644 --- a/tests-integration/tests/countmin_test/sketch.rs +++ b/tests-integration/tests/countmin_test/sketch.rs @@ -15,12 +15,15 @@ // specific language governing permissions and limitations // under the License. +use std::hash::Hash; use std::panic::AssertUnwindSafe; use std::panic::catch_unwind; use datasketches::countmin::CountMinSketch; use datasketches::countmin::UnsignedCountMinValue; use datasketches::error::ErrorKind; +use datasketches::hash::value::natural_extend; +use datasketches::hash::value::raw_bytes; use googletest::assert_that; use googletest::prelude::ge; use googletest::prelude::le; @@ -415,8 +418,11 @@ fn test_truncated_non_empty_payload_is_rejected_before_table_allocation() { #[test] fn test_invalid_hashes_return_error() { - let error = CountMinSketch::::new(0, 5).unwrap_err(); - assert_eq!(error.kind(), ErrorKind::InvalidArgument); + for rows in [0, 128, 255] { + let error = CountMinSketch::::new(rows, 5).unwrap_err(); + assert_eq!(error.kind(), ErrorKind::InvalidArgument); + } + assert!(CountMinSketch::::new(127, 5).is_ok()); } #[test] @@ -432,3 +438,29 @@ fn test_merge_incompatible() { let error = left.merge(&right).unwrap_err(); assert_eq!(error.kind(), ErrorKind::InvalidArgument); } + +#[test] +fn value_wrappers_match_portable_item_encodings() { + fn assert_same_item(left: I, right: J) { + let mut a = CountMinSketch::::new(3, 1000).unwrap(); + let mut b = a.clone(); + a.update_with_weight(&left, 7); + b.update_with_weight(&right, 7); + assert_eq!(a, b); + } + + assert_same_item(natural_extend::from_i8(-2), -2i64); + assert_same_item(natural_extend::from_i16(-2), -2i64); + assert_same_item(natural_extend::from_i32(-2), -2i64); + assert_same_item(natural_extend::from_u8(u8::MAX), u64::from(u8::MAX)); + assert_same_item(natural_extend::from_u16(u16::MAX), u64::from(u16::MAX)); + assert_same_item(natural_extend::from_u32(u32::MAX), u64::from(u32::MAX)); + assert_same_item( + -2i64, + raw_bytes::from_slice(&[0xfe, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff]), + ); + assert_same_item( + 0x0102_0304_0506_0708u64, + raw_bytes::from_slice(&[8, 7, 6, 5, 4, 3, 2, 1]), + ); +} diff --git a/tests-integration/tests/serde_tests/countmin.rs b/tests-integration/tests/serde_tests/countmin.rs index 4a62faaf..a0a019d1 100644 --- a/tests-integration/tests/serde_tests/countmin.rs +++ b/tests-integration/tests/serde_tests/countmin.rs @@ -16,15 +16,237 @@ // under the License. use std::fs; +use std::hash::Hash; use datasketches::countmin::CountMinSketch; use datasketches::error::ErrorKind; +use datasketches::hash::value::natural_extend; +use datasketches::hash::value::raw_bytes; use googletest::assert_that; use googletest::prelude::contains_substring; use tests_integration::ZERO_HASH_SEED; use crate::serialization_test_data; +fn words_to_bytes(words: [u64; N]) -> Vec { + words.into_iter().flat_map(u64::to_le_bytes).collect() +} + +#[test] +fn v2_reads_and_writes_the_published_image() { + // TCK discussion #18, section 7.4: update(1, 10), update(9, 810), update("apple", 5). + let bytes = words_to_bytes([ + 0x0000_0000_0012_0202, + 0x0093_cc02_0000_0009, + 825, + 810, + 0, + 0, + 0, + 5, + 0, + 10, + 0, + 0, + 0, + 5, + 810, + 0, + 0, + 0, + 0, + 10, + 0, + ]); + let mut sketch = CountMinSketch::::new(2, 9).unwrap(); + let mut empty = bytes[..16].to_vec(); + empty[3] = 1; + assert_eq!(sketch.serialize(), empty); + assert_eq!(CountMinSketch::::deserialize(&empty).unwrap(), sketch); + + sketch.update_with_weight(1i64, 10); + sketch.update_with_weight(9i64, 810); + sketch.update_with_weight(raw_bytes::from_str("apple"), 5); + assert_eq!(sketch.serialize(), bytes); + + let mut restored = CountMinSketch::::deserialize(&bytes).unwrap(); + assert_eq!(restored, sketch); + assert_eq!(restored.estimate(1u64), 10); + assert_eq!(restored.estimate(natural_extend::from_i8(9)), 810); + assert_eq!(restored.estimate(raw_bytes::from_str("apple")), 5); + restored.update_with_weight(raw_bytes::from_str("apple"), -2); + assert_eq!(restored.estimate(raw_bytes::from_str("apple")), 3); + assert_eq!(restored.total_weight(), 827); + sketch.merge(&restored).unwrap(); + assert_eq!(sketch.estimate(raw_bytes::from_str("apple")), 8); + assert_eq!(sketch.total_weight(), 1652); +} + +#[test] +fn v2_matches_known_bucket_vectors() { + fn check(item: I, seed: u64, expected: [usize; 3]) { + let mut sketch = CountMinSketch::::with_seed(3, 1000, seed).unwrap(); + sketch.update_with_weight(&item, 7); + let bytes = sketch.serialize(); + for (row, bucket) in expected.into_iter().enumerate() { + let row_start = 24 + row * 1000 * 8; + let counts: Vec<(usize, i64)> = bytes[row_start..row_start + 1000 * 8] + .chunks_exact(8) + .map(|bytes| i64::from_le_bytes(bytes.try_into().unwrap())) + .enumerate() + .filter(|(_, count)| *count != 0) + .collect(); + assert_eq!(counts, [(bucket, 7)], "seed {seed}, row {row}"); + } + } + + // Java XxHash64 vectors from TCK discussion #18, section 7.3. + check(-1i64, 9001, [256, 641, 308]); + check(natural_extend::from_i8(-1), 9001, [256, 641, 308]); + check(i64::MIN, 9001, [951, 0, 298]); + check(raw_bytes::from_str("apple"), 9001, [130, 747, 144]); + check( + raw_bytes::from_slice("été".as_bytes()), + 9001, + [806, 491, 163], + ); + + // Independently computed with XXH64 over little-endian u64 inputs. + check(1u64, 9001, [185, 898, 4]); + check(9u64, 9001, [490, 217, 440]); + check(natural_extend::from_u16(9), 9001, [490, 217, 440]); + check(1u64, 42, [468, 459, 696]); + check(9u64, 42, [841, 428, 322]); +} + +#[test] +fn legacy_image_preserves_wrappers_and_custom_hash_inputs() { + #[derive(Hash)] + struct Key(u64); + + // Generated by Rust v0.6.0: raw_bytes("apple", 5), natural_extend(-1i8, 7), + // Key(42, 11), and raw_bytes("", 13), with two rows and nine buckets. + let bytes = words_to_bytes([ + 0x0000_0000_0012_0102, + 0x0093_cc02_0000_0009, + 36, + 7, + 0, + 16, + 0, + 13, + 0, + 0, + 0, + 0, + 11, + 0, + 12, + 0, + 0, + 0, + 13, + 0, + 0, + ]); + let mut sketch = CountMinSketch::::deserialize(&bytes).unwrap(); + assert_eq!(sketch.serialize(), bytes); + + let restored = CountMinSketch::::deserialize(&bytes).unwrap(); + assert_eq!(restored.estimate(raw_bytes::from_str("apple")), 12); + assert_eq!(restored.estimate(natural_extend::from_i8(-1)), 7); + assert_eq!(restored.estimate(Key(42)), 11); + assert_eq!(restored.estimate(raw_bytes::from_str("")), 13); + + sketch.merge(&restored).unwrap(); + sketch.update(Key(42)); + assert_eq!(sketch.estimate(Key(42)), 23); + assert_eq!(sketch.total_weight(), 73); + assert_eq!(sketch.serialize()[1], 1); + assert_eq!( + CountMinSketch::::deserialize(&sketch.serialize()).unwrap(), + sketch + ); +} + +#[test] +fn v2_row_limit_does_not_reject_legacy_images() { + for rows in [128, 255] { + let mut bytes = CountMinSketch::::new(1, 3).unwrap().serialize(); + bytes[12] = rows; + assert_eq!( + CountMinSketch::::deserialize(&bytes) + .unwrap_err() + .kind(), + ErrorKind::InvalidData + ); + bytes[1] = 1; + let legacy = CountMinSketch::::deserialize(&bytes).unwrap(); + assert_eq!(legacy.num_hashes(), rows); + assert_eq!(legacy.serialize(), bytes); + } +} + +// Generated by CountMinSketch v0.5.0 with one row, five buckets, seed 123, and key 1 at weight 10. +fn rust_v1_custom_seed_image() -> Vec { + words_to_bytes([ + 0x0000_0000_0012_0102, + 0x0005_b601_0000_0005, + 10, + 0, + 0, + 0, + 0, + 10, + ]) +} + +#[test] +fn legacy_rust_v1_image_preserves_queries_updates_and_merges() { + let image = rust_v1_custom_seed_image(); + let mut sketch = CountMinSketch::::deserialize_with_seed(&image, 123).unwrap(); + assert_eq!(sketch.total_weight(), 10); + assert_eq!(sketch.estimate(1u64), 10); + assert_eq!(sketch.serialize(), image); + + sketch.update_with_weight(1u64, 5); + assert_eq!(sketch.total_weight(), 15); + assert_eq!(sketch.estimate(1u64), 15); + assert_eq!(sketch.serialize()[1], 1); + + let mut merged = CountMinSketch::::deserialize_with_seed(&image, 123).unwrap(); + merged.merge(&sketch).unwrap(); + assert_eq!(merged.total_weight(), 25); + assert_eq!(merged.estimate(1u64), 25); +} + +#[test] +fn rejects_merge_across_serial_versions() { + let legacy = + CountMinSketch::::deserialize_with_seed(&rust_v1_custom_seed_image(), 123).unwrap(); + let mut current = CountMinSketch::::with_seed(1, 5, 123).unwrap(); + current.update_with_weight(1u64, 10); + let before = current.clone(); + + assert_eq!( + current.merge(&legacy).unwrap_err().kind(), + ErrorKind::InvalidArgument + ); + assert_eq!(current, before); +} + +#[test] +fn rejects_unknown_serial_version() { + let mut bytes = CountMinSketch::::new(1, 5).unwrap().serialize(); + bytes[1] = 3; + let error = CountMinSketch::::deserialize(&bytes).unwrap_err(); + assert_eq!(error.kind(), ErrorKind::InvalidData); + assert_that!( + error.message(), + contains_substring("expected 1 or 2, got 3") + ); +} + #[test] fn rejects_weights_that_cannot_bound_counter_arithmetic() { let mut sketch = CountMinSketch::::new(2, 8).unwrap(); @@ -43,9 +265,9 @@ fn rejects_weights_that_cannot_bound_counter_arithmetic() { } } -// This test validates binary format compatibility (deserialize + byte round-trip) for -// datasketches-cpp CountMin snapshots. It intentionally does not assert estimate equivalence -// against original input keys because per-row hash seed derivation differs across implementations. +// This helper validates binary format compatibility (deserialize + byte round-trip) for +// version 1 datasketches-cpp CountMin snapshots. It intentionally does not assert estimate +// equivalence because version 1 row-seed derivation differs across implementations. fn assert_cpp_snapshot( filename: &str, seed: u64,