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 2a19b8d fix(countmin): preserve integer precision during scaling
(#285)
2a19b8d is described below
commit 2a19b8da2b92e19554a5c7a6d8a110f107189226
Author: Jaideep Pyne <[email protected]>
AuthorDate: Sat Oct 3 19:48:08 2026 +0530
fix(countmin): preserve integer precision during scaling (#285)
Co-authored-by: tison <[email protected]>
---
CHANGELOG.md | 2 +
benchmarks/countmin/{mod.rs => decay.rs} | 16 ++++-
benchmarks/countmin/mod.rs | 1 +
datasketches/src/countmin/sketch.rs | 29 +++++----
datasketches/src/countmin/value.rs | 32 +++++++--
tests-integration/tests/countmin_test/sketch.rs | 87 +++++++++++++++++++++++++
6 files changed, 149 insertions(+), 18 deletions(-)
diff --git a/CHANGELOG.md b/CHANGELOG.md
index c9739f2..ff73f71 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -20,6 +20,7 @@ All significant changes to this project will be documented in
this file.
* `KllSketch::quantile` and `ReqSketch::quantile` are faster at ranks `0.0`
and `1.0`, with query work independent of the number of retained samples.
* `BloomFilter::insert` is faster for integer and raw-byte inputs.
`BloomFilter::contains_and_insert` is also faster when checking already-present
integer values.
* `CountMinSketch` updates are faster for integer and raw-byte inputs.
+* `CountMinSketch::decay(1.0)` now takes constant time.
* `CpcSketch` updates are faster for integer and raw-byte inputs.
* `FrequentItemsSketch` updates are faster for integer and raw-byte keys.
* `HllSketch` updates are faster for integer and raw-byte inputs.
@@ -30,6 +31,7 @@ All significant changes to this project will be documented in
this file.
### Bug fixes
+* `CountMinSketch::decay` and `upper_bound` now scale integer weights without
intermediate floating-point rounding, including signed error bounds. Products
use the exact binary value of the floating-point factor and truncate toward
zero; for example, `decay(0.99)` changes a counter of `100` to `98` rather than
`99`.
* `KllSketch` and `ReqSketch` quantile queries, including their sorted views,
now return the exact stream minimum and maximum at ranks `0.0` and `1.0`, even
after compaction or when the stream weight exceeds `2^53`.
* Empty compact Theta and Tuple sketches now serialize with a zero seed hash,
matching the canonical cross-language encoding. Deserialization associates
empty sketches with the supplied seed, including legacy Theta v2 images with
zero or mismatched seed hashes.
* `CountMinSketch` updates now panic and merges return `InvalidArgument` if
the total absolute weight would exceed the counter type's maximum. Both leave
the sketch unchanged, including in release builds.
diff --git a/benchmarks/countmin/mod.rs b/benchmarks/countmin/decay.rs
similarity index 63%
copy from benchmarks/countmin/mod.rs
copy to benchmarks/countmin/decay.rs
index 1dea33b..82de772 100644
--- a/benchmarks/countmin/mod.rs
+++ b/benchmarks/countmin/decay.rs
@@ -15,4 +15,18 @@
// specific language governing permissions and limitations
// under the License.
-mod update;
+use datasketches::countmin::CountMinSketch;
+use divan::Bencher;
+use divan::black_box;
+
+#[divan::bench(args = [0.5, 0.99, 1.0])]
+fn u64(bencher: Bencher, factor: f64) {
+ let mut sketch = CountMinSketch::<u64>::new(4, 16_384).unwrap();
+ for item in 0..128_u64 {
+ sketch.update_with_weight(item, (1 << 53) + item);
+ }
+
+ bencher
+ .with_inputs(|| sketch.clone())
+ .bench_local_refs(|sketch| black_box(sketch).decay(black_box(factor)));
+}
diff --git a/benchmarks/countmin/mod.rs b/benchmarks/countmin/mod.rs
index 1dea33b..375b6d8 100644
--- a/benchmarks/countmin/mod.rs
+++ b/benchmarks/countmin/mod.rs
@@ -15,4 +15,5 @@
// specific language governing permissions and limitations
// under the License.
+mod decay;
mod update;
diff --git a/datasketches/src/countmin/sketch.rs
b/datasketches/src/countmin/sketch.rs
index bffdca1..6c02dc0 100644
--- a/datasketches/src/countmin/sketch.rs
+++ b/datasketches/src/countmin/sketch.rs
@@ -132,7 +132,7 @@ impl<T: CountMinValue> CountMinSketch<T> {
std::f64::consts::E / self.num_buckets as f64
}
- /// Returns `true` if the sketch has not seen any updates.
+ /// Returns `true` if the total weight is zero.
pub fn is_empty(&self) -> bool {
self.total_weight == T::ZERO
}
@@ -260,11 +260,9 @@ impl<T: CountMinValue> CountMinSketch<T> {
}
/// Returns the upper bound on the true frequency of the given item.
- ///
- /// Clamps the bound to `T::MAX` if adding the error would overflow.
pub fn upper_bound<I: Hash>(&self, item: I) -> T {
let estimate = self.estimate(item);
- let error = self.total_weight.scale(self.relative_error());
+ let error = self.total_weight.scale_nonnegative(self.relative_error());
estimate.checked_add(error).unwrap_or(T::MAX)
}
@@ -514,9 +512,7 @@ impl<T: CountMinValue> CountMinSketch<T> {
}
impl<T: UnsignedCountMinValue> CountMinSketch<T> {
- /// Divides every counter by two, truncating toward zero.
- ///
- /// Useful for exponential decay where counts represent recent activity.
+ /// Halves all counters and the total weight, rounding down.
///
/// # Examples
///
@@ -535,10 +531,12 @@ impl<T: UnsignedCountMinValue> CountMinSketch<T> {
self.total_weight = self.total_weight.halve();
}
- /// Multiplies every counter by `decay` and truncates back into `T`.
+ /// Scales all counters and the total weight by `decay`, rounding down.
///
- /// Values are truncated toward zero after multiplication; choose `decay`
in `(0, 1]`.
- /// The total weight is scaled by the same factor to keep bounds
consistent.
+ /// Rounding uses the actual `f64` value of the factor, so `decay(0.99)`
changes a
+ /// count of `100` to `98`.
+ ///
+ /// A factor of `1.0` leaves the sketch unchanged; `0.5` is equivalent to
[`Self::halve`].
///
/// # Panics
///
@@ -556,10 +554,17 @@ impl<T: UnsignedCountMinValue> CountMinSketch<T> {
/// ```
pub fn decay(&mut self, decay: f64) {
assert!(decay > 0.0 && decay <= 1.0, "decay must be within (0, 1]");
+ if decay == 1.0 {
+ return;
+ }
+ if decay == 0.5 {
+ self.halve();
+ return;
+ }
for c in &mut self.counts {
- *c = c.scale(decay)
+ *c = c.scale_nonnegative(decay)
}
- self.total_weight = self.total_weight.scale(decay);
+ self.total_weight = self.total_weight.scale_nonnegative(decay);
}
}
diff --git a/datasketches/src/countmin/value.rs
b/datasketches/src/countmin/value.rs
index 6e9ec93..93d78e1 100644
--- a/datasketches/src/countmin/value.rs
+++ b/datasketches/src/countmin/value.rs
@@ -39,7 +39,7 @@ mod private {
fn checked_abs(self) -> Option<Self>;
fn checked_add(self, other: Self) -> Option<Self>;
- fn scale(self, factor: f64) -> Self;
+ fn scale_nonnegative(self, factor: f64) -> Self;
fn to_bytes(self) -> [u8; 8];
fn try_from_bytes(bytes: [u8; 8]) -> Result<Self, Error>;
}
@@ -49,6 +49,27 @@ mod private {
}
}
+#[inline]
+fn scale_u64(value: u64, factor: f64) -> u64 {
+ debug_assert!(factor > 0.0 && factor <= 1.0);
+
+ // Decode factor = significand / 2^shift. The product needs at most
+ // 64 + 53 bits, so u128 preserves it exactly until the final truncation.
+ let bits = factor.to_bits();
+ let exponent = ((bits >> 52) & 0x7ff) as u32;
+ let fraction = bits & ((1_u64 << 52) - 1);
+ // Normal values have an implicit leading bit and exponent bias 1023;
+ // subnormals have no leading bit and use a fixed exponent of -1074.
+ let (significand, shift) = if exponent == 0 {
+ (fraction, 1074)
+ } else {
+ (fraction | (1_u64 << 52), 1075 - exponent)
+ };
+ (u128::from(value) * u128::from(significand))
+ .checked_shr(shift)
+ .unwrap_or(0) as u64
+}
+
macro_rules! impl_signed {
($name:ty, $min:expr, $max:expr) => {
impl private::CountMinValue for $name {
@@ -67,8 +88,9 @@ macro_rules! impl_signed {
}
#[inline(always)]
- fn scale(self, factor: f64) -> Self {
- ((self as f64) * factor).trunc() as $name
+ fn scale_nonnegative(self, factor: f64) -> Self {
+ debug_assert!(self >= 0);
+ scale_u64(self as u64, factor) as $name
}
#[inline(always)]
@@ -118,8 +140,8 @@ macro_rules! impl_unsigned {
}
#[inline(always)]
- fn scale(self, factor: f64) -> Self {
- ((self as f64) * factor).trunc() as $name
+ fn scale_nonnegative(self, factor: f64) -> Self {
+ scale_u64(self as u64, factor) as $name
}
#[inline(always)]
diff --git a/tests-integration/tests/countmin_test/sketch.rs
b/tests-integration/tests/countmin_test/sketch.rs
index 785b86e..ba2ab67 100644
--- a/tests-integration/tests/countmin_test/sketch.rs
+++ b/tests-integration/tests/countmin_test/sketch.rs
@@ -19,6 +19,7 @@ use std::panic::AssertUnwindSafe;
use std::panic::catch_unwind;
use datasketches::countmin::CountMinSketch;
+use datasketches::countmin::UnsignedCountMinValue;
use datasketches::error::ErrorKind;
use googletest::assert_that;
use googletest::prelude::ge;
@@ -75,6 +76,26 @@ fn upper_bound_clamps_on_overflow() {
assert_eq!(signed.upper_bound("x"), i8::MAX);
}
+#[test]
+fn upper_bound_preserves_large_integer_precision_for_both_counter_types() {
+ let mut signed = CountMinSketch::<i64>::new(3, 128).unwrap();
+ let mut unsigned = CountMinSketch::<u64>::new(3, 128).unwrap();
+ signed.update_with_weight("item", i64::MAX);
+ unsigned.update_with_weight("item", i64::MAX as u64);
+ assert_eq!(signed.estimate("missing"), 0);
+ assert_eq!(unsigned.estimate("missing"), 0);
+
+ // Multiplying this epsilon by 2^63 gives an exactly representable integer.
+ // Subtracting one from the weight subtracts epsilon, so truncation must
drop one.
+ let error = (signed.relative_error() * (1_u64 << 63) as f64) as u64 - 1;
+ assert_eq!(signed.upper_bound("missing"), error as i64);
+ assert_eq!(unsigned.upper_bound("missing"), error);
+
+ let mut negative = CountMinSketch::<i64>::new(3, 128).unwrap();
+ negative.update_with_weight("item", -i64::MAX);
+ assert_eq!(negative.upper_bound("item"), -i64::MAX + error as i64);
+}
+
#[test]
fn test_init_defaults() {
let sketch = CountMinSketch::<i64>::new(3, 5).unwrap();
@@ -261,6 +282,72 @@ fn test_decay() {
}
}
+#[test]
+fn test_decay_preserves_large_integer_precision() {
+ for weight in [(1_u64 << 53) + 1, u64::MAX - 1, u64::MAX] {
+ for (factor, expected) in [
+ (1.0, weight),
+ (0.5, weight / 2),
+ (0.75, ((u128::from(weight) * 3) / 4) as u64),
+ (1.0_f64.next_down(), weight - weight.div_ceil(1 << 53)),
+ (2.0_f64.powi(-63), weight >> 63),
+ (2.0_f64.powi(-64), 0),
+ (f64::from_bits(1), 0),
+ ] {
+ let mut sketch = CountMinSketch::<u64>::new(3, 128).unwrap();
+ sketch.update_with_weight("item", weight);
+ sketch.decay(factor);
+ assert_eq!(
+ sketch.total_weight(),
+ expected,
+ "weight={weight}, factor={factor}"
+ );
+ assert_eq!(sketch.estimate("item"), expected);
+ let restored =
CountMinSketch::<u64>::deserialize(&sketch.serialize()).unwrap();
+ assert_eq!(restored.total_weight(), expected);
+ }
+ }
+}
+
+#[test]
+fn decay_truncates_the_exact_binary_product() {
+ for (weight, factor, expected) in [
+ (100, 0.99, 98),
+ (100, 0.99_f64.next_up(), 99),
+ (10, 0.3, 2),
+ (10, 0.3_f64.next_up(), 3),
+ ] {
+ let mut sketch = CountMinSketch::<u64>::new(3, 128).unwrap();
+ sketch.update_with_weight("item", weight);
+ sketch.decay(factor);
+ assert_eq!(sketch.total_weight(), expected);
+ assert_eq!(sketch.estimate("item"), expected);
+ }
+}
+
+#[test]
+fn decay_identity_and_halving_preserve_the_whole_sketch() {
+ fn check<T: UnsignedCountMinValue + std::fmt::Debug>(weights: [T; 3]) {
+ let mut sketch = CountMinSketch::<T>::new(3, 128).unwrap();
+ for (item, weight) in weights.into_iter().enumerate() {
+ sketch.update_with_weight(item, weight);
+ }
+
+ let mut decayed = sketch.clone();
+ decayed.decay(1.0);
+ assert_eq!(decayed, sketch);
+
+ decayed.decay(0.5);
+ sketch.halve();
+ assert_eq!(decayed, sketch);
+ }
+
+ check([u8::MAX / 2, u8::MAX / 4, 1]);
+ check([u16::MAX / 2, u16::MAX / 4, 1]);
+ check([u32::MAX / 2, u32::MAX / 4, 1]);
+ check([u64::MAX / 2, u64::MAX / 4, 1]);
+}
+
#[test]
fn test_merge() {
let mut left = CountMinSketch::<i64>::new(3, 64).unwrap();
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
