From c7f44aca64da07f7df8d6747e9770cdc3515b8ae Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 3 Sep 2026 11:18:23 -0600 Subject: [PATCH 1/2] Revert "Merge pull request #481 from ProjectASAP/fix/frequency-sketch-unit-weight" This reverts commit d4e19882a066c85b77fd23b0da0da21cdef51471, reversing changes made to a05e5fb4b509977abd068e9b5b29dc7beb743d0d. --- .../precompute_engine/accumulator_factory.rs | 49 ++----------------- 1 file changed, 5 insertions(+), 44 deletions(-) diff --git a/data_plane/src/precompute_engine/accumulator_factory.rs b/data_plane/src/precompute_engine/accumulator_factory.rs index 0885caa2..9da5bed5 100644 --- a/data_plane/src/precompute_engine/accumulator_factory.rs +++ b/data_plane/src/precompute_engine/accumulator_factory.rs @@ -532,9 +532,6 @@ 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, @@ -559,8 +556,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(), 1.0); + fn update_keyed(&mut self, key: &KeyByLabelValues, value: f64, _timestamp_ms: i64) { + self.acc.inner.update(&key.to_semicolon_str(), value); } impl_clone_accumulator_methods!(acc); @@ -686,9 +683,7 @@ 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. Like bare CMS, each -/// observation contributes one occurrence; value-weighted top-k uses the -/// separate heap updater. +/// shared `CmsAccumulatorUpdater` with bare CMS. pub struct CountSketchAccumulatorUpdater { acc: CountSketchAccumulator, row_num: usize, @@ -713,8 +708,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(), 1.0); + fn update_keyed(&mut self, key: &KeyByLabelValues, value: f64, _timestamp_ms: i64) { + self.acc.inner.update(&key.to_semicolon_str(), value); } impl_clone_accumulator_methods!(acc); @@ -1173,40 +1168,6 @@ 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(); From ac997f1c11fe2c181c135dffbbf074403fdaf2a1 Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 3 Sep 2026 11:20:02 -0600 Subject: [PATCH 2/2] test(precompute): lock weighted sketch semantics --- .../precompute_engine/accumulator_factory.rs | 44 +++++++++++++++++++ 1 file changed, 44 insertions(+) diff --git a/data_plane/src/precompute_engine/accumulator_factory.rs b/data_plane/src/precompute_engine/accumulator_factory.rs index 9da5bed5..2e96eec8 100644 --- a/data_plane/src/precompute_engine/accumulator_factory.rs +++ b/data_plane/src/precompute_engine/accumulator_factory.rs @@ -532,6 +532,12 @@ impl AccumulatorUpdater for MultipleIncreaseAccumulatorUpdater { // CmsAccumulatorUpdater (CountMinSketch) // --------------------------------------------------------------------------- +/// Keyed weighted-frequency updater. +/// +/// A raw Prometheus sample represents the observed metric value, so a bare CMS +/// adds `value` for its key. Counting each received sample as one is a distinct +/// event-count operation and requires an explicit typed plan contract; it must +/// not be inferred from the sketch algorithm alone. pub struct CmsAccumulatorUpdater { acc: CountMinSketchAccumulator, row_num: usize, @@ -684,6 +690,10 @@ impl AccumulatorUpdater for CmsHeapAccumulatorUpdater { /// `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. +/// +/// As with bare CMS, each raw Prometheus sample contributes its `value`. +/// Unit event counting must be selected explicitly by a future typed plan +/// contract rather than being implied by `SummaryKind::CountSketch`. pub struct CountSketchAccumulatorUpdater { acc: CountSketchAccumulator, row_num: usize, @@ -1168,6 +1178,40 @@ mod tests { assert_eq!(acc.type_name(), "MultipleSumAccumulator"); } + #[test] + fn bare_cms_adds_sample_values() { + let mut updater = CmsAccumulatorUpdater::new(4, 256); + let key = KeyByLabelValues::new_with_labels(vec!["api".to_string()]); + + updater.update_keyed(&key, 2.0, 1000); + updater.update_keyed(&key, 3.0, 2000); + updater.update_keyed(&key, 5.0, 3000); + + let acc = updater.snapshot_accumulator(); + let cms = acc + .as_any() + .downcast_ref::() + .expect("should be a CountMinSketchAccumulator"); + assert_eq!(cms.query_key(&key), 10.0); + } + + #[test] + fn bare_count_sketch_adds_sample_values() { + let mut updater = CountSketchAccumulatorUpdater::new(5, 256); + let key = KeyByLabelValues::new_with_labels(vec!["api".to_string()]); + + updater.update_keyed(&key, 2.0, 1000); + updater.update_keyed(&key, 3.0, 2000); + updater.update_keyed(&key, 5.0, 3000); + + let acc = updater.snapshot_accumulator(); + let count_sketch = acc + .as_any() + .downcast_ref::() + .expect("should be a CountSketchAccumulator"); + assert_eq!(count_sketch.query_key(&key), 10.0); + } + #[test] fn test_reset_clears_state() { let mut updater = SumAccumulatorUpdater::new();