From 18ee0e93d9f701730e89888b67d6c5e13ac01387 Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 3 Sep 2026 11:02:54 -0600 Subject: [PATCH] fix(precompute): count frequency sketch events --- .../precompute_engine/accumulator_factory.rs | 49 +++++++++++++++++-- 1 file changed, 44 insertions(+), 5 deletions(-) diff --git a/data_plane/src/precompute_engine/accumulator_factory.rs b/data_plane/src/precompute_engine/accumulator_factory.rs index 9da5bed5..0885caa2 100644 --- a/data_plane/src/precompute_engine/accumulator_factory.rs +++ b/data_plane/src/precompute_engine/accumulator_factory.rs @@ -532,6 +532,9 @@ impl AccumulatorUpdater for MultipleIncreaseAccumulatorUpdater { // CmsAccumulatorUpdater (CountMinSketch) // --------------------------------------------------------------------------- +/// Bare CMS realizes canonical `SketchQuery::PointCount` frequency semantics. +/// Each observation contributes one occurrence regardless of its scalar sample +/// value. Value-weighted top-k is a separate `CmsHeapAccumulatorUpdater` mode. pub struct CmsAccumulatorUpdater { acc: CountMinSketchAccumulator, row_num: usize, @@ -556,8 +559,8 @@ impl AccumulatorUpdater for CmsAccumulatorUpdater { ); } - fn update_keyed(&mut self, key: &KeyByLabelValues, value: f64, _timestamp_ms: i64) { - self.acc.inner.update(&key.to_semicolon_str(), value); + fn update_keyed(&mut self, key: &KeyByLabelValues, _value: f64, _timestamp_ms: i64) { + self.acc.inner.update(&key.to_semicolon_str(), 1.0); } impl_clone_accumulator_methods!(acc); @@ -683,7 +686,9 @@ impl AccumulatorUpdater for CmsHeapAccumulatorUpdater { /// (signed rows, median-of-rows estimator) — distinct math from /// `CmsAccumulatorUpdater`'s CMS (min-of-rows). Closes, on the raw-metric /// ingest path, the conflation bug where `SummaryKind::CountSketch` silently -/// shared `CmsAccumulatorUpdater` with bare CMS. +/// shared `CmsAccumulatorUpdater` with bare CMS. Like bare CMS, each +/// observation contributes one occurrence; value-weighted top-k uses the +/// separate heap updater. pub struct CountSketchAccumulatorUpdater { acc: CountSketchAccumulator, row_num: usize, @@ -708,8 +713,8 @@ impl AccumulatorUpdater for CountSketchAccumulatorUpdater { ); } - fn update_keyed(&mut self, key: &KeyByLabelValues, value: f64, _timestamp_ms: i64) { - self.acc.inner.update(&key.to_semicolon_str(), value); + fn update_keyed(&mut self, key: &KeyByLabelValues, _value: f64, _timestamp_ms: i64) { + self.acc.inner.update(&key.to_semicolon_str(), 1.0); } impl_clone_accumulator_methods!(acc); @@ -1168,6 +1173,40 @@ mod tests { assert_eq!(acc.type_name(), "MultipleSumAccumulator"); } + #[test] + fn bare_cms_counts_events_instead_of_summing_sample_values() { + let mut updater = CmsAccumulatorUpdater::new(4, 256); + let key = KeyByLabelValues::new_with_labels(vec!["checkout".to_string()]); + + for value in [1_000.0, -7.0, 42.5, 0.0, 9_999.0] { + updater.update_keyed(&key, value, 0); + } + + let acc = updater.take_accumulator(); + let cms = acc + .as_any() + .downcast_ref::() + .expect("must emit CountMinSketchAccumulator"); + assert_eq!(cms.query_key(&key), 5.0); + } + + #[test] + fn bare_count_sketch_counts_events_instead_of_summing_sample_values() { + let mut updater = CountSketchAccumulatorUpdater::new(5, 256); + let key = KeyByLabelValues::new_with_labels(vec!["checkout".to_string()]); + + for value in [1_000.0, -7.0, 42.5, 0.0, 9_999.0] { + updater.update_keyed(&key, value, 0); + } + + let acc = updater.take_accumulator(); + let count_sketch = acc + .as_any() + .downcast_ref::() + .expect("must emit CountSketchAccumulator"); + assert_eq!(count_sketch.query_key(&key), 5.0); + } + #[test] fn test_reset_clears_state() { let mut updater = SumAccumulatorUpdater::new();