fix(query-engine): run top-k CMSHeap on FastPath

asap-query-engine/src/engines/simple_engine/mod.rs

@@ -5318,7 +5318,7 @@ mod stage_e4_instant_wrapper_equivalence_tests {
         ));
         let mut sketch = CountMinSketchWithHeapAccumulator::new(3, 1024, 32);
         for (host, value) in [("host-a", 100.0), ("host-b", 50.0), ("host-c", 10.0)] {
-            sketch.inner.update(host, value);
+            sketch.update(host, value as i32);
         }
         let output = PrecomputedOutput::new(2000, 3000, None, 1);
         store

asap-query-engine/src/engines/simple_engine/promql.rs

@@ -1776,7 +1776,7 @@ mod topk_pipeline_tests {
         let mut sketch = CountMinSketchWithHeapAccumulator::new(3, 1024, 32);
         for i in 1..=15u64 {
             let srcip = format!("10.0.0.{i}");
-            sketch.inner.update(&srcip, (i * 10) as f64);
+            sketch.update(&srcip, (i * 10) as i32);
         }
 
         let output =
@@ -1894,7 +1894,7 @@ mod topk_pipeline_tests {
         let window = &context.store_plan.values_query;
         let mut sketch = CountMinSketchWithHeapAccumulator::new(3, 1024, 32);
         for i in 1..=15u64 {
-            sketch.inner.update(&format!("10.0.0.{i}"), (i * 10) as f64);
+            sketch.update(&format!("10.0.0.{i}"), (i * 10) as i32);
         }
         store
             .insert_precomputed_output(
@@ -1981,7 +1981,7 @@ mod topk_pipeline_tests {
         let mut sketch = CountMinSketchWithHeapAccumulator::new(3, 1024, 32);
         for i in 1..=15u64 {
             let srcip = format!("10.0.0.{i}");
-            sketch.inner.update(&srcip, (i * 10) as f64);
+            sketch.update(&srcip, (i * 10) as i32);
         }
 
         let output =
@@ -2155,7 +2155,7 @@ mod topk_pipeline_tests {
         let mut sketch = CountMinSketchWithHeapAccumulator::new(3, 1024, 32);
         for i in 1..=15u64 {
             let srcip = format!("10.0.0.{i}");
-            sketch.inner.update(&srcip, (i * 10) as f64);
+            sketch.update(&srcip, (i * 10) as i32);
         }
 
         let output =
@@ -2213,7 +2213,7 @@ mod topk_pipeline_tests {
         let mut sketch = CountMinSketchWithHeapAccumulator::new(3, 1024, 32);
         for i in 1..=15u64 {
             let srcip = format!("10.0.0.{i}");
-            sketch.inner.update(&srcip, (i * 10) as f64);
+            sketch.update(&srcip, (i * 10) as i32);
         }
         let output =
             PrecomputedOutput::new(window.start_timestamp, window.end_timestamp, None, AGG_ID);

asap-query-engine/src/engines/simple_engine/sql.rs

@@ -1477,7 +1477,7 @@ mod topk_pipeline_tests {
         let mut sketch = CountMinSketchWithHeapAccumulator::new(3, 1024, 32);
         for i in 1..=15u64 {
             let srcip = format!("10.0.0.{i}");
-            sketch.inner.update(&srcip, (i * 10) as f64);
+            sketch.update(&srcip, (i * 10) as i32);
         }
 
         let output =
@@ -1547,9 +1547,9 @@ mod topk_pipeline_tests {
         // heap keys join in that order regardless of query_output_labels'
         // own order.
         let mut sketch = CountMinSketchWithHeapAccumulator::new(3, 1024, 32);
-        sketch.inner.update("10.0.0.1;shared-src", 100.0);
-        sketch.inner.update("10.0.0.2;shared-src", 200.0);
-        sketch.inner.update("10.0.0.3;other-src", 50.0);
+        sketch.update("10.0.0.1;shared-src", 100);
+        sketch.update("10.0.0.2;shared-src", 200);
+        sketch.update("10.0.0.3;other-src", 50);
 
         let output =
             PrecomputedOutput::new(window.start_timestamp, window.end_timestamp, None, AGG_ID);

asap-query-engine/src/precompute_engine/accumulator_factory.rs

@@ -758,8 +758,12 @@ impl AccumulatorUpdater for CmsWithHeapAccumulatorUpdater {
     }
 
     fn update_keyed(&mut self, key: &KeyByLabelValues, value: f64, _timestamp_ms: i64) {
-        let weight = if self.count_events { 1.0 } else { value };
-        self.acc.inner.update(&key.to_semicolon_str(), weight);
+        let count = if self.count_events {
+            1
+        } else {
+            value.round() as i32
+        };
+        self.acc.update(&key.to_semicolon_str(), count);
     }
 
     impl_accumulator_methods!(acc);
@@ -775,7 +779,7 @@ impl AccumulatorUpdater for CmsWithHeapAccumulatorUpdater {
 
     fn memory_usage_bytes(&self) -> usize {
         std::mem::size_of::<CountMinSketchWithHeapAccumulator>()
-            + self.row_num * self.col_num * std::mem::size_of::<f64>()
+            + self.row_num * self.col_num * std::mem::size_of::<i32>()
     }
 }
 
@@ -1928,14 +1932,24 @@ mod tests {
 
         let key = KeyByLabelValues::new_with_labels(vec!["10.0.0.1".to_string()]);
         for _ in 0..5 {
-            updater.update_keyed(&key, 10.0, 0);
+            updater.update_keyed(&key, 1.6, 0);
         }
         let acc = updater.take_accumulator();
         let cms = acc
             .as_any()
             .downcast_ref::<CountMinSketchWithHeapAccumulator>()
             .unwrap();
-        assert_eq!(cms.query_key(&key), 50.0, "sum of 5×10 == 50");
+        assert_eq!(cms.query_key(&key), 10.0, "sum of 5×round(1.6) == 10");
+    }
+
+    #[test]
+    fn test_cms_with_heap_memory_uses_i32_counters() {
+        let updater = CmsWithHeapAccumulatorUpdater::new(3, 7, 32, true);
+
+        assert_eq!(
+            updater.memory_usage_bytes(),
+            std::mem::size_of::<CountMinSketchWithHeapAccumulator>() + 3 * 7 * 4
+        );
     }
 
     #[test]