diff --git a/data_plane/src/precompute_engine/accumulator_factory.rs b/data_plane/src/precompute_engine/accumulator_factory.rs index 0885caa2..2e96eec8 100644 --- a/data_plane/src/precompute_engine/accumulator_factory.rs +++ b/data_plane/src/precompute_engine/accumulator_factory.rs @@ -532,9 +532,12 @@ 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. +/// 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, @@ -559,8 +562,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 +689,11 @@ 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. +/// +/// 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, @@ -713,8 +718,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); @@ -1174,37 +1179,37 @@ mod tests { } #[test] - fn bare_cms_counts_events_instead_of_summing_sample_values() { + fn bare_cms_adds_sample_values() { let mut updater = CmsAccumulatorUpdater::new(4, 256); - let key = KeyByLabelValues::new_with_labels(vec!["checkout".to_string()]); + let key = KeyByLabelValues::new_with_labels(vec!["api".to_string()]); - for value in [1_000.0, -7.0, 42.5, 0.0, 9_999.0] { - updater.update_keyed(&key, value, 0); - } + updater.update_keyed(&key, 2.0, 1000); + updater.update_keyed(&key, 3.0, 2000); + updater.update_keyed(&key, 5.0, 3000); - let acc = updater.take_accumulator(); + let acc = updater.snapshot_accumulator(); let cms = acc .as_any() .downcast_ref::() - .expect("must emit CountMinSketchAccumulator"); - assert_eq!(cms.query_key(&key), 5.0); + .expect("should be a CountMinSketchAccumulator"); + assert_eq!(cms.query_key(&key), 10.0); } #[test] - fn bare_count_sketch_counts_events_instead_of_summing_sample_values() { + fn bare_count_sketch_adds_sample_values() { let mut updater = CountSketchAccumulatorUpdater::new(5, 256); - let key = KeyByLabelValues::new_with_labels(vec!["checkout".to_string()]); + let key = KeyByLabelValues::new_with_labels(vec!["api".to_string()]); - for value in [1_000.0, -7.0, 42.5, 0.0, 9_999.0] { - updater.update_keyed(&key, value, 0); - } + updater.update_keyed(&key, 2.0, 1000); + updater.update_keyed(&key, 3.0, 2000); + updater.update_keyed(&key, 5.0, 3000); - let acc = updater.take_accumulator(); + let acc = updater.snapshot_accumulator(); let count_sketch = acc .as_any() .downcast_ref::() - .expect("must emit CountSketchAccumulator"); - assert_eq!(count_sketch.query_key(&key), 5.0); + .expect("should be a CountSketchAccumulator"); + assert_eq!(count_sketch.query_key(&key), 10.0); } #[test]