diff --git a/AGENTS.md b/AGENTS.md index f5f631db..f8ac5190 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -4,6 +4,4 @@ Before planning or modifying this repository, read [CONTRIBUTING.md](CONTRIBUTIN For test changes, pay particular attention to the "Integration test layout" and "Serialization snapshots" sections. Keep the documented workflow synchronized with structural changes, and run the applicable `cargo x check`, `cargo x test`, and `cargo x lint` commands before handing work back. -Apply the changelog guidance in [CONTRIBUTING.md](CONTRIBUTING.md) to every change. Update the permanent `Unreleased` section in the same pull request for significant user-visible behavior, and do not add entries mechanically for excluded maintenance work. - -Treat `CHANGELOG.md` as release notes for users rather than a summary of implementation work. Name the affected API or workload and the observable outcome, and keep performance claims within the scenario supported by evidence. +For every change, follow the [changelog guidance](CONTRIBUTING.md#changelog) in `CONTRIBUTING.md` as the single source of truth. diff --git a/CHANGELOG.md b/CHANGELOG.md index 8dd97ec6..7e1c7258 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,13 +6,25 @@ All significant changes to this project will be documented in this file. ### Breaking changes +* `BloomFilter::union` and `BloomFilter::intersect` now return `Result`. Callers must handle incompatible filter configurations instead of relying on a panic. +* `CountMinSketch::merge` now returns `Result`. Callers must handle incompatible sketch configurations instead of relying on a panic. +* `CountMinSketch::{suggest_num_buckets, suggest_num_hashes}` now return `Result`. Callers must handle invalid or unsupported targets; successful suggestions are valid inputs to `CountMinSketch::new`. +* `CpcUnion::update` now returns `Result`. Callers must handle seed mismatches instead of relying on a panic. +* Remove `BloomFilterBuilder::suggest_num_bits`, `suggest_num_hashes_from_accuracy`, and `suggest_num_hashes_from_fpp`. Use `with_accuracy(...).build()` for target-based sizing or `with_size(...).build()` for an explicit precomputed configuration. +* `CpcSketch::max_serialized_bytes` now returns `Result` and reports an invalid `lg_k` instead of panicking. +* `FrequentItemsSketch::new` now rejects map sizes below the minimum of 8 instead of silently rounding them up. +* Replace `FrequentItemsSketch::epsilon_for_lg` with the fallible `epsilon_for_max_map_size`, and change `apriori_error` to accept the same maximum map size plus an unsigned stream weight. These helpers now match the constructor's units, and `max_map_size` exposes the configured value. +* Replace the `is_f32` flag on `TDigestMut::deserialize` with separate `deserialize` and `deserialize_f32` entry points, making the serialized precision explicit at the call site. +* Remove `CpcUnion::num_coupons`, which exposed internal union state solely for tests. Inspect the resulting `CpcSketch` when diagnostics are needed. +* Remove the `TupleEntry` re-export. Tuple sketch iterators already expose retained entries as `(hash, &summary)` pairs without leaking the private storage representation. * `ThetaIntersection::to_sketch` and `TupleIntersection::to_sketch` now return `Option`. Callers must handle `None` until the intersection receives its first successful update. * `BloomFilterBuilder`, `ThetaSketchBuilder`, `ThetaUnionBuilder`, `TupleSketchBuilder`, and `TupleUnionBuilder` now validate their configuration when `build` is called, and `build` returns `Result`. Callers must propagate or handle construction errors. * `BloomFilterBuilder::{MIN_NUM_BITS, MAX_NUM_BITS, MIN_NUM_HASHES, MAX_NUM_HASHES}` are no longer public. Callers should pass configurations to `build` and handle `InvalidArgument` instead of prevalidating against these constants. -* Fallible sketch and operator constructors now return `Result` directly from `new` or `with_seed`. `ReqSketch` and `TDigestMut` no longer provide `try_new`, and the Count-Min parameter suggestion methods also return `Result`. +* Fallible sketch and operator constructors now return `Result` directly from `new` or `with_seed`. `TDigestMut` no longer provides `try_new`. ### New features +* `TDigest` can now be serialized and deserialized directly without converting through `TDigestMut` at the call site. * Add Relative Error Quantiles (REQ) sketches behind the `req` feature, including configurable high- or low-rank accuracy, rank, quantile, PMF, and CDF queries, merging, totally ordered custom item types, the `ReqFloat` adapter for non-NaN floating-point values, and C++/Java-compatible serialization. ### Performance improvements @@ -22,7 +34,8 @@ All significant changes to this project will be documented in this file. ### Bug fixes -* Count-Min parameter suggestions now return constructor-valid values and reject relative-error targets that require more buckets than the sketch supports. +* Bloom filter accuracy construction now rejects targets that exceed the maximum serialized filter size instead of silently reducing capacity and violating the requested false-positive probability. +* T-Digest CDF and PMF queries now accept an empty split-point slice and return the single all-values bin instead of panicking. * Bloom filter deserialization now rejects malformed images with inconsistent counts or payload lengths, while valid images with a dirty cached count are restored correctly. * `FrequentItemsSketch` now enforces the cross-language map-size limit of `2^30` consistently. Oversized construction returns `InvalidArgument`, and malformed or oversized serialized images return `InvalidData` instead of panicking or attempting excessive allocation. * T-Digest compression now supports `k = u16::MAX` without overflowing. diff --git a/Cargo.lock b/Cargo.lock index cb1b0c9e..87ec3e30 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -155,9 +155,9 @@ checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801" [[package]] name = "chacha20" -version = "0.10.1" +version = "0.10.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d524456ba66e72eb8b115ff89e01e497f8e6d11d78b70b1aa13c0fbd97540a81" +checksum = "65c35e4b699c7e15ccbe7ee35c005e4fc0a278d22238a2857e6ce2dadeda1b06" dependencies = [ "cfg-if", "cpufeatures", diff --git a/README.md b/README.md index 674d361f..3143336e 100644 --- a/README.md +++ b/README.md @@ -81,7 +81,9 @@ See the [API documentation](https://docs.rs/datasketches) for configuration, acc The minimum supported Rust version is 1.86.0. The crate currently supports little-endian targets only. -Supported serialization formats are tested with fixtures produced by Apache DataSketches Java, C++, and Go through the [DataSketches TCK](https://github.com/apache/datasketches-tck). When values must hash identically across language implementations, use the compatibility wrappers in `hash::value`. +Supported serialization formats are tested with fixtures produced by Apache DataSketches Java, C++, and Go through the [DataSketches TCK](https://github.com/apache/datasketches-tck). + +Serialization compatibility does not imply that an ordinary Rust `Hash` implementation produces the same update bytes as another language. When sketches must represent the same inputs across implementations, use `hash::value::{raw_bytes, canonical_float, sign_extend, natural_extend}` (and the constructors within those modules) to match the other language implementations’ hashing rules. Other DataSketches implementations skip empty strings, so skip them before updating when that behavior matters. See the [changelog](CHANGELOG.md) for release notes and migration guidance. diff --git a/benchmarks/tdigest/merge.rs b/benchmarks/tdigest/merge.rs index 306c3172..13559c66 100644 --- a/benchmarks/tdigest/merge.rs +++ b/benchmarks/tdigest/merge.rs @@ -111,7 +111,7 @@ fn serialized_partials(bencher: Bencher, rows_per_partial: usize) { .bench_local(|| { let mut merged = TDigestMut::default(); for partial in black_box(&partials) { - let partial = TDigestMut::deserialize(partial, false).unwrap(); + let partial = TDigestMut::deserialize(partial).unwrap(); merged.merge(&partial); } black_box(merged.quantile(0.5)) @@ -131,7 +131,7 @@ fn serialized_overlapping_partials(bencher: Bencher) { .bench_local(|| { let mut merged = TDigestMut::default(); for partial in black_box(&partials) { - let partial = TDigestMut::deserialize(partial, false).unwrap(); + let partial = TDigestMut::deserialize(partial).unwrap(); merged.merge(&partial); } black_box(merged.quantile(0.5)) diff --git a/benchmarks/tdigest/serde.rs b/benchmarks/tdigest/serde.rs index 85688150..7d5e0a82 100644 --- a/benchmarks/tdigest/serde.rs +++ b/benchmarks/tdigest/serde.rs @@ -100,7 +100,7 @@ fn deserialize_small_partial_groups(bencher: Bencher) { .bench_local(|| { let digests = bytes .iter() - .map(|bytes| TDigestMut::deserialize(bytes, false).unwrap()) + .map(|bytes| TDigestMut::deserialize(bytes).unwrap()) .collect::>(); black_box(digests) }); @@ -131,7 +131,7 @@ fn deserialize_partial_groups(bencher: Bencher) { .bench_local(|| { let digests = bytes .iter() - .map(|bytes| TDigestMut::deserialize(bytes, false).unwrap()) + .map(|bytes| TDigestMut::deserialize(bytes).unwrap()) .collect::>(); black_box(digests) }); diff --git a/datasketches/src/bloom/mod.rs b/datasketches/src/bloom/mod.rs index 92e5ed82..dabecdbd 100644 --- a/datasketches/src/bloom/mod.rs +++ b/datasketches/src/bloom/mod.rs @@ -63,7 +63,8 @@ //! //! ## By Accuracy (Recommended) //! -//! Automatically calculates optimal size and hash functions: +//! Derive the size and hash-function count from an expected distinct-item count and a target +//! false-positive probability: //! //! ``` //! use datasketches::bloom::BloomFilterBuilder; @@ -77,6 +78,12 @@ //! .unwrap(); //! ``` //! +//! `max_items` is a sizing assumption, not an insertion limit. The filter continues accepting +//! distinct items beyond that count, but its false-positive probability can then exceed the target. +//! Accuracy inputs are validated by `build`: `max_items` must be positive, `fpp` must be in +//! `(0.0, 1.0]`, and the requested target must fit the serialized Bloom filter format. An `fpp` of +//! `1.0` is accepted and creates the smallest allocation: 64 bits and one hash function. +//! //! ## By Size (Manual) //! //! Specify requested bit count and hash functions (rounded up to a multiple of 64 bits): @@ -92,6 +99,10 @@ //! .unwrap(); //! ``` //! +//! Manual construction requires a positive bit count supported by the serialized format and a +//! hash-function count in `1..=32767`. The requested bit count is rounded up to a multiple of 64, +//! which is the value returned by [`BloomFilter::capacity`]. +//! //! # Set Operations //! //! Bloom filters support efficient set operations: @@ -110,12 +121,12 @@ //! filter2.insert("b"); //! //! // Union: recognizes items from either filter -//! filter1.union(&filter2); +//! filter1.union(&filter2).unwrap(); //! assert!(filter1.contains(&"a")); //! assert!(filter1.contains(&"b")); //! //! // Intersect: recognizes only items in both filters -//! // filter1.intersect(&filter2); +//! // filter1.intersect(&filter2).unwrap(); //! //! // Invert: approximately inverts set membership //! // filter1.invert(); diff --git a/datasketches/src/bloom/sketch.rs b/datasketches/src/bloom/sketch.rs index da0c471e..6a119e91 100644 --- a/datasketches/src/bloom/sketch.rs +++ b/datasketches/src/bloom/sketch.rs @@ -159,10 +159,11 @@ impl BloomFilter { /// After merging, this filter will recognize items from either filter /// (plus any false positives from either). /// - /// # Panics + /// # Errors /// - /// Panics if the filters are not compatible (different size, hashes, or seed). - /// Use [`is_compatible()`](Self::is_compatible) to check first. + /// Returns an error if the filters are not compatible (different size, number of hashes, or + /// seed). Use [`is_compatible()`](Self::is_compatible) to check first when an error is not + /// expected. /// /// # Examples /// @@ -181,15 +182,16 @@ impl BloomFilter { /// f1.insert("a"); /// f2.insert("b"); /// - /// f1.union(&f2); + /// f1.union(&f2).unwrap(); /// assert!(f1.contains(&"a")); /// assert!(f1.contains(&"b")); /// ``` - pub fn union(&mut self, other: &BloomFilter) { - assert!( - self.is_compatible(other), - "Cannot union incompatible Bloom filters" - ); + pub fn union(&mut self, other: &BloomFilter) -> Result<(), Error> { + if !self.is_compatible(other) { + return Err(Error::invalid_argument( + "Bloom filters must have matching capacity, number of hashes, and seed", + )); + } // Count bits during union operation (single pass) let mut num_bits_set = 0; @@ -198,6 +200,7 @@ impl BloomFilter { num_bits_set += word.count_ones() as u64; } self.num_bits_set = num_bits_set; + Ok(()) } /// Intersects this filter with another via bitwise AND. @@ -205,9 +208,10 @@ impl BloomFilter { /// After intersection, this filter will recognize only items present in both /// filters (plus false positives). /// - /// # Panics + /// # Errors /// - /// Panics if the filters are not compatible (different size, hashes, or seed). + /// Returns an error if the filters are not compatible (different size, number of hashes, or + /// seed). /// /// # Examples /// @@ -228,15 +232,16 @@ impl BloomFilter { /// f2.insert("b"); /// f2.insert("c"); /// - /// f1.intersect(&f2); + /// f1.intersect(&f2).unwrap(); /// assert!(f1.contains(&"b")); // In both /// // "a" and "c" likely return false now /// ``` - pub fn intersect(&mut self, other: &BloomFilter) { - assert!( - self.is_compatible(other), - "Cannot intersect incompatible Bloom filters" - ); + pub fn intersect(&mut self, other: &BloomFilter) -> Result<(), Error> { + if !self.is_compatible(other) { + return Err(Error::invalid_argument( + "Bloom filters must have matching capacity, number of hashes, and seed", + )); + } // Count bits during intersect operation (single pass) let mut num_bits_set = 0; @@ -245,6 +250,7 @@ impl BloomFilter { num_bits_set += word.count_ones() as u64; } self.num_bits_set = num_bits_set; + Ok(()) } /// Inverts all bits in the filter. @@ -612,6 +618,9 @@ impl BloomFilter { /// * [`with_size()`](Self::with_size): Specify requested bit count and hash functions (manual) /// /// Configuration is stored without validation and checked when [`build()`](Self::build) is called. +/// Accuracy construction treats `max_items` as a sizing assumption, not an insertion limit. The +/// filter continues accepting items beyond that count, but its false-positive probability can then +/// exceed the requested target. #[derive(Debug, Clone)] pub struct BloomFilterBuilder { mode: BloomFilterBuilderMode, @@ -637,11 +646,15 @@ impl BloomFilterBuilder { /// Maximum allowed number of hash functions. const MAX_NUM_HASHES: u16 = i16::MAX as u16; - /// Creates a builder with optimal parameters for a target accuracy. + /// Creates a builder that derives its parameters from a target accuracy. /// - /// Automatically calculates the optimal number of bits and hash functions - /// to achieve the desired false positive probability for a given number of items. - /// The parameters are validated when [`build()`](Self::build) is called. + /// Uses the standard Bloom filter sizing formulas to choose the requested number of bits and + /// hash functions. The parameters are validated when [`build()`](Self::build) is called. + /// + /// `max_items` is the expected maximum number of distinct items, not a hard insertion limit. + /// Inserting more distinct items remains valid but can increase the false-positive probability + /// beyond `fpp`. An `fpp` of `1.0` is accepted and creates the smallest allocation: 64 bits and + /// one hash function. /// /// # Arguments /// @@ -675,6 +688,9 @@ impl BloomFilterBuilder { /// The underlying storage is word-based, so the actual capacity is rounded /// up to the next multiple of 64 bits. /// + /// `num_bits` must be positive and fit the serialized Bloom filter format. `num_hashes` must be + /// in the range `1..=32767`. These constraints are checked by [`build()`](Self::build). + /// /// # Arguments /// /// * `num_bits`: Total number of bits in the filter. @@ -720,8 +736,13 @@ impl BloomFilterBuilder { /// /// # Errors /// - /// Returns an error if the configured accuracy or size parameters are outside their supported - /// ranges. + /// In accuracy mode, returns an error if `max_items` is zero, `fpp` is outside `(0.0, 1.0]`, or + /// the target requires more bits than the serialized format supports. + /// + /// In manual size mode, returns an error if `num_bits` is zero or exceeds the serialized format + /// limit, or if `num_hashes` is outside `1..=32767`. + /// + /// Valid configurations may still request more memory than the current process can allocate. pub fn build(self) -> Result { let (num_bits, num_hashes) = match self.mode { BloomFilterBuilderMode::Accuracy { max_items, fpp } => { @@ -729,12 +750,24 @@ impl BloomFilterBuilder { return Err(Error::invalid_argument("max_items must be greater than 0")); } if !(fpp > 0.0 && fpp <= 1.0) { - return Err(Error::invalid_argument( - "fpp must be between 0.0 and 1.0 (inclusive of 1.0)", - )); + return Err(Error::invalid_argument("fpp must be in (0.0, 1.0]")); } - let num_bits = Self::suggest_num_bits(max_items, fpp); - let num_hashes = Self::suggest_num_hashes_from_accuracy(max_items, num_bits); + + let n = max_items as f64; + let ln2_squared = std::f64::consts::LN_2 * std::f64::consts::LN_2; + let bits = (-n * fpp.ln() / ln2_squared).ceil(); + if bits > Self::MAX_NUM_BITS as f64 { + return Err(Error::invalid_argument(format!( + "target accuracy requires {bits:.0} bits, but at most {} are supported", + Self::MAX_NUM_BITS + ))); + } + + let num_bits = (bits as u64).max(Self::MIN_NUM_BITS); + let num_hashes = (num_bits as f64 / n * std::f64::consts::LN_2).ceil().clamp( + f64::from(Self::MIN_NUM_HASHES), + f64::from(Self::MAX_NUM_HASHES), + ) as u16; (num_bits, num_hashes) } BloomFilterBuilderMode::Size { @@ -770,74 +803,4 @@ impl BloomFilterBuilder { bit_array, }) } - - /// Suggests optimal number of bits given max items and target FPP. - /// - /// Formula: `m = -n * ln(p) / (ln(2)^2)` - /// where n = max_items, p = fpp - /// - /// # Examples - /// - /// ``` - /// use datasketches::bloom::BloomFilterBuilder; - /// - /// let bits = BloomFilterBuilder::suggest_num_bits(1000, 0.01); - /// assert!(bits > 9000 && bits < 10000); // ~9585 bits - /// ``` - pub fn suggest_num_bits(max_items: u64, fpp: f64) -> u64 { - let n = max_items as f64; - let p = fpp; - let ln2_squared = std::f64::consts::LN_2 * std::f64::consts::LN_2; - - let bits = (-n * p.ln() / ln2_squared).ceil() as u64; - - bits.clamp(Self::MIN_NUM_BITS, Self::MAX_NUM_BITS) - } - - /// Suggests optimal number of hash functions given max items and bit count. - /// - /// Formula: `k = (m/n) * ln(2)` - /// where m = num_bits, n = max_items - /// - /// # Examples - /// - /// ``` - /// use datasketches::bloom::BloomFilterBuilder; - /// - /// let hashes = BloomFilterBuilder::suggest_num_hashes_from_accuracy(1000, 10000); - /// assert_eq!(hashes, 7); // Optimal k ≈ 6.93 - /// ``` - pub fn suggest_num_hashes_from_accuracy(max_items: u64, num_bits: u64) -> u16 { - let m = num_bits as f64; - let n = max_items as f64; - - // Ceil to avoid selecting too few hashes. - let k = (m / n * std::f64::consts::LN_2).ceil(); - k.clamp( - f64::from(Self::MIN_NUM_HASHES), - f64::from(Self::MAX_NUM_HASHES), - ) as u16 - } - - /// Suggests optimal number of hash functions from target FPP. - /// - /// Formula: `k = -log2(p)` - /// where p = fpp - /// - /// # Examples - /// - /// ``` - /// use datasketches::bloom::BloomFilterBuilder; - /// - /// let hashes = BloomFilterBuilder::suggest_num_hashes_from_fpp(0.01); - /// assert_eq!(hashes, 7); // -log2(0.01) ≈ 6.64 - /// ``` - pub fn suggest_num_hashes_from_fpp(fpp: f64) -> u16 { - // Ceil to avoid selecting too few hashes. - let k = -fpp.log2(); - k.ceil().clamp( - f64::from(Self::MIN_NUM_HASHES), - f64::from(Self::MAX_NUM_HASHES), - ) as u16 - } } diff --git a/datasketches/src/countmin/sketch.rs b/datasketches/src/countmin/sketch.rs index 6745f8fe..484e6c4f 100644 --- a/datasketches/src/countmin/sketch.rs +++ b/datasketches/src/countmin/sketch.rs @@ -254,9 +254,9 @@ impl CountMinSketch { /// Merges another sketch into this one. /// - /// # Panics + /// # Errors /// - /// Panics if the sketches have incompatible configurations. + /// Returns an error if the sketches have different numbers of hashes, bucket counts, or seeds. /// /// # Examples /// @@ -269,22 +269,23 @@ impl CountMinSketch { /// left.update("apple"); /// right.update_with_weight("banana", 2); /// - /// left.merge(&right); + /// left.merge(&right).unwrap(); /// assert!(left.estimate("banana") >= 2); /// ``` - pub fn merge(&mut self, other: &CountMinSketch) { - if std::ptr::eq(self, other) { - panic!("Cannot merge a sketch with itself."); + pub fn merge(&mut self, other: &CountMinSketch) -> Result<(), Error> { + if self.num_hashes != other.num_hashes + || self.num_buckets != other.num_buckets + || self.seed != other.seed + { + return Err(Error::invalid_argument( + "Count-Min sketches must have matching numbers of hashes, bucket counts, and seeds", + )); } - assert_eq!(self.num_hashes, other.num_hashes); - assert_eq!(self.num_buckets, other.num_buckets); - assert_eq!(self.seed, other.seed); - assert_eq!(self.counts.len(), other.counts.len()); - let counts_len = self.counts.len(); - for i in 0..counts_len { - self.counts[i] = self.counts[i] + other.counts[i]; + for (count, other_count) in self.counts.iter_mut().zip(&other.counts) { + *count = *count + *other_count; } self.total_weight = self.total_weight + other.total_weight; + Ok(()) } /// Serializes this sketch into the DataSketches CountMin format. diff --git a/datasketches/src/cpc/sketch.rs b/datasketches/src/cpc/sketch.rs index 8ebaeb42..8599e46d 100644 --- a/datasketches/src/cpc/sketch.rs +++ b/datasketches/src/cpc/sketch.rs @@ -754,7 +754,7 @@ impl CpcSketch { table_num_entries, table_data_words ))); } - let k = 1usize << lg_k; + let k = 1 << lg_k; if has_window && window_data_words.saturating_mul(32) < k { return Err(Error::deserial(format!( "window data ({} words) is too short for lg_k = {lg_k}", @@ -880,14 +880,15 @@ impl CpcSketch { /// /// For small values of `n` the size can be much smaller. /// - /// # Panics + /// # Errors /// - /// Panics if `lg_k` is not in the range `[4, 26]`. - pub fn max_serialized_bytes(lg_k: u8) -> usize { - assert!( - (MIN_LG_K..=MAX_LG_K).contains(&lg_k), - "lg_k out of range; got {lg_k}", - ); + /// Returns an error if `lg_k` is not in the range `[4, 26]`. + pub fn max_serialized_bytes(lg_k: u8) -> Result { + if !(MIN_LG_K..=MAX_LG_K).contains(&lg_k) { + return Err(Error::invalid_argument(format!( + "lg_k must be in [{MIN_LG_K}, {MAX_LG_K}], got {lg_k}" + ))); + } // These empirical values for the 99.9th percentile of size in bytes were measured using // 100,000 trials. The value for each trial is the maximum of 5*16=80 measurements @@ -916,20 +917,21 @@ impl CpcSketch { 314656, // lg_k = 19 ]; - if lg_k <= EMPIRICAL_SIZE_MAX_LGK { + let max_bytes = if lg_k <= EMPIRICAL_SIZE_MAX_LGK { EMPIRICAL_MAX_SIZE_BYTES[(lg_k - MIN_LG_K) as usize] + MAX_PREAMBLE_SIZE_BYTES } else { let k = 1 << lg_k; ((EMPIRICAL_MAX_SIZE_FACTOR * k as f64) as usize) + MAX_PREAMBLE_SIZE_BYTES - } + }; + Ok(max_bytes) } } -// testing methods impl CpcSketch { /// Returns `true` if the sketch's internal state is valid. /// - /// This is primarily for testing and validation purposes. + /// This is intended for testing and validation purposes. + #[doc(hidden)] pub fn validate(&self) -> bool { let bit_matrix = self.build_bit_matrix(); let num_bits_set = count_bits_set_in_matrix(&bit_matrix); @@ -938,7 +940,8 @@ impl CpcSketch { /// Returns the number of coupons in the sketch. /// - /// This is primarily for testing and validation purposes. + /// This is intended for testing and validation purposes. + #[doc(hidden)] pub fn num_coupons(&self) -> u32 { self.num_coupons } diff --git a/datasketches/src/cpc/union.rs b/datasketches/src/cpc/union.rs index ebae9d0a..51742a30 100644 --- a/datasketches/src/cpc/union.rs +++ b/datasketches/src/cpc/union.rs @@ -133,8 +133,8 @@ impl CpcUnion { /// s2.update(&"banana"); /// /// let mut union = CpcUnion::new(12).unwrap(); - /// union.update(&s1); - /// union.update(&s2); + /// union.update(&s1).unwrap(); + /// union.update(&s2).unwrap(); /// /// let result = union.to_sketch(); /// assert_eq!(result.estimate().trunc(), 2.0); @@ -210,15 +210,21 @@ impl CpcUnion { /// Updates this union with a `CpcSketch`. /// - /// # Panics + /// # Errors /// - /// Panics if the seed of the provided sketch does not match the seed of this union. - pub fn update(&mut self, sketch: &CpcSketch) { - assert_eq!(self.seed, sketch.seed()); + /// Returns an error if the seed of the provided sketch does not match the seed of this union. + pub fn update(&mut self, sketch: &CpcSketch) -> Result<(), Error> { + if self.seed != sketch.seed() { + return Err(Error::invalid_argument(format!( + "CPC sketch seed must match union seed: expected {}, got {}", + self.seed, + sketch.seed() + ))); + } let flavor = sketch.flavor(); if flavor == Flavor::Empty { - return; + return Ok(()); } if sketch.lg_k() < self.lg_k { @@ -250,7 +256,7 @@ impl CpcUnion { // are equal. if old_flavor == Flavor::Empty && self.lg_k == sketch.lg_k() { *old_sketch = sketch.clone(); - return; + return Ok(()); } walk_table_updating_sketch(old_sketch, sketch.surprising_value_table()); @@ -263,7 +269,7 @@ impl CpcUnion { self.state = UnionState::BitMatrix(bit_matrix); } - return; + return Ok(()); } // If flavor is past SPARSE mode, the state must have been converted to bitMatrix. @@ -275,7 +281,7 @@ impl CpcUnion { if flavor == Flavor::Sparse { // [Case B] Sparse, bitMatrix valid, accumulator == null or_table_into_matrix(old_matrix, self.lg_k, sketch.surprising_value_table()); - return; + return Ok(()); } if matches!(flavor, Flavor::Hybrid | Flavor::Pinned) { @@ -289,7 +295,7 @@ impl CpcUnion { sketch.lg_k(), ); or_table_into_matrix(old_matrix, self.lg_k, sketch.surprising_value_table()); - return; + return Ok(()); } // [Case D] Sliding, bitMatrix valid, accumulator == null @@ -300,6 +306,7 @@ impl CpcUnion { or_matrix_into_matrix(old_matrix, self.lg_k, &src_matrix, sketch.lg_k()); } } + Ok(()) } fn reduce_k(&mut self, new_lg_k: u8) { @@ -349,19 +356,6 @@ impl CpcUnion { } } -// testing methods -impl CpcUnion { - /// Returns the number of coupons in the union. - /// - /// This is primarily for testing and validation purposes. - pub fn num_coupons(&self) -> u32 { - match &self.state { - UnionState::Accumulator(sketch) => sketch.num_coupons, - UnionState::BitMatrix(matrix) => count_bits_set_in_matrix(matrix), - } - } -} - fn or_window_into_matrix( dst_matrix: &mut [u64], dst_lg_k: u8, diff --git a/datasketches/src/frequencies/reverse_purge_item_hash_map.rs b/datasketches/src/frequencies/reverse_purge_item_hash_map.rs index f32e9ab1..ce3d62dc 100644 --- a/datasketches/src/frequencies/reverse_purge_item_hash_map.rs +++ b/datasketches/src/frequencies/reverse_purge_item_hash_map.rs @@ -141,7 +141,7 @@ impl ReversePurgeItemHashMap { pub fn purge(&mut self, sample_size: usize) -> u64 { let limit = sample_size.min(self.num_active).min(MAX_SAMPLE_SIZE); let mut samples = Vec::with_capacity(limit); - let mut i = 0usize; + let mut i = 0; while samples.len() < limit { if self.is_active(i) { samples.push(self.values[i]); diff --git a/datasketches/src/frequencies/sketch.rs b/datasketches/src/frequencies/sketch.rs index 0f9c4b08..7a520453 100644 --- a/datasketches/src/frequencies/sketch.rs +++ b/datasketches/src/frequencies/sketch.rs @@ -39,11 +39,12 @@ type SerializeItem = fn(&mut SketchBytes, &T); type DeserializeItems = fn(SketchSlice<'_>, usize) -> Result, Error>; const LG_MIN_MAP_SIZE: u8 = 3; +const MIN_MAP_SIZE: usize = 1 << LG_MIN_MAP_SIZE; // Java represents map sizes as positive `int` powers of two, while the C++ // implementation uses 32-bit table indices. Keep Rust configurations within // the same cross-language range. const LG_MAX_MAP_SIZE: u8 = 30; -const MAX_MAP_SIZE: usize = 1usize << LG_MAX_MAP_SIZE; +const MAX_MAP_SIZE: usize = 1 << LG_MAX_MAP_SIZE; const SAMPLE_SIZE: usize = 1024; const EPSILON_FACTOR: f64 = 3.5; const LOAD_FACTOR_NUMERATOR: usize = 3; @@ -51,7 +52,29 @@ const LOAD_FACTOR_DENOMINATOR: usize = 4; fn map_capacity_for_lg(lg_map_size: u8) -> usize { debug_assert!(lg_map_size <= LG_MAX_MAP_SIZE); - (1usize << lg_map_size) * LOAD_FACTOR_NUMERATOR / LOAD_FACTOR_DENOMINATOR + (1 << lg_map_size) * LOAD_FACTOR_NUMERATOR / LOAD_FACTOR_DENOMINATOR +} + +fn lg_for_max_map_size(max_map_size: usize) -> Result { + if !max_map_size.is_power_of_two() { + return Err(Error::invalid_argument("max_map_size must be a power of 2")); + } + if max_map_size < MIN_MAP_SIZE { + return Err(Error::invalid_argument(format!( + "max_map_size must be at least {MIN_MAP_SIZE}" + ))); + } + if max_map_size > MAX_MAP_SIZE { + return Err(Error::invalid_argument(format!( + "max_map_size must not exceed {MAX_MAP_SIZE}" + ))); + } + Ok(max_map_size.trailing_zeros() as u8) +} + +fn epsilon_for_lg(lg_max_map_size: u8) -> f64 { + debug_assert!((LG_MIN_MAP_SIZE..=LG_MAX_MAP_SIZE).contains(&lg_max_map_size)); + EPSILON_FACTOR / (1u64 << lg_max_map_size) as f64 } fn validate_lg_map_sizes(lg_max: u8, lg_cur: u8) -> Result<(), Error> { @@ -137,8 +160,8 @@ impl FrequentItemsSketch { /// /// # Errors /// - /// Returns an error if `max_map_size` is not a power of two or exceeds `2^30`, the maximum - /// supported by the cross-language format implementations. + /// Returns an error if `max_map_size` is not a power of two in the range `[8, 2^30]`. The upper + /// bound is the maximum supported by the cross-language format implementations. /// /// # Examples /// @@ -151,15 +174,7 @@ impl FrequentItemsSketch { /// assert_eq!(sketch.num_active_items(), 2); /// ``` pub fn new(max_map_size: usize) -> Result { - if !max_map_size.is_power_of_two() { - return Err(Error::invalid_argument("max_map_size must be a power of 2")); - } - if max_map_size > MAX_MAP_SIZE { - return Err(Error::invalid_argument(format!( - "max_map_size must not exceed {MAX_MAP_SIZE}" - ))); - } - let lg_max_map_size = max_map_size.trailing_zeros() as u8; + let lg_max_map_size = lg_for_max_map_size(max_map_size)?; Ok(Self::with_lg_map_sizes(lg_max_map_size, LG_MIN_MAP_SIZE)) } @@ -245,17 +260,43 @@ impl FrequentItemsSketch { /// Returns the epsilon error parameter for this sketch. pub fn epsilon(&self) -> f64 { - Self::epsilon_for_lg(self.lg_max_map_size) + epsilon_for_lg(self.lg_max_map_size) } - /// Returns the epsilon error parameter for the given `lg_max_map_size`. - pub fn epsilon_for_lg(lg_max_map_size: u8) -> f64 { - EPSILON_FACTOR / (1u64 << lg_max_map_size) as f64 + /// Returns the epsilon error parameter for the given maximum map size. + /// + /// # Errors + /// + /// Returns an error if `max_map_size` is not a power of two in the range `[8, 2^30]`. + /// + /// # Examples + /// + /// ``` + /// use datasketches::frequencies::FrequentItemsSketch; + /// + /// let epsilon = FrequentItemsSketch::::epsilon_for_max_map_size(1024).unwrap(); + /// assert_eq!(epsilon, 3.5 / 1024.0); + /// ``` + pub fn epsilon_for_max_map_size(max_map_size: usize) -> Result { + Ok(epsilon_for_lg(lg_for_max_map_size(max_map_size)?)) } - /// Returns the a priori error estimate. - pub fn apriori_error(lg_max_map_size: u8, estimated_total_weight: i64) -> f64 { - Self::epsilon_for_lg(lg_max_map_size) * estimated_total_weight as f64 + /// Returns the a priori error estimate for a maximum map size and estimated total weight. + /// + /// # Errors + /// + /// Returns an error if `max_map_size` is not a power of two in the range `[8, 2^30]`. + /// + /// # Examples + /// + /// ``` + /// use datasketches::frequencies::FrequentItemsSketch; + /// + /// let error = FrequentItemsSketch::::apriori_error(1024, 10_000).unwrap(); + /// assert_eq!(error, 3.5 / 1024.0 * 10_000.0); + /// ``` + pub fn apriori_error(max_map_size: usize, estimated_total_weight: u64) -> Result { + Ok(Self::epsilon_for_max_map_size(max_map_size)? * estimated_total_weight as f64) } /// Returns the maximum map capacity for this sketch. @@ -272,6 +313,11 @@ impl FrequentItemsSketch { self.cur_map_cap } + /// Returns the configured maximum map size. + pub fn max_map_size(&self) -> usize { + 1 << self.lg_max_map_size + } + /// Returns the configured `lg_max_map_size`. pub fn lg_max_map_size(&self) -> u8 { self.lg_max_map_size @@ -513,7 +559,7 @@ impl FrequentItemsSketch { lg_cur <= lg_max, "lg_cur_map_size must not exceed lg_max_map_size" ); - let map = ReversePurgeItemHashMap::new(1usize << lg_cur); + let map = ReversePurgeItemHashMap::new(1 << lg_cur); let cur_map_cap = map.capacity(); let max_map_cap = map_capacity_for_lg(lg_max); let sample_size = SAMPLE_SIZE.min(max_map_cap); diff --git a/datasketches/src/lib.rs b/datasketches/src/lib.rs index 066b2376..1f36e79a 100644 --- a/datasketches/src/lib.rs +++ b/datasketches/src/lib.rs @@ -17,11 +17,48 @@ //! # Apache® DataSketches™ Core Rust Library Component //! -//! The Sketching Core Library provides a range of stochastic streaming algorithms and closely -//! related Rust technologies that are particularly useful when integrating this technology into -//! systems that must deal with massive data. +//! This crate provides compact, mergeable summaries for answering queries over large data streams. +//! It implements a subset of the algorithms available in the other Apache DataSketches language +//! components. //! -//! This library is divided into modules that constitute distinct groups of functionality. +//! ## Enabling sketches +//! +//! Sketch implementations are opt-in Cargo features; this crate enables none by default. Enable +//! only the algorithms an application uses: +//! +//! ```text +//! cargo add datasketches --features hll,theta +//! ``` +//! +//! Each feature exposes a same-named module. For example, `hll` exposes `datasketches::hll` and +//! `tdigest` exposes `datasketches::tdigest`. +//! +//! ## Choosing a sketch +//! +//! * Use `bloom` for probabilistic membership queries. +//! * Use `countmin` for point-frequency estimates and `frequencies` for discovering heavy hitters. +//! * Use `hll` for fast distinct counts, `cpc` for compact serialized distinct counts, or `theta` +//! when set operations are required. +//! * Use `req` or `tdigest` for ranks and quantiles. REQ targets configurable high- or low-rank +//! accuracy; T-Digest emphasizes distribution tails. +//! * Use `tuple` when retained Theta keys need application-defined summaries. +//! +//! See each module's documentation for accuracy, memory, serialization, and update examples. +//! +//! ## Cross-language hashing +//! +//! Compatible serialization does not by itself make ordinary Rust [`Hash`](std::hash::Hash) +//! input compatible with Java, C++, or Go. Rust strings and slices include type-specific framing, +//! and short integers require different widening rules for different sketch families. When +//! sketches must represent the same updates across languages, use the wrappers in [`hash::value`]: +//! +//! * `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. +//! +//! Other DataSketches implementations skip empty strings rather than hashing them. Check for empty +//! input before updating when that cross-language behavior is required. #![cfg_attr(docsrs, feature(doc_cfg))] #![deny(missing_docs)] diff --git a/datasketches/src/req/sketch.rs b/datasketches/src/req/sketch.rs index e2f1d944..af93c6c4 100644 --- a/datasketches/src/req/sketch.rs +++ b/datasketches/src/req/sketch.rs @@ -438,38 +438,6 @@ where } } - /// Returns per-level info: `(level_index, num_items, capacity, weight)`. - /// Internal/test API; subject to change. - #[doc(hidden)] - pub fn level_info(&self) -> Vec<(usize, u32, u32, u64)> { - self.compactors - .iter() - .enumerate() - .map(|(i, c)| (i, c.num_items(), c.nominal_capacity(), c.weight())) - .collect() - } - - /// Total nominal capacity across all levels. Internal/test API. - #[doc(hidden)] - pub fn total_nominal_capacity(&self) -> u32 { - self.compactors.iter().map(|c| c.nominal_capacity()).sum() - } - - /// Total retained items across all levels. Internal/test API. - #[doc(hidden)] - pub fn total_retained_items(&self) -> u32 { - self.compactors.iter().map(|c| c.num_items()).sum() - } - - /// Sum of `level_items × level_weight` across compactors. Internal/test API. - #[doc(hidden)] - pub fn computed_total_weight(&self) -> u64 { - self.compactors - .iter() - .map(|c| c.num_items() as u64 * c.weight()) - .sum() - } - fn flags_byte(&self) -> u8 { let mut flags = 0u8; if self.is_empty() { @@ -498,7 +466,7 @@ where { // Fixed sketch preamble: 8 bytes (preamble_ints, serial_version, family, // flags, k(2), num_levels, num_raw_items). - let mut size = 8usize; + let mut size = 8; if self.is_empty() { return size; } diff --git a/datasketches/src/tdigest/sketch.rs b/datasketches/src/tdigest/sketch.rs index bd27a4c0..613c2c7d 100644 --- a/datasketches/src/tdigest/sketch.rs +++ b/datasketches/src/tdigest/sketch.rs @@ -498,91 +498,26 @@ impl TDigestMut { /// let mut sketch = TDigestMut::new(100).unwrap(); /// sketch.update(1.0); /// let bytes = sketch.serialize(); - /// let decoded = TDigestMut::deserialize(&bytes, false).unwrap(); + /// let decoded = TDigestMut::deserialize(&bytes).unwrap(); /// assert_eq!(decoded.max_value(), Some(1.0)); /// ``` pub fn serialize(&mut self) -> Vec { self.compress(); - let centroids = self.buffer.compressed_centroids(); - - let mut total_size = 0; - if self.is_empty() || self.is_single_value() { - // 1 byte preamble - // + 1 byte serial version - // + 1 byte family - // + 2 bytes k - // + 1 byte flags - // + 2 bytes unused - total_size += size_of::(); - } else { - // all of the above - // + 4 bytes num centroids - // + 4 bytes num buffered - total_size += size_of::() * 2; - } - if self.is_empty() { - // nothing more - } else if self.is_single_value() { - // + 8 bytes single value - total_size += size_of::(); - } else { - // + 8 bytes min - // + 8 bytes max - total_size += size_of::() * 2; - // + (8+8) bytes per centroid - total_size += centroids.len() * (size_of::() + size_of::()); - } - - let mut bytes = SketchBytes::with_capacity(total_size); - bytes.write_u8(match self.total_weight() { - 0 => PREAMBLE_LONGS_EMPTY_OR_SINGLE, - 1 => PREAMBLE_LONGS_EMPTY_OR_SINGLE, - _ => PREAMBLE_LONGS_MULTIPLE, - }); - bytes.write_u8(SERIAL_VERSION); - bytes.write_u8(Family::TDIGEST.id); - bytes.write_u16_le(self.k); - bytes.write_u8({ - let mut flags = 0; - if self.is_empty() { - flags |= FLAGS_IS_EMPTY; - } - if self.is_single_value() { - flags |= FLAGS_IS_SINGLE_VALUE; - } - if self.reverse_merge { - flags |= FLAGS_REVERSE_MERGE; - } - flags - }); - bytes.write_u16_le(0); // unused - if self.is_empty() { - return bytes.into_bytes(); - } - if self.is_single_value() { - bytes.write_f64_le(self.min); - return bytes.into_bytes(); - } - bytes.write_u32_le(centroids.len() as u32); - bytes.write_u32_le(0); // unused - bytes.write_f64_le(self.min); - bytes.write_f64_le(self.max); - for centroid in centroids { - bytes.write_f64_le(centroid.mean); - bytes.write_u64_le(centroid.weight.get()); - } - bytes.into_bytes() + serialize_compressed( + self.k, + self.reverse_merge, + self.min, + self.max, + self.buffer.compressed_centroids(), + self.compressed_weight, + ) } - /// Deserializes a mutable t-digest from bytes. - /// - /// Supports reading compact format with (float, int) centroids as opposed to (double, long) to - /// represent (mean, weight). [^1] - /// - /// Supports reading format of the reference implementation (auto-detected) [^2]. + /// Deserializes a mutable t-digest from the standard double-precision format. /// - /// [^1]: This is to support reading the `tdigest` format from the C++ implementation. - /// [^2]: + /// The format of the [reference implementation](https://github.com/tdunning/t-digest) is + /// auto-detected. Use [`deserialize_f32()`](Self::deserialize_f32) for the compact + /// DataSketches C++ `tdigest` format. /// /// # Examples /// @@ -593,10 +528,23 @@ impl TDigestMut { /// sketch.update(1.0); /// sketch.update(2.0); /// let bytes = sketch.serialize(); - /// let decoded = TDigestMut::deserialize(&bytes, false).unwrap(); + /// let decoded = TDigestMut::deserialize(&bytes).unwrap(); /// assert_eq!(decoded.max_value(), Some(2.0)); /// ``` - pub fn deserialize(bytes: &[u8], is_f32: bool) -> Result { + pub fn deserialize(bytes: &[u8]) -> Result { + Self::deserialize_impl(bytes, false) + } + + /// Deserializes a mutable t-digest from the compact single-precision DataSketches format. + /// + /// This format stores centroid means and weights as `(f32, u32)` and is emitted by the C++ + /// `tdigest` implementation. Its header does not identify the scalar width, so callers + /// must select this entry point explicitly. + pub fn deserialize_f32(bytes: &[u8]) -> Result { + Self::deserialize_impl(bytes, true) + } + + fn deserialize_impl(bytes: &[u8], is_f32: bool) -> Result { let mut cursor = SketchSlice::new(bytes); let preamble_longs = cursor @@ -853,10 +801,6 @@ impl TDigestMut { } } - fn is_single_value(&self) -> bool { - self.total_weight() == 1 - } - /// Processes unmerged values and merges centroids if needed. fn compress(&mut self) { let additional_weight = self.buffer.unmerged_len() as u64; @@ -950,6 +894,71 @@ impl TDigestMut { } } +fn serialize_compressed( + k: u16, + reverse_merge: bool, + min: f64, + max: f64, + centroids: &[Centroid], + total_weight: u64, +) -> Vec { + let is_empty = centroids.is_empty(); + let is_single_value = total_weight == 1; + let mut total_size = if is_empty || is_single_value { + // Preamble, serial version, family, k, flags, and two unused bytes. + size_of::() + } else { + // The short header plus centroid and buffered-value counts. + size_of::() * 2 + }; + if is_single_value { + total_size += size_of::(); + } else if !is_empty { + total_size += size_of::() * 2; + total_size += centroids.len() * (size_of::() + size_of::()); + } + + let mut bytes = SketchBytes::with_capacity(total_size); + bytes.write_u8(if is_empty || is_single_value { + PREAMBLE_LONGS_EMPTY_OR_SINGLE + } else { + PREAMBLE_LONGS_MULTIPLE + }); + bytes.write_u8(SERIAL_VERSION); + bytes.write_u8(Family::TDIGEST.id); + bytes.write_u16_le(k); + bytes.write_u8({ + let mut flags = 0; + if is_empty { + flags |= FLAGS_IS_EMPTY; + } + if is_single_value { + flags |= FLAGS_IS_SINGLE_VALUE; + } + if reverse_merge { + flags |= FLAGS_REVERSE_MERGE; + } + flags + }); + bytes.write_u16_le(0); // unused + if is_empty { + return bytes.into_bytes(); + } + if is_single_value { + bytes.write_f64_le(min); + return bytes.into_bytes(); + } + bytes.write_u32_le(centroids.len() as u32); + bytes.write_u32_le(0); // no buffered values + bytes.write_f64_le(min); + bytes.write_f64_le(max); + for centroid in centroids { + bytes.write_f64_le(centroid.mean); + bytes.write_u64_le(centroid.weight.get()); + } + bytes.into_bytes() +} + /// Immutable (frozen) T-Digest sketch for estimating quantiles and ranks. /// /// See the [module level documentation](super) for more. @@ -998,6 +1007,46 @@ impl TDigest { self.centroids_weight } + /// Serializes this immutable t-digest to bytes. + /// + /// # Examples + /// + /// ``` + /// use datasketches::tdigest::TDigest; + /// use datasketches::tdigest::TDigestMut; + /// + /// let mut sketch = TDigestMut::new(100).unwrap(); + /// sketch.update(1.0); + /// let digest = sketch.freeze(); + /// let bytes = digest.serialize(); + /// let decoded = TDigest::deserialize(&bytes).unwrap(); + /// assert_eq!(decoded.max_value(), Some(1.0)); + /// ``` + pub fn serialize(&self) -> Vec { + serialize_compressed( + self.k, + self.reverse_merge, + self.min, + self.max, + &self.centroids, + self.centroids_weight, + ) + } + + /// Deserializes an immutable t-digest from the standard double-precision format. + /// + /// The format of the [reference implementation](https://github.com/tdunning/t-digest) is + /// auto-detected. Use [`deserialize_f32()`](Self::deserialize_f32) for the compact + /// DataSketches C++ `tdigest` format. + pub fn deserialize(bytes: &[u8]) -> Result { + Ok(TDigestMut::deserialize(bytes)?.freeze()) + } + + /// Deserializes an immutable t-digest from the compact single-precision DataSketches format. + pub fn deserialize_f32(bytes: &[u8]) -> Result { + Ok(TDigestMut::deserialize_f32(bytes)?.freeze()) + } + fn view(&self) -> TDigestView<'_> { TDigestView { min: self.min, @@ -1021,7 +1070,7 @@ impl TDigest { /// stream given the split points. The value at array position j of the returned CDF array /// is the sum of the returned values in positions 0 through j of the returned PMF array. /// This can be viewed as array of ranks of the given split points plus one more value that - /// is always 1. + /// is always 1. An empty `split_points` slice returns the single value `[1.0]`. /// /// Returns `None` if this t-digest is empty. /// @@ -1059,6 +1108,7 @@ impl TDigest { /// /// An array of m+1 doubles each of which is an approximation to the fraction of the input /// stream values (the mass) that fall into one of those intervals. + /// An empty `split_points` slice returns the single value `[1.0]`. /// /// Returns `None` if this t-digest is empty. /// @@ -1380,15 +1430,10 @@ impl TDigestView<'_> { /// They must be unique, monotonically increasing and not NaN. #[track_caller] fn check_split_points(split_points: &[f64]) { - let len = split_points.len(); - if len == 1 && split_points[0].is_nan() { + if split_points.iter().any(|split_point| split_point.is_nan()) { panic!("split_points must not contain NaN values: {split_points:?}"); } - for i in 0..len - 1 { - if split_points[i] < split_points[i + 1] { - // we must use this positive condition because NaN comparisons are always false - continue; - } + if !split_points.windows(2).all(|pair| pair[0] < pair[1]) { panic!("split_points must be unique and monotonically increasing: {split_points:?}"); } } diff --git a/datasketches/src/thetafamily/tuple/mod.rs b/datasketches/src/thetafamily/tuple/mod.rs index c057ee54..9c3fcf12 100644 --- a/datasketches/src/thetafamily/tuple/mod.rs +++ b/datasketches/src/thetafamily/tuple/mod.rs @@ -50,7 +50,6 @@ mod sketch; mod union; pub use self::a_not_b::TupleANotB; -pub use self::hash_table::TupleEntry; pub use self::intersection::TupleIntersection; pub use self::jaccard_similarity::TupleJaccardSimilarity; pub use self::policy::DefaultUnionPolicy; diff --git a/datasketches/src/thetafamily/tuple/sketch.rs b/datasketches/src/thetafamily/tuple/sketch.rs index 76494ecb..e13a462b 100644 --- a/datasketches/src/thetafamily/tuple/sketch.rs +++ b/datasketches/src/thetafamily/tuple/sketch.rs @@ -414,9 +414,8 @@ where /// Compact (immutable) Tuple sketch. /// -/// This is the serialization-friendly form: a compact array of retained [`TupleEntry`] values -/// (hash plus summary) plus theta and a 16-bit seed hash. It can be ordered (sorted ascending by -/// hash) or unordered. +/// This is the serialization-friendly form: a compact array of retained hash-summary pairs plus +/// theta and a 16-bit seed hash. It can be ordered (sorted ascending by hash) or unordered. #[derive(Clone, Debug)] pub struct CompactTupleSketch { entries: Vec>, @@ -677,7 +676,7 @@ impl CompactTupleSketch { let mut theta = MAX_THETA; let num_entries = if pre_longs == 1 { - 1usize + 1 } else { let n = cursor .read_u32_le() diff --git a/tests-integration/tests/bloom_test/sketch.rs b/tests-integration/tests/bloom_test/sketch.rs index 15f44a96..3032fd99 100644 --- a/tests-integration/tests/bloom_test/sketch.rs +++ b/tests-integration/tests/bloom_test/sketch.rs @@ -68,14 +68,14 @@ fn test_union_and_intersection() { let right_bits = right.bits_used(); let mut intersection = left.clone(); - intersection.intersect(&right); + intersection.intersect(&right).unwrap(); assert!(intersection.contains(&"shared")); let intersection_bits = intersection.bits_used(); assert_that!(intersection_bits, le(left_bits)); assert_that!(intersection_bits, le(right_bits)); let mut union = left; - union.union(&right); + union.union(&right).unwrap(); assert!(union.contains(&"shared")); assert!(union.contains(&"left")); assert!(union.contains(&"right")); @@ -122,25 +122,25 @@ fn test_compatibility_checks_all_configuration() { } #[test] -#[should_panic(expected = "Cannot union incompatible Bloom filters")] fn test_union_rejects_incompatible_filters() { let mut left = filter(); let right = BloomFilterBuilder::with_size(NUM_BITS, NUM_HASHES) .seed(SEED + 1) .build() .unwrap(); - left.union(&right); + let error = left.union(&right).unwrap_err(); + assert_eq!(error.kind(), ErrorKind::InvalidArgument); } #[test] -#[should_panic(expected = "Cannot intersect incompatible Bloom filters")] fn test_intersection_rejects_incompatible_filters() { let mut left = filter(); let right = BloomFilterBuilder::with_size(NUM_BITS, NUM_HASHES) .seed(SEED + 1) .build() .unwrap(); - left.intersect(&right); + let error = left.intersect(&right).unwrap_err(); + assert_eq!(error.kind(), ErrorKind::InvalidArgument); } #[test] @@ -159,10 +159,19 @@ fn test_accuracy_builder_rejects_zero_items_at_build() { #[test] fn test_accuracy_builder_rejects_invalid_probability_at_build() { - let error = BloomFilterBuilder::with_accuracy(100, 1.5) - .build() - .unwrap_err(); - assert_eq!(error.kind(), ErrorKind::InvalidArgument); + for fpp in [0.0, 1.5, f64::NAN] { + let error = BloomFilterBuilder::with_accuracy(100, fpp) + .build() + .unwrap_err(); + assert_eq!(error.kind(), ErrorKind::InvalidArgument); + } +} + +#[test] +fn test_accuracy_builder_accepts_one_probability() { + let filter = BloomFilterBuilder::with_accuracy(100, 1.0).build().unwrap(); + assert_eq!(filter.capacity(), 64); + assert_eq!(filter.num_hashes(), 1); } #[test] @@ -176,3 +185,11 @@ fn test_size_builder_rejects_zero_hashes_at_build() { let error = BloomFilterBuilder::with_size(128, 0).build().unwrap_err(); assert_eq!(error.kind(), ErrorKind::InvalidArgument); } + +#[test] +fn test_accuracy_builder_rejects_unrepresentable_target() { + let error = BloomFilterBuilder::with_accuracy(u64::MAX, 0.01) + .build() + .unwrap_err(); + assert_eq!(error.kind(), ErrorKind::InvalidArgument); +} diff --git a/tests-integration/tests/countmin_test/sketch.rs b/tests-integration/tests/countmin_test/sketch.rs index 752ec886..d22cf30c 100644 --- a/tests-integration/tests/countmin_test/sketch.rs +++ b/tests-integration/tests/countmin_test/sketch.rs @@ -218,7 +218,7 @@ fn test_merge() { right.update("a"); right.update("b"); } - left.merge(&right); + left.merge(&right).unwrap(); assert_eq!(left.total_weight(), 18); assert_that!(left.estimate("a"), ge(14)); assert_that!(left.estimate("b"), ge(4)); @@ -272,11 +272,11 @@ fn test_invalid_buckets_return_error() { } #[test] -#[should_panic] fn test_merge_incompatible() { let mut left = CountMinSketch::::new(3, 64).unwrap(); let right = CountMinSketch::::new(2, 64).unwrap(); - left.merge(&right); + let error = left.merge(&right).unwrap_err(); + assert_eq!(error.kind(), ErrorKind::InvalidArgument); } #[test] diff --git a/tests-integration/tests/cpc_test/union.rs b/tests-integration/tests/cpc_test/union.rs index 54fb1086..91801652 100644 --- a/tests-integration/tests/cpc_test/union.rs +++ b/tests-integration/tests/cpc_test/union.rs @@ -36,14 +36,14 @@ fn test_two_values() { let mut sketch = CpcSketch::new(11).unwrap(); sketch.update(1); let mut union = CpcUnion::new(11).unwrap(); - union.update(&sketch); + union.update(&sketch).unwrap(); let result = union.to_sketch(); assert!(!result.is_empty()); assert_eq!(result.estimate(), 1.0); sketch.update(2); - union.update(&sketch); + union.update(&sketch).unwrap(); let result = union.to_sketch(); assert!(!result.is_empty()); assert_that!( @@ -60,7 +60,7 @@ fn test_custom_seed() { sketch.update(3); let mut union = CpcUnion::with_seed(11, 123).unwrap(); - union.update(&sketch); + union.update(&sketch).unwrap(); let result = union.to_sketch(); assert!(!result.is_empty()); assert_that!( @@ -70,7 +70,6 @@ fn test_custom_seed() { } #[test] -#[should_panic] fn test_custom_seed_mismatch() { let mut sketch = CpcSketch::with_seed(11, 123).unwrap(); sketch.update(1); @@ -78,26 +77,27 @@ fn test_custom_seed_mismatch() { sketch.update(3); let mut union = CpcUnion::with_seed(11, 234).unwrap(); - union.update(&sketch); + let error = union.update(&sketch).unwrap_err(); + assert_eq!(error.kind(), ErrorKind::InvalidArgument); } #[test] -fn test_large_values() { +fn test_sliding_union_matches_single_sketch() { let mut key = 0; let mut sketch = CpcSketch::new(11).unwrap(); let mut union = CpcUnion::new(11).unwrap(); - for _ in 0..1000 { + for _ in 0..32 { let mut tmp = CpcSketch::new(11).unwrap(); - for _ in 0..10000 { + for _ in 0..8192 { sketch.update(key); tmp.update(key); key += 1; } - union.update(&tmp); + union.update(&tmp).unwrap(); } let result = union.to_sketch(); assert!(!result.is_empty()); - assert_eq!(result.num_coupons(), union.num_coupons()); + assert!(result.num_coupons() >= 27 * (1 << 11) / 8); let estimate = sketch.estimate(); assert_that!( result.estimate(), @@ -112,7 +112,7 @@ fn test_reduce_k_empty() { sketch.update(i); } let mut union = CpcUnion::new(12).unwrap(); - union.update(&sketch); + union.update(&sketch).unwrap(); let result = union.to_sketch(); assert_eq!(result.lg_k(), 11); assert_that!( @@ -129,13 +129,13 @@ fn test_reduce_k_sparse() { for i in 0..100 { sketch12.update(i); } - union.update(&sketch12); + union.update(&sketch12).unwrap(); let mut sketch11 = CpcSketch::new(11).unwrap(); for i in 0..1000 { sketch11.update(i); } - union.update(&sketch11); + union.update(&sketch11).unwrap(); let result = union.to_sketch(); assert_eq!(result.lg_k(), 11); @@ -153,13 +153,13 @@ fn test_reduce_k_window() { for i in 0..500 { sketch12.update(i); } - union.update(&sketch12); + union.update(&sketch12).unwrap(); let mut sketch11 = CpcSketch::new(11).unwrap(); for i in 0..1000 { sketch11.update(i); } - union.update(&sketch11); + union.update(&sketch11).unwrap(); let result = union.to_sketch(); assert_eq!(result.lg_k(), 11); @@ -190,6 +190,6 @@ fn test_union_estimated_size() { for i in 0..1000 { sketch.update(i); } - union.update(&sketch); + union.update(&sketch).unwrap(); assert_eq!(union.estimated_size(), 16496); } diff --git a/tests-integration/tests/cpc_test/update.rs b/tests-integration/tests/cpc_test/update.rs index e7a22ebf..ab806f6e 100644 --- a/tests-integration/tests/cpc_test/update.rs +++ b/tests-integration/tests/cpc_test/update.rs @@ -17,6 +17,7 @@ use datasketches::common::NumStdDev; use datasketches::cpc::CpcSketch; +use datasketches::error::ErrorKind; use googletest::assert_that; use googletest::prelude::ge; use googletest::prelude::le; @@ -60,3 +61,12 @@ fn test_many_values() { assert_that!(sketch.estimate(), le(sketch.upper_bound(NumStdDev::One))); assert!(sketch.validate()); } + +#[test] +fn test_max_serialized_bytes_validates_lg_k() { + assert!(CpcSketch::max_serialized_bytes(11).unwrap() > 0); + for lg_k in [3, 27] { + let error = CpcSketch::max_serialized_bytes(lg_k).unwrap_err(); + assert_eq!(error.kind(), ErrorKind::InvalidArgument); + } +} diff --git a/tests-integration/tests/cpc_test/wrapper.rs b/tests-integration/tests/cpc_test/wrapper.rs index 2a1367f9..24d118e7 100644 --- a/tests-integration/tests/cpc_test/wrapper.rs +++ b/tests-integration/tests/cpc_test/wrapper.rs @@ -51,8 +51,8 @@ fn test_cpc_wrapper() { assert_that!(concat_wrapper.upper_bound(NumStdDev::Two), eq(dst_ub)); let mut union = CpcUnion::new(lg_k).unwrap(); - union.update(&sk1); - union.update(&sk2); + union.update(&sk1).unwrap(); + union.update(&sk2).unwrap(); let merged = union.to_sketch(); let merged_est = merged.estimate(); let merged_lb = merged.lower_bound(NumStdDev::Two); diff --git a/tests-integration/tests/frequencies_test/update.rs b/tests-integration/tests/frequencies_test/update.rs index 09dbe48d..9c1ed001 100644 --- a/tests-integration/tests/frequencies_test/update.rs +++ b/tests-integration/tests/frequencies_test/update.rs @@ -69,20 +69,27 @@ fn test_capacity_and_epsilon_helpers() { let longs: FrequentItemsSketch = FrequentItemsSketch::new(8).unwrap(); assert_eq!(longs.current_map_capacity(), 6); assert_eq!(longs.maximum_map_capacity(), 6); + assert_eq!(longs.max_map_size(), 8); assert_eq!(longs.lg_cur_map_size(), 3); assert_eq!(longs.lg_max_map_size(), 3); - let epsilon = FrequentItemsSketch::::epsilon_for_lg(10); + let epsilon = FrequentItemsSketch::::epsilon_for_max_map_size(1024).unwrap(); let expected = 3.5 / 1024.0; assert_that!(epsilon, near(expected, 1e-12)); - let apriori = FrequentItemsSketch::::apriori_error(10, 10_000); + let apriori = FrequentItemsSketch::::apriori_error(1024, 10_000).unwrap(); assert_that!(apriori, near(expected * 10_000.0, 1e-9)); + let invalid_epsilon = FrequentItemsSketch::::epsilon_for_max_map_size(6).unwrap_err(); + assert_eq!(invalid_epsilon.kind(), ErrorKind::InvalidArgument); + let invalid_apriori = FrequentItemsSketch::::apriori_error(4, 10_000).unwrap_err(); + assert_eq!(invalid_apriori.kind(), ErrorKind::InvalidArgument); + let items: FrequentItemsSketch = FrequentItemsSketch::new(1024).unwrap(); assert_that!(items.epsilon(), near(expected, 1e-12)); assert_eq!(items.current_map_capacity(), 6); assert_eq!(items.maximum_map_capacity(), 768); + assert_eq!(items.max_map_size(), 1024); assert_eq!(items.lg_max_map_size(), 10); } @@ -551,20 +558,16 @@ fn test_longs_reset() { } #[test] -fn test_longs_invalid_map_size_returns_error() { - let error = FrequentItemsSketch::::new(6).unwrap_err(); - assert_eq!(error.kind(), ErrorKind::InvalidArgument); -} - -#[test] -fn test_items_invalid_map_size_returns_error() { - let error = FrequentItemsSketch::::new(6).unwrap_err(); - assert_eq!(error.kind(), ErrorKind::InvalidArgument); +fn test_invalid_map_size_returns_error() { + for max_map_size in [1, 2, 4, 6] { + let error = FrequentItemsSketch::::new(max_map_size).unwrap_err(); + assert_eq!(error.kind(), ErrorKind::InvalidArgument); + } } #[test] fn test_map_size_above_cross_language_limit_returns_error() { - let error = FrequentItemsSketch::::new(1usize << 31).unwrap_err(); + let error = FrequentItemsSketch::::new(1 << 31).unwrap_err(); assert_eq!(error.kind(), ErrorKind::InvalidArgument); } diff --git a/tests-integration/tests/serde_tests/tdigest.rs b/tests-integration/tests/serde_tests/tdigest.rs index 63b0eb0b..c5bd0fe1 100644 --- a/tests-integration/tests/serde_tests/tdigest.rs +++ b/tests-integration/tests/serde_tests/tdigest.rs @@ -18,6 +18,7 @@ use std::fs; use std::path::PathBuf; +use datasketches::tdigest::TDigest; use datasketches::tdigest::TDigestMut; use googletest::assert_that; use googletest::prelude::all; @@ -46,7 +47,12 @@ fn patterned_digest(k: u16, len: usize, salt: usize) -> TDigestMut { fn test_sketch_file(path: PathBuf, n: u64, with_buffer: bool, is_f32: bool) { let bytes = fs::read(&path).unwrap(); - let td = TDigestMut::deserialize(&bytes, is_f32).unwrap(); + let td = if is_f32 { + TDigestMut::deserialize_f32(&bytes) + } else { + TDigestMut::deserialize(&bytes) + } + .unwrap(); let td = td.freeze(); let path = path.display(); @@ -111,7 +117,7 @@ fn test_deserialize_from_reference_implementation() { ] { let path = serialization_test_data("reference_files", filename); let bytes = fs::read(&path).unwrap(); - let td = TDigestMut::deserialize(&bytes, false).unwrap(); + let td = TDigestMut::deserialize(&bytes).unwrap(); let td = td.freeze(); let n = 10000; @@ -174,7 +180,7 @@ fn test_empty() { assert_eq!(bytes.len(), 8); let td = td.freeze(); - let deserialized_td = TDigestMut::deserialize(&bytes, false).unwrap(); + let deserialized_td = TDigestMut::deserialize(&bytes).unwrap(); let deserialized_td = deserialized_td.freeze(); assert_eq!(td.k(), deserialized_td.k()); assert_eq!(td.total_weight(), deserialized_td.total_weight()); @@ -190,7 +196,7 @@ fn test_single_value() { let bytes = td.serialize(); assert_eq!(bytes.len(), 16); - let deserialized_td = TDigestMut::deserialize(&bytes, false).unwrap(); + let deserialized_td = TDigestMut::deserialize(&bytes).unwrap(); let deserialized_td = deserialized_td.freeze(); assert_eq!(deserialized_td.k(), 200); assert_eq!(deserialized_td.total_weight(), 1); @@ -210,7 +216,7 @@ fn test_many_values() { assert_eq!(bytes.len(), 1584); let td = td.freeze(); - let deserialized_td = TDigestMut::deserialize(&bytes, false).unwrap(); + let deserialized_td = TDigestMut::deserialize(&bytes).unwrap(); let deserialized_td = deserialized_td.freeze(); assert_eq!(td.k(), deserialized_td.k()); assert_eq!(td.total_weight(), deserialized_td.total_weight()); @@ -221,6 +227,21 @@ fn test_many_values() { assert_eq!(td.quantile(0.5), deserialized_td.quantile(0.5)); } +#[test] +fn test_frozen_roundtrip() { + let tdigest = patterned_digest(100, 1000, 7); + let expected = tdigest.freeze(); + + let bytes = expected.serialize(); + let actual = TDigest::deserialize(&bytes).unwrap(); + + assert_eq!(actual.k(), expected.k()); + assert_eq!(actual.total_weight(), expected.total_weight()); + assert_eq!(actual.min_value(), expected.min_value()); + assert_eq!(actual.max_value(), expected.max_value()); + assert_eq!(actual.quantile(0.5), expected.quantile(0.5)); +} + #[test] fn test_serialized_bytes_stable_for_full_and_merged_digests() { let mut full_buffer = patterned_digest(200, 1_641, 0); @@ -252,10 +273,10 @@ fn test_serialized_bytes_stable_for_full_and_merged_digests() { let mut left = patterned_digest(10, 199, 2); let left = left.serialize(); - let mut left = TDigestMut::deserialize(&left, false).unwrap(); + let mut left = TDigestMut::deserialize(&left).unwrap(); let mut right = patterned_digest(10, 199, 3); let right = right.serialize(); - let right = TDigestMut::deserialize(&right, false).unwrap(); + let right = TDigestMut::deserialize(&right).unwrap(); left.merge(&right); let bytes = left.serialize(); assert_eq!(bytes.len(), 272); @@ -276,7 +297,7 @@ fn test_updates_normalize_overfull_deserialized_buffer_without_centroids() { bytes.extend_from_slice(&10_f64.to_le_bytes()); } - let mut tdigest = TDigestMut::deserialize(&bytes, false).unwrap(); + let mut tdigest = TDigestMut::deserialize(&bytes).unwrap(); for _ in 0..10_000 { tdigest.update(10.0); } @@ -290,7 +311,7 @@ fn test_updates_normalize_overfull_deserialized_buffer_without_centroids() { let serialized = tdigest.serialize(); assert_eq!(&serialized[12..16], &0_u32.to_le_bytes()); - let roundtrip = TDigestMut::deserialize(&serialized, false).unwrap(); + let roundtrip = TDigestMut::deserialize(&serialized).unwrap(); assert_eq!(roundtrip.total_weight(), 10_841); assert_eq!(roundtrip.min_value(), Some(1.0)); assert_eq!(roundtrip.max_value(), Some(10.0)); @@ -309,7 +330,7 @@ fn test_updates_normalize_overfull_deserialized_mixed_buffer() { bytes.extend_from_slice(&1_000_f64.to_le_bytes()); } - let mut tdigest = TDigestMut::deserialize(&bytes, false).unwrap(); + let mut tdigest = TDigestMut::deserialize(&bytes).unwrap(); for _ in 0..10_000 { tdigest.update(1_000.0); } @@ -323,7 +344,7 @@ fn test_updates_normalize_overfull_deserialized_mixed_buffer() { let serialized = tdigest.serialize(); assert_eq!(&serialized[12..16], &0_u32.to_le_bytes()); - let roundtrip = TDigestMut::deserialize(&serialized, false).unwrap(); + let roundtrip = TDigestMut::deserialize(&serialized).unwrap(); assert_eq!(roundtrip.total_weight(), 11_681); assert_eq!(roundtrip.min_value(), Some(1.0)); assert_eq!(roundtrip.max_value(), Some(1_000.0)); @@ -338,7 +359,7 @@ fn test_deserialize_rejects_truncated_large_payload_before_allocation() { bytes[8..12].copy_from_slice(&u32::MAX.to_le_bytes()); bytes[12..16].copy_from_slice(&u32::MAX.to_le_bytes()); - assert!(TDigestMut::deserialize(&bytes, false).is_err()); + assert!(TDigestMut::deserialize(&bytes).is_err()); } #[test] @@ -354,7 +375,7 @@ fn test_large_weights_produce_finite_extreme_quantile() { bytes[40..48].copy_from_slice(&((1_u64 << 52) - 1).to_le_bytes()); bytes[56..64].copy_from_slice(&(1_u64 << 52).to_le_bytes()); - let mut tdigest = TDigestMut::deserialize(&bytes, false).unwrap(); + let mut tdigest = TDigestMut::deserialize(&bytes).unwrap(); let quantile = tdigest.quantile(0.25).unwrap(); assert_that!(quantile, all!(is_finite(), ge(lower), le(f64::MAX))); } diff --git a/tests-integration/tests/tdigest_test/sketch.rs b/tests-integration/tests/tdigest_test/sketch.rs index 36732cf2..c0cced92 100644 --- a/tests-integration/tests/tdigest_test/sketch.rs +++ b/tests-integration/tests/tdigest_test/sketch.rs @@ -68,6 +68,19 @@ fn test_one_value() { assert_eq!(tdigest.quantile(1.0), Some(1.0)); } +#[test] +fn test_empty_split_points_define_one_bin() { + let mut tdigest = TDigestMut::new(100).unwrap(); + tdigest.update(1.0); + + assert_eq!(tdigest.cdf(&[]), Some(vec![1.0])); + assert_eq!(tdigest.pmf(&[]), Some(vec![1.0])); + + let tdigest = tdigest.freeze(); + assert_eq!(tdigest.cdf(&[]), Some(vec![1.0])); + assert_eq!(tdigest.pmf(&[]), Some(vec![1.0])); +} + #[test] fn test_maximum_k() { let mut tdigest = TDigestMut::new(u16::MAX).unwrap();