This is an automated email from the ASF dual-hosted git repository.
tisonkun pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datasketches-rust.git
The following commit(s) were added to refs/heads/main by this push:
new 7f4c145 fix(frequencies): define emptiness by stream weight (#281)
7f4c145 is described below
commit 7f4c145fe9abbf186c065c48b771d4fb61252a21
Author: tison <[email protected]>
AuthorDate: Sat Sep 26 16:44:30 2026 +0800
fix(frequencies): define emptiness by stream weight (#281)
---
CHANGELOG.md | 2 +
datasketches/src/frequencies/serialization.rs | 4 +-
datasketches/src/frequencies/sketch.rs | 74 +++++++++----
tests-integration/tests/frequencies_test/update.rs | 80 ++++++++++++--
tests-integration/tests/serde_tests/frequencies.rs | 118 ++++++++++++++-------
5 files changed, 207 insertions(+), 71 deletions(-)
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 6ea1c96..7144ea5 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -6,6 +6,7 @@ All significant changes to this project will be documented in
this file.
### Breaking changes
+* `FrequentItemsSketch::is_empty` now checks whether the total stream weight
is zero. A sketch whose counters were all removed by a purge remains non-empty;
use `num_active_items() == 0` to check whether any items are retained.
* Remove `BloomFilter::invert`. Bit inversion has no sound set-membership
interpretation; use the new `BloomFilter::difference` for the approximate
set-difference (A NOT B) use case it was meant to serve.
* Move `SearchCriteria` from `req` to `common` and remove its `Default`
implementation. Import `datasketches::common::SearchCriteria` and explicitly
choose `Inclusive` or `Exclusive` for each query.
@@ -23,6 +24,7 @@ All significant changes to this project will be documented in
this file.
### Bug fixes
+* `FrequentItemsSketch` updates and merges now panic before modifying the
sketch if the total stream weight would overflow, including in release builds.
Deserialization rejects non-empty images with zero stream weight or item
weights whose sum exceeds the declared stream weight.
* Fix T-Digest `merge` so it preserves `min`/`max` from the other digest
instead of re-deriving them from centroid means after compression.
* T-Digest deserialization now rejects unknown or conflicting flags, reversed
extrema, out-of-range values, unsorted centroids, and non-empty images without
stored values.
diff --git a/datasketches/src/frequencies/serialization.rs
b/datasketches/src/frequencies/serialization.rs
index d5a8b61..12def20 100644
--- a/datasketches/src/frequencies/serialization.rs
+++ b/datasketches/src/frequencies/serialization.rs
@@ -25,9 +25,9 @@ use crate::error::Error;
/// Serialization version.
pub const SERIAL_VERSION: u8 = 1;
-/// Preamble longs for empty sketch.
+/// Preamble longs for a sketch with zero stream weight.
pub const PREAMBLE_LONGS_EMPTY: u8 = 1;
-/// Preamble longs for non-empty sketch.
+/// Preamble longs for a sketch with positive stream weight, even if it
retains no items.
pub const PREAMBLE_LONGS_NONEMPTY: u8 = 4;
/// Empty flag mask (both bits for compatibility).
diff --git a/datasketches/src/frequencies/sketch.rs
b/datasketches/src/frequencies/sketch.rs
index 2b163f2..1eed1a7 100644
--- a/datasketches/src/frequencies/sketch.rs
+++ b/datasketches/src/frequencies/sketch.rs
@@ -178,17 +178,13 @@ impl<T: Eq + Hash> FrequentItemsSketch<T> {
Ok(Self::with_lg_map_sizes(lg_max_map_size, LG_MIN_MAP_SIZE))
}
- /// Returns `true` if the sketch has no active items.
+ /// Returns `true` if the total stream weight is zero.
///
/// A purge can remove all active items while retaining a non-zero total
weight and
- /// maximum error. Use [`Self::total_weight`] to distinguish that state
from a newly created
- /// or reset sketch.
+ /// maximum error. Such a sketch is not empty. Use
[`Self::num_active_items`] to check
+ /// whether any items are retained.
pub fn is_empty(&self) -> bool {
- self.hash_map.num_active() == 0
- }
-
- fn is_initial_state(&self) -> bool {
- self.stream_weight == 0 && self.offset == 0 &&
self.hash_map.num_active() == 0
+ self.stream_weight == 0
}
/// Returns the number of active items being tracked.
@@ -335,6 +331,10 @@ impl<T: Eq + Hash> FrequentItemsSketch<T> {
/// Updates the sketch with a count of one.
///
+ /// # Panics
+ ///
+ /// Panics without modifying the sketch if the total stream weight would
exceed `u64::MAX`.
+ ///
/// # Examples
///
/// ```
@@ -352,6 +352,10 @@ impl<T: Eq + Hash> FrequentItemsSketch<T> {
///
/// A count of zero is a no-op.
///
+ /// # Panics
+ ///
+ /// Panics without modifying the sketch if the total stream weight would
exceed `u64::MAX`.
+ ///
/// # Examples
///
/// ```
@@ -365,7 +369,10 @@ impl<T: Eq + Hash> FrequentItemsSketch<T> {
if count == 0 {
return;
}
- self.stream_weight += count;
+ self.stream_weight = self
+ .stream_weight
+ .checked_add(count)
+ .expect("total stream weight overflow");
self.hash_map.adjust_or_put_value(item, count);
self.maybe_resize_or_purge();
}
@@ -376,6 +383,10 @@ impl<T: Eq + Hash> FrequentItemsSketch<T> {
/// only allocates an owned item when it is newly inserted, so updating an
/// already-tracked item is allocation free.
///
+ /// # Panics
+ ///
+ /// Panics without modifying the sketch if the total stream weight would
exceed `u64::MAX`.
+ ///
/// # Examples
///
/// ```
@@ -400,6 +411,10 @@ impl<T: Eq + Hash> FrequentItemsSketch<T> {
/// item by reference and only allocates an owned item when it is newly
/// inserted. A count of zero is a no-op.
///
+ /// # Panics
+ ///
+ /// Panics without modifying the sketch if the total stream weight would
exceed `u64::MAX`.
+ ///
/// # Examples
///
/// ```
@@ -417,7 +432,10 @@ impl<T: Eq + Hash> FrequentItemsSketch<T> {
if count == 0 {
return;
}
- self.stream_weight += count;
+ self.stream_weight = self
+ .stream_weight
+ .checked_add(count)
+ .expect("total stream weight overflow");
self.hash_map.adjust_or_put_value_ref(item, count);
self.maybe_resize_or_purge();
}
@@ -427,6 +445,10 @@ impl<T: Eq + Hash> FrequentItemsSketch<T> {
/// The other sketch may have a different map size. The merged sketch
respects the
/// larger error tolerance of the inputs.
///
+ /// # Panics
+ ///
+ /// Panics without modifying the sketch if the combined stream weight
would exceed `u64::MAX`.
+ ///
/// # Examples
///
/// ```
@@ -443,10 +465,13 @@ impl<T: Eq + Hash> FrequentItemsSketch<T> {
where
T: Clone,
{
- if other.is_initial_state() {
+ if other.is_empty() {
return;
}
- let merged_total = self.stream_weight + other.stream_weight;
+ let merged_total = self
+ .stream_weight
+ .checked_add(other.stream_weight)
+ .expect("total stream weight overflow");
for (item, count) in other.hash_map.iter() {
self.update_with_count_ref(item, count);
}
@@ -576,7 +601,7 @@ impl<T: Eq + Hash> FrequentItemsSketch<T> {
count_serialize_size: CountSerializeSize<T>,
serialize_item: SerializeItem<T>,
) -> Vec<u8> {
- if self.is_initial_state() {
+ if self.is_empty() {
let mut bytes = SketchBytes::with_capacity(PREAMBLE_LONGS_EMPTY as
usize * 8);
bytes.write_u8(PREAMBLE_LONGS_EMPTY);
bytes.write_u8(SERIAL_VERSION);
@@ -649,15 +674,17 @@ impl<T: Eq + Hash> FrequentItemsSketch<T> {
ensure_serial_version_is(SERIAL_VERSION, serial_version)?;
validate_lg_map_sizes(lg_max, lg_cur)?;
- let is_empty = (flags & EMPTY_FLAG_MASK) != 0;
+ ensure_preamble_longs_in(&[PREAMBLE_LONGS_EMPTY,
PREAMBLE_LONGS_NONEMPTY], pre_longs)?;
+ let is_empty = pre_longs == PREAMBLE_LONGS_EMPTY;
+ if ((flags & EMPTY_FLAG_MASK) != 0) != is_empty {
+ return Err(Error::deserial("empty flag does not match preamble
longs"));
+ }
if is_empty {
- ensure_preamble_longs_in(&[PREAMBLE_LONGS_EMPTY], pre_longs)?;
// Java also restores empty images at the minimum size. `lg_cur`
does
// not carry item state here and must not control an eager
allocation.
return Ok(Self::with_lg_map_sizes(lg_max, LG_MIN_MAP_SIZE));
}
- ensure_preamble_longs_in(&[PREAMBLE_LONGS_NONEMPTY], pre_longs)?;
let active_items = cursor
.read_u32_le()
.map_err(insufficient_data("active_items"))?;
@@ -675,6 +702,11 @@ impl<T: Eq + Hash> FrequentItemsSketch<T> {
let stream_weight = cursor
.read_u64_le()
.map_err(insufficient_data("stream_weight"))?;
+ if stream_weight == 0 {
+ return Err(Error::deserial(
+ "non-empty sketch must have a positive stream weight",
+ ));
+ }
let offset_val =
cursor.read_u64_le().map_err(insufficient_data("offset"))?;
// Each active item has an eight-byte weight before its encoded key.
Check
@@ -691,10 +723,16 @@ impl<T: Eq + Hash> FrequentItemsSketch<T> {
}
let mut values = Vec::with_capacity(active_items);
+ let mut remaining_weight = stream_weight;
for i in 0..active_items {
- values.push(cursor.read_u64_le().map_err(|error| {
+ let value = cursor.read_u64_le().map_err(|error| {
Error::insufficient_data_of("frequent item weight",
error).with_context("index", i)
- })?);
+ })?;
+ // Reconstructing the map must not overflow or exceed the declared
stream weight.
+ remaining_weight = remaining_weight
+ .checked_sub(value)
+ .ok_or_else(|| Error::deserial("item weights exceed total
stream weight"))?;
+ values.push(value);
}
let items = deserialize_items(cursor, active_items)?;
diff --git a/tests-integration/tests/frequencies_test/update.rs
b/tests-integration/tests/frequencies_test/update.rs
index 9c1ed00..06489d3 100644
--- a/tests-integration/tests/frequencies_test/update.rs
+++ b/tests-integration/tests/frequencies_test/update.rs
@@ -15,6 +15,10 @@
// specific language governing permissions and limitations
// under the License.
+use std::hash::Hash;
+use std::panic::AssertUnwindSafe;
+use std::panic::catch_unwind;
+
use datasketches::error::ErrorKind;
use datasketches::frequencies::ErrorType;
use datasketches::frequencies::FrequentItemsSketch;
@@ -506,24 +510,80 @@ fn test_items_merge_empty_is_noop() {
assert_eq!(sketch.estimate(&1), 1);
}
-#[test]
-fn test_merge_preserves_purged_empty_state() {
- let mut purged: FrequentItemsSketch<i64> =
FrequentItemsSketch::new(32).unwrap();
- for item in 0..=(32 * 3 / 4) {
- purged.update(item);
+fn check_purged_state<T: Eq + Hash + Clone>(make_item: impl Fn(i64) -> T) {
+ let mut purged = FrequentItemsSketch::new(256).unwrap();
+ for item in 0..193 {
+ purged.update(make_item(item));
}
- assert!(purged.is_empty());
- assert_eq!(purged.total_weight(), 25);
+ assert!(!purged.is_empty());
+ assert_eq!(purged.num_active_items(), 0);
+ assert_eq!(purged.total_weight(), 193);
assert_eq!(purged.maximum_error(), 1);
- let mut merged: FrequentItemsSketch<i64> =
FrequentItemsSketch::new(32).unwrap();
+ let mut merged = FrequentItemsSketch::new(256).unwrap();
merged.merge(&purged);
- assert!(merged.is_empty());
+ assert!(!merged.is_empty());
assert_eq!(merged.num_active_items(), 0);
assert_eq!(merged.total_weight(), purged.total_weight());
assert_eq!(merged.maximum_error(), purged.maximum_error());
- assert_eq!(merged.upper_bound(&1000), purged.upper_bound(&1000));
+ assert_eq!(merged.upper_bound(&make_item(0)), 1);
+
+ let mut nonempty = FrequentItemsSketch::new(256).unwrap();
+ nonempty.update(make_item(1000));
+ nonempty.merge(&purged);
+ assert_eq!(nonempty.total_weight(), 194);
+ assert_eq!(nonempty.maximum_error(), 1);
+ assert_eq!(nonempty.upper_bound(&make_item(0)), 1);
+
+ purged.reset();
+ assert!(purged.is_empty());
+ assert_eq!(purged.total_weight(), 0);
+ assert_eq!(purged.maximum_error(), 0);
+}
+
+#[test]
+fn test_longs_purged_state() {
+ check_purged_state(|item| item);
+}
+
+#[test]
+fn test_items_purged_state() {
+ check_purged_state(|item| item.to_string());
+}
+
+#[test]
+fn test_update_weight_overflow_preserves_state() {
+ let mut sketch = FrequentItemsSketch::<String>::new(8).unwrap();
+ sketch.update_with_count_ref("a", u64::MAX - 1);
+ sketch.update("b".to_string());
+ assert_eq!(sketch.total_weight(), u64::MAX);
+ let before = sketch.serialize();
+
+ assert!(catch_unwind(AssertUnwindSafe(||
sketch.update("c".to_string()))).is_err());
+ assert_eq!(sketch.serialize(), before);
+ assert!(catch_unwind(AssertUnwindSafe(||
sketch.update_ref("a"))).is_err());
+ assert_eq!(sketch.serialize(), before);
+ assert!(!sketch.is_empty());
+}
+
+#[test]
+fn test_merge_weight_overflow_preserves_state() {
+ let mut sketch = FrequentItemsSketch::<i64>::new(8).unwrap();
+ sketch.update_with_count(1, u64::MAX - 7);
+ let mut purged = FrequentItemsSketch::<i64>::new(8).unwrap();
+ for item in 0..7 {
+ purged.update(item);
+ }
+ assert_eq!(purged.num_active_items(), 0);
+
+ sketch.merge(&purged);
+ assert_eq!(sketch.total_weight(), u64::MAX);
+ assert_eq!(sketch.maximum_error(), 1);
+ let before = sketch.serialize();
+ assert!(catch_unwind(AssertUnwindSafe(|| sketch.merge(&purged))).is_err());
+ assert_eq!(sketch.serialize(), before);
+ assert!(!sketch.is_empty());
}
#[test]
diff --git a/tests-integration/tests/serde_tests/frequencies.rs
b/tests-integration/tests/serde_tests/frequencies.rs
index a1c8f22..21f3228 100644
--- a/tests-integration/tests/serde_tests/frequencies.rs
+++ b/tests-integration/tests/serde_tests/frequencies.rs
@@ -117,69 +117,105 @@ fn test_empty_round_trip() {
assert_eq!(restored.maximum_error(), 0);
}
-#[test]
-fn test_purged_to_empty_round_trip() {
+fn check_purged_round_trip<T: FrequentItemValue>(make_item: impl Fn(i64) -> T)
{
// Saturating the map with count-1 items makes the purge median 1, which
// removes every counter while retaining stream and error state.
- let mut sketch = FrequentItemsSketch::<i64>::new(32).unwrap();
- for i in 0..=(32 * 3 / 4) {
- sketch.update(i);
+ let mut sketch = FrequentItemsSketch::<T>::new(256).unwrap();
+ for i in 0..193 {
+ sketch.update(make_item(i));
}
- assert!(sketch.is_empty());
+ assert!(!sketch.is_empty());
assert_eq!(sketch.num_active_items(), 0);
- assert_eq!(sketch.total_weight(), 25);
+ assert_eq!(sketch.total_weight(), 193);
assert_eq!(sketch.maximum_error(), 1);
- assert_eq!(sketch.upper_bound(&1000), 1);
+ assert_eq!(sketch.upper_bound(&make_item(0)), 1);
let bytes = sketch.serialize();
assert_eq!(bytes.len(), 4 * size_of::<u64>());
- let restored = FrequentItemsSketch::<i64>::deserialize(&bytes).unwrap();
- assert!(restored.is_empty());
+ assert_eq!(bytes[0], 4);
+ assert_eq!(bytes[5], 0);
+ let mut restored = FrequentItemsSketch::<T>::deserialize(&bytes).unwrap();
+ assert!(!restored.is_empty());
assert_eq!(restored.num_active_items(), 0);
assert_eq!(restored.total_weight(), sketch.total_weight());
assert_eq!(restored.maximum_error(), sketch.maximum_error());
- assert_eq!(restored.upper_bound(&1000), sketch.upper_bound(&1000));
+ assert_eq!(restored.upper_bound(&make_item(0)), 1);
assert_eq!(restored.serialize(), bytes);
+
+ restored.reset();
+ assert!(restored.is_empty());
+ assert_eq!(restored.total_weight(), 0);
+ assert_eq!(restored.maximum_error(), 0);
+ assert_eq!(restored.serialize().len(), 8);
}
#[test]
-fn test_zero_stream_weight_does_not_discard_other_state() {
- // Simulate a wrapped stream weight or an inconsistent but accepted
serialized image.
- const STREAM_WEIGHT_OFFSET: usize = 2 * size_of::<u64>();
+fn test_longs_purged_round_trip() {
+ check_purged_round_trip(|item| item);
+}
+#[test]
+fn test_items_purged_round_trip() {
+ check_purged_round_trip(|item| item.to_string());
+}
+
+#[test]
+fn test_deserialize_rejects_zero_stream_weight() {
let mut active_sketch = FrequentItemsSketch::<i64>::new(32).unwrap();
active_sketch.update_with_count(7, 3);
- let mut active_bytes = active_sketch.serialize();
- active_bytes[STREAM_WEIGHT_OFFSET..STREAM_WEIGHT_OFFSET +
size_of::<u64>()].fill(0);
-
- let active_restored =
FrequentItemsSketch::<i64>::deserialize(&active_bytes).unwrap();
- assert_eq!(active_restored.total_weight(), 0);
- assert_eq!(active_restored.num_active_items(), 1);
- assert_eq!(active_restored.estimate(&7), 3);
- assert_eq!(active_restored.serialize(), active_bytes);
-
- let mut active_merged = FrequentItemsSketch::<i64>::new(32).unwrap();
- active_merged.merge(&active_restored);
- assert_eq!(active_merged.num_active_items(), 1);
- assert_eq!(active_merged.estimate(&7), 3);
-
let mut purged_sketch = FrequentItemsSketch::<i64>::new(32).unwrap();
for item in 0..=(32 * 3 / 4) {
purged_sketch.update(item);
}
- let mut purged_bytes = purged_sketch.serialize();
- purged_bytes[STREAM_WEIGHT_OFFSET..STREAM_WEIGHT_OFFSET +
size_of::<u64>()].fill(0);
-
- let purged_restored =
FrequentItemsSketch::<i64>::deserialize(&purged_bytes).unwrap();
- assert_eq!(purged_restored.total_weight(), 0);
- assert_eq!(purged_restored.num_active_items(), 0);
- assert_eq!(purged_restored.maximum_error(), 1);
- assert_eq!(purged_restored.serialize(), purged_bytes);
-
- let mut purged_merged = FrequentItemsSketch::<i64>::new(32).unwrap();
- purged_merged.merge(&purged_restored);
- assert_eq!(purged_merged.num_active_items(), 0);
- assert_eq!(purged_merged.maximum_error(), 1);
+ for sketch in [active_sketch, purged_sketch] {
+ let mut bytes = sketch.serialize();
+ bytes[16..24].fill(0);
+ let error =
FrequentItemsSketch::<i64>::deserialize(&bytes).unwrap_err();
+ assert_eq!(error.kind(), ErrorKind::InvalidData);
+ }
+}
+
+#[test]
+fn test_deserialize_rejects_counters_exceeding_stream_weight() {
+ let mut sketch = FrequentItemsSketch::<i64>::new(32).unwrap();
+ sketch.update(1);
+ sketch.update(2);
+ let bytes = sketch.serialize();
+ for stream_weight in [1, u64::MAX] {
+ let mut corrupt = bytes.clone();
+ corrupt[16..24].copy_from_slice(&stream_weight.to_le_bytes());
+ // The second case would overflow while reconstructing the counters.
+ corrupt[32..40].copy_from_slice(&stream_weight.to_le_bytes());
+ let error =
FrequentItemsSketch::<i64>::deserialize(&corrupt).unwrap_err();
+ assert_eq!(error.kind(), ErrorKind::InvalidData);
+ }
+}
+
+#[test]
+fn test_deserialize_empty_legacy_flags() {
+ let mut bytes = FrequentItemsSketch::<i64>::new(256).unwrap().serialize();
+ for flag in [1, 4, 5] {
+ bytes[5] = flag;
+ assert!(
+ FrequentItemsSketch::<i64>::deserialize(&bytes)
+ .unwrap()
+ .is_empty()
+ );
+ }
+}
+
+#[test]
+fn test_deserialize_rejects_inconsistent_empty_flag() {
+ let mut sketch = FrequentItemsSketch::<i64>::new(256).unwrap();
+ let mut empty = sketch.serialize();
+ empty[5] = 0;
+ sketch.update(1);
+ let mut nonempty = sketch.serialize();
+ nonempty[5] = 5;
+ for bytes in [empty, nonempty] {
+ let error =
FrequentItemsSketch::<i64>::deserialize(&bytes).unwrap_err();
+ assert_eq!(error.kind(), ErrorKind::InvalidData);
+ }
}
#[test]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
