fix(precompute): count frequency sketch events

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::<CountMinSketchAccumulator>()
+            .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::<CountSketchAccumulator>()
+            .expect("must emit CountSketchAccumulator");
+        assert_eq!(count_sketch.query_key(&key), 5.0);
+    }
+
     #[test]
     fn test_reset_clears_state() {
         let mut updater = SumAccumulatorUpdater::new();