fix(query-engine): MultipleSum weights samples by 1 for count sub_type (#717)

* fix(query-engine): MultipleSum weights samples by 1 for count sub_type

MultipleSumAccumulatorUpdater always summed the raw sample value,
so an ingested count precompute held Σvalue instead of Σ1 whenever
aggregation_sub_type=="count". Mirror the CountMinSketch fix: extract
a shared, strict sub_type-to-weighting helper and use it for both.

Fixes #503

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01RzTeeyqi31cRJ1BkMFWWRx

* fix(planner,query-engine): propagate MultipleSum's strict subtype contract

#503 made MultipleSum reject an empty/invalid aggregation_sub_type, but
two producers/validators of MultipleSum configs weren't updated to match
(roborev review #197):

- candidate_gen.rs derived "" for (Sum, MultipleSum) candidates instead
  of "sum", so every optimizer-selected MultipleSum candidate for a Sum
  statistic would fail to ingest.
- engine startup validation only ran create_accumulator_updater over
  CountMinSketch/CountMinSketchWithHeap configs, so an invalid MultipleSum
  subtype would pass startup and fail lazily per-batch instead of failing
  fast.

Deferred: capability_matching.rs's Sum/Count candidate lists aren't
subtype-aware for MultipleSum either, but neither the streaming planner
nor the optimizer ever generates a MultipleSum+"count" config today
(Statistic::Count routes to CountMinSketch/CountMinSketchWithHeap only),
so this is unreachable outside a hand-authored config -- same category
as MultipleSubpopulation, which #503 already scoped out.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01RzTeeyqi31cRJ1BkMFWWRx

---------

Co-authored-by: Claude Sonnet 5 <noreply@anthropic.com>

Author: milindsrivastava1997

asap-planner-rs/src/optimizer/candidate_gen.rs

@@ -239,7 +239,7 @@ fn derive_sub_type(stat: Statistic, agg_type: AggregationType) -> String {
         (Statistic::Min, _) => "min",
         (Statistic::Max, _) => "max",
         (Statistic::Topk, _) => "topk",
-        (Statistic::Sum, AggregationType::CountMinSketch) => "sum",
+        (Statistic::Sum, AggregationType::CountMinSketch | AggregationType::MultipleSum) => "sum",
         (Statistic::Count, AggregationType::CountMinSketch) => "count",
         _ => "",
     }
@@ -363,6 +363,27 @@ mod tests {
             .any(|c| c.config.is_none() && c.query_method == QueryMethod::Exact));
     }
 
+    #[test]
+    fn multiple_sum_candidates_get_a_non_empty_sub_type() {
+        // MultipleSum's factory now rejects an empty aggregation_sub_type (#503) --
+        // the optimizer must derive "sum" for it, same as it already does for
+        // CountMinSketch, or every MultipleSum candidate fails to ingest.
+        let aqe = make_aqe(Statistic::Sum, 300_000, 60_000);
+        let candidates = enumerate_candidates(&aqe, 15_000);
+        let multiple_sum_configs: Vec<_> = candidates
+            .iter()
+            .filter_map(|c| c.config.as_ref())
+            .filter(|cfg| cfg.aggregation_type == AggregationType::MultipleSum)
+            .collect();
+        assert!(
+            !multiple_sum_configs.is_empty(),
+            "expected at least one MultipleSum candidate"
+        );
+        for cfg in multiple_sum_configs {
+            assert_eq!(cfg.aggregation_sub_type, "sum");
+        }
+    }
+
     #[test]
     fn stamps_dataset_label_group_count_on_every_candidate() {
         let aqe = make_aqe(Statistic::Sum, 300_000, 60_000);

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

@@ -461,22 +461,18 @@ impl AccumulatorUpdater for DeltaSetAggregatorUpdater {
 
 pub struct MultipleSumAccumulatorUpdater {
     acc: MultipleSumAccumulator,
+    count_events: bool,
 }
 
 impl MultipleSumAccumulatorUpdater {
-    pub fn new() -> Self {
+    pub fn new(count_events: bool) -> Self {
         Self {
             acc: MultipleSumAccumulator::new(),
+            count_events,
         }
     }
 }
 
-impl Default for MultipleSumAccumulatorUpdater {
-    fn default() -> Self {
-        Self::new()
-    }
-}
-
 impl AccumulatorUpdater for MultipleSumAccumulatorUpdater {
     fn update_single(&mut self, _value: f64, _timestamp_ms: i64) {
         debug_assert!(
@@ -486,7 +482,8 @@ impl AccumulatorUpdater for MultipleSumAccumulatorUpdater {
     }
 
     fn update_keyed(&mut self, key: &KeyByLabelValues, value: f64, _timestamp_ms: i64) {
-        self.acc.update(key.clone(), value);
+        let weight = if self.count_events { 1.0 } else { value };
+        self.acc.update(key.clone(), weight);
     }
 
     impl_accumulator_methods!(acc);
@@ -845,20 +842,19 @@ fn cms_params(config: &AggregationConfig) -> Result<(usize, usize), String> {
     Ok((row_num, col_num))
 }
 
-/// Resolve the weighting semantics for a plain Count-Min Sketch.
+/// Resolve the weighting semantics for aggregation types that dispatch SUM vs
+/// COUNT via `aggregation_sub_type` (plain CountMinSketch, MultipleSum).
 ///
-/// Unlike the heap variant, plain CMS uses `aggregation_sub_type` to
-/// distinguish approximate SUM from approximate COUNT. Do not silently
-/// default malformed configs: the wrong weighting produces plausible but
-/// incorrect results.
-fn cms_count_events_for_sub_type(sub_type: &str) -> Result<bool, String> {
+/// Do not silently default malformed configs: the wrong weighting produces
+/// plausible but incorrect results.
+fn sum_or_count_events_for_sub_type(agg_type_name: &str, sub_type: &str) -> Result<bool, String> {
     if sub_type.eq_ignore_ascii_case("count") {
         Ok(true)
     } else if sub_type.eq_ignore_ascii_case("sum") {
         Ok(false)
     } else {
         Err(format!(
-            "CountMinSketch requires aggregation_sub_type 'sum' or 'count', got '{sub_type}'"
+            "{agg_type_name} requires aggregation_sub_type 'sum' or 'count', got '{sub_type}'"
         ))
     }
 }
@@ -948,7 +944,7 @@ pub fn create_accumulator_updater(
             other => Err(format!("Unknown SingleSubpopulation sub_type '{other}'")),
         },
         AggregationType::MultipleSubpopulation => match sub_type {
-            "Sum" | "sum" => Ok(Box::new(MultipleSumAccumulatorUpdater::new())),
+            "Sum" | "sum" => Ok(Box::new(MultipleSumAccumulatorUpdater::new(false))),
             "Min" | "min" => Ok(Box::new(MultipleMinMaxAccumulatorUpdater::new(false))),
             "Max" | "max" => Ok(Box::new(MultipleMinMaxAccumulatorUpdater::new(true))),
             "Increase" | "increase" => Ok(Box::new(MultipleIncreaseAccumulatorUpdater::new())),
@@ -969,7 +965,10 @@ pub fn create_accumulator_updater(
         AggregationType::DatasketchesKLL => {
             Ok(Box::new(KllAccumulatorUpdater::new(kll_k_param(config)?)))
         }
-        AggregationType::MultipleSum => Ok(Box::new(MultipleSumAccumulatorUpdater::new())),
+        AggregationType::MultipleSum => {
+            let count_events = sum_or_count_events_for_sub_type("MultipleSum", sub_type)?;
+            Ok(Box::new(MultipleSumAccumulatorUpdater::new(count_events)))
+        }
         AggregationType::MultipleIncrease => {
             Ok(Box::new(MultipleIncreaseAccumulatorUpdater::new()))
         }
@@ -983,7 +982,7 @@ pub fn create_accumulator_updater(
         AggregationType::Increase => Ok(Box::new(IncreaseAccumulatorUpdater::new())),
         AggregationType::CountMinSketch => {
             let (row_num, col_num) = cms_params(config)?;
-            let count_events = cms_count_events_for_sub_type(sub_type)?;
+            let count_events = sum_or_count_events_for_sub_type("CountMinSketch", sub_type)?;
             Ok(Box::new(CmsAccumulatorUpdater::new(
                 row_num,
                 col_num,
@@ -1017,6 +1016,7 @@ pub fn create_accumulator_updater(
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::data_model::MultipleSubpopulationAggregate;
     use asap_types::enums::{AggregationType, WindowType};
 
     #[test]
@@ -1066,7 +1066,7 @@ mod tests {
 
     #[test]
     fn test_multiple_sum_updater() {
-        let mut updater = MultipleSumAccumulatorUpdater::new();
+        let mut updater = MultipleSumAccumulatorUpdater::new(false);
         assert!(updater.is_keyed());
 
         let key_a = KeyByLabelValues::new_with_labels(vec!["a".to_string()]);
@@ -1137,7 +1137,7 @@ mod tests {
         )));
         assert!(config_is_keyed(&make_config(
             AggregationType::MultipleSum,
-            ""
+            "sum"
         )));
         assert!(config_is_keyed(&make_config(
             AggregationType::MultipleIncrease,
@@ -1157,7 +1157,7 @@ mod tests {
         for (agg_type, sub_type) in &[
             (AggregationType::SingleSubpopulation, "Sum"),
             (AggregationType::MultipleSubpopulation, "Sum"),
-            (AggregationType::MultipleSum, ""),
+            (AggregationType::MultipleSum, "sum"),
         ] {
             let config = make_config(*agg_type, sub_type);
             let updater = create_accumulator_updater(&config).unwrap();
@@ -1615,6 +1615,120 @@ mod tests {
         assert!(err.contains(" count "));
     }
 
+    fn multiple_sum_config(sub_type: &str) -> AggregationConfig {
+        AggregationConfig::new(
+            200,
+            AggregationType::MultipleSum,
+            sub_type.to_string(),
+            std::collections::HashMap::new(),
+            promql_utilities::data_model::key_by_label_names::KeyByLabelNames::new(vec![]),
+            promql_utilities::data_model::key_by_label_names::KeyByLabelNames::new(vec![]),
+            promql_utilities::data_model::key_by_label_names::KeyByLabelNames::new(vec![]),
+            String::new(),
+            1_000,
+            1_000,
+            WindowType::Tumbling,
+            "test_metric".to_string(),
+            "test_metric".to_string(),
+            None,
+            None,
+            None,
+            None,
+        )
+    }
+
+    #[test]
+    fn test_multiple_sum_count_subtype_uses_unit_weight() {
+        let config = multiple_sum_config("count");
+        let mut updater = create_accumulator_updater(&config).unwrap();
+        let key = KeyByLabelValues::new_with_labels(vec!["host-a".to_string()]);
+
+        for _ in 0..5 {
+            updater.update_keyed(&key, 1_000.0, 0);
+        }
+
+        let acc = updater.take_accumulator();
+        let multiple_sum = acc
+            .as_any()
+            .downcast_ref::<MultipleSumAccumulator>()
+            .expect("MultipleSum accumulator");
+        assert_eq!(
+            multiple_sum
+                .query(
+                    promql_utilities::query_logics::enums::Statistic::Count,
+                    &key,
+                    None
+                )
+                .unwrap(),
+            5.0
+        );
+    }
+
+    #[test]
+    fn test_multiple_sum_sum_subtype_uses_sample_weight() {
+        let config = multiple_sum_config("sum");
+        let mut updater = create_accumulator_updater(&config).unwrap();
+        let key = KeyByLabelValues::new_with_labels(vec!["host-a".to_string()]);
+
+        for _ in 0..5 {
+            updater.update_keyed(&key, 10.0, 0);
+        }
+
+        let acc = updater.take_accumulator();
+        let multiple_sum = acc
+            .as_any()
+            .downcast_ref::<MultipleSumAccumulator>()
+            .expect("MultipleSum accumulator");
+        assert_eq!(
+            multiple_sum
+                .query(
+                    promql_utilities::query_logics::enums::Statistic::Sum,
+                    &key,
+                    None
+                )
+                .unwrap(),
+            50.0
+        );
+    }
+
+    #[test]
+    fn test_multiple_sum_rejects_empty_subtype() {
+        let config = multiple_sum_config("");
+        let err = match create_accumulator_updater(&config) {
+            Ok(_) => panic!("empty MultipleSum subtype must fail"),
+            Err(err) => err,
+        };
+        assert!(err.contains("sum") && err.contains("count"));
+    }
+
+    #[test]
+    fn test_multiple_sum_rejects_unknown_subtype() {
+        let config = multiple_sum_config("frequency");
+        let err = match create_accumulator_updater(&config) {
+            Ok(_) => panic!("unknown MultipleSum subtype must fail"),
+            Err(err) => err,
+        };
+        assert!(err.contains("frequency"));
+    }
+
+    #[test]
+    fn test_multiple_sum_accepts_case_insensitive_subtype() {
+        for sub_type in ["COUNT", "SuM"] {
+            create_accumulator_updater(&multiple_sum_config(sub_type))
+                .unwrap_or_else(|err| panic!("subtype '{sub_type}' should be accepted: {err}"));
+        }
+    }
+
+    #[test]
+    fn test_multiple_sum_rejects_whitespace_padded_subtype() {
+        let config = multiple_sum_config(" count ");
+        let err = match create_accumulator_updater(&config) {
+            Ok(_) => panic!("whitespace-padded MultipleSum subtype must fail"),
+            Err(err) => err,
+        };
+        assert!(err.contains(" count "));
+    }
+
     fn cms_heap_params_required() -> std::collections::HashMap<String, serde_json::Value> {
         let mut p = std::collections::HashMap::new();
         p.insert("depth".to_string(), serde_json::json!(3_u64));

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

@@ -258,9 +258,10 @@ impl PrecomputeEngine {
 
 /// Validate aggregation configs before starting any worker or ingest task.
 ///
-/// Count-Min Sketch configs carry their semantic contract in subtype fields or
-/// parameters; allowing an invalid value to reach the lazy worker path would
-/// leave the engine running while silently losing that contract.
+/// Count-Min Sketch and MultipleSum configs carry their semantic contract in
+/// subtype fields or parameters; allowing an invalid value to reach the lazy
+/// worker path would leave the engine running while silently losing that
+/// contract.
 fn validate_startup_aggregation_configs(
     streaming_config: &StreamingConfig,
 ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
@@ -269,7 +270,9 @@ fn validate_startup_aggregation_configs(
     for (&aggregation_id, config) in streaming_config.get_all_aggregation_configs() {
         if !matches!(
             config.aggregation_type,
-            AggregationType::CountMinSketch | AggregationType::CountMinSketchWithHeap
+            AggregationType::CountMinSketch
+                | AggregationType::CountMinSketchWithHeap
+                | AggregationType::MultipleSum
         ) {
             continue;
         }
@@ -362,6 +365,49 @@ mod tests {
         assert!(err.to_string().contains("sum") && err.to_string().contains("count"));
     }
 
+    #[tokio::test]
+    async fn run_rejects_invalid_multiple_sum_subtype_before_starting_workers() {
+        let multiple_sum = AggregationConfig::new(
+            1,
+            AggregationType::MultipleSum,
+            String::new(),
+            HashMap::new(),
+            promql_utilities::data_model::key_by_label_names::KeyByLabelNames::new(vec![]),
+            promql_utilities::data_model::key_by_label_names::KeyByLabelNames::new(vec![
+                "host".to_string()
+            ]),
+            promql_utilities::data_model::key_by_label_names::KeyByLabelNames::new(vec![]),
+            String::new(),
+            1_000,
+            1_000,
+            WindowType::Tumbling,
+            "requests_total".to_string(),
+            "requests_total".to_string(),
+            None,
+            None,
+            None,
+            None,
+        );
+        let engine = PrecomputeEngine::new(
+            PrecomputeEngineConfig {
+                num_workers: 1,
+                late_data_policy: LateDataPolicy::Drop,
+                ..PrecomputeEngineConfig::default()
+            },
+            Arc::new(StreamingConfig::new(HashMap::from([(1, multiple_sum)]))),
+            Arc::new(NoopOutputSink::new()),
+            vec![Box::new(ShutdownSource)],
+        );
+
+        let result = engine.run().await;
+        let err = match result {
+            Ok(()) => panic!("invalid MultipleSum subtype must fail before startup"),
+            Err(err) => err,
+        };
+        assert!(err.to_string().contains("aggregation_id 1"));
+        assert!(err.to_string().contains("sum") && err.to_string().contains("count"));
+    }
+
     #[tokio::test]
     async fn run_rejects_invalid_cms_with_heap_subtype_before_starting_workers() {
         let mut parameters = HashMap::new();