From b41eb1e89c258080091a37f1ab78ac506b6a6057 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Fri, 4 Sep 2026 22:22:04 -0400 Subject: [PATCH 1/4] refactor(asap-types): delete dead SingleSubpopulation/MultipleSubpopulation aggregation types (#670) Both variants were only ever constructed in #[cfg(test)] code; the real statistic-to-aggregation-type mapping never produces them. Removing them also deletes their legacy factory arm that reused aggregation_sub_type as an inner-accumulator-kind string and hardcoded SUM semantics for nested CMS (issue #670, Finding 856), since the dead code is gone rather than preserved. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01K2SLcSk9UnZi5iAWKt26kH --- .../src/query_logics/enums.rs | 10 +- .../src/optimizer/sketch_properties.rs | 4 - asap-planner-rs/src/planner/sketch.rs | 2 - .../src/bin/test_e2e_precompute.rs | 4 +- .../precompute_engine/accumulator_factory.rs | 88 ++-------- .../src/precompute_engine/worker.rs | 164 ++++-------------- .../src/stores/simple_map_store/mod.rs | 4 +- .../tests/e2e_netflow_single_second.rs | 4 +- ...mpute_wall_clock_fallback_active_ingest.rs | 4 +- 9 files changed, 56 insertions(+), 228 deletions(-) diff --git a/asap-common/dependencies/rs/promql_utilities/src/query_logics/enums.rs b/asap-common/dependencies/rs/promql_utilities/src/query_logics/enums.rs index 6af8c27..4ceb779 100644 --- a/asap-common/dependencies/rs/promql_utilities/src/query_logics/enums.rs +++ b/asap-common/dependencies/rs/promql_utilities/src/query_logics/enums.rs @@ -289,9 +289,6 @@ pub enum AggregationType { SetAggregator, DeltaSetAggregator, HLL, - // ---------- legacy config wrapper names ---------- - SingleSubpopulation, - MultipleSubpopulation, } impl AggregationType { @@ -310,8 +307,6 @@ impl AggregationType { AggregationType::SetAggregator => "SetAggregator", AggregationType::DeltaSetAggregator => "DeltaSetAggregator", AggregationType::HLL => "HLL", - AggregationType::SingleSubpopulation => "SingleSubpopulation", - AggregationType::MultipleSubpopulation => "MultipleSubpopulation", } } @@ -319,8 +314,7 @@ impl AggregationType { pub fn is_keyed(self) -> bool { matches!( self, - AggregationType::MultipleSubpopulation - | AggregationType::MultipleSum + AggregationType::MultipleSum | AggregationType::MultipleIncrease | AggregationType::MultipleMinMax | AggregationType::CountMinSketch @@ -377,8 +371,6 @@ impl FromStr for AggregationType { "SetAggregator" => Ok(AggregationType::SetAggregator), "DeltaSetAggregator" => Ok(AggregationType::DeltaSetAggregator), "HLL" | "HyperLogLog" => Ok(AggregationType::HLL), - "SingleSubpopulation" => Ok(AggregationType::SingleSubpopulation), - "MultipleSubpopulation" => Ok(AggregationType::MultipleSubpopulation), // Legacy accumulator-suffixed aliases "SumAccumulator" | "SumAggregator" | "sum" => Ok(AggregationType::Sum), "IncreaseAccumulator" | "IncreaseAggregator" | "increase" => { diff --git a/asap-planner-rs/src/optimizer/sketch_properties.rs b/asap-planner-rs/src/optimizer/sketch_properties.rs index 2f0e918..6a56b2e 100644 --- a/asap-planner-rs/src/optimizer/sketch_properties.rs +++ b/asap-planner-rs/src/optimizer/sketch_properties.rs @@ -33,10 +33,6 @@ pub fn sketch_properties(t: AggregationType) -> SketchProperties { p(true, false, false) } AggregationType::HLL => p(true, false, false), - // Legacy wrapper types: properties unknown; treat conservatively. - AggregationType::SingleSubpopulation | AggregationType::MultipleSubpopulation => { - p(false, false, false) - } } } diff --git a/asap-planner-rs/src/planner/sketch.rs b/asap-planner-rs/src/planner/sketch.rs index 1184c7c..076b270 100644 --- a/asap-planner-rs/src/planner/sketch.rs +++ b/asap-planner-rs/src/planner/sketch.rs @@ -138,8 +138,6 @@ pub fn build_sketch_parameters( m.insert("k".to_string(), serde_json::Value::Number(k.into())); Ok(m) } - - other => Err(format!("Aggregation type {} not supported", other)), } } diff --git a/asap-query-engine/src/bin/test_e2e_precompute.rs b/asap-query-engine/src/bin/test_e2e_precompute.rs index 6f542bf..be733f7 100644 --- a/asap-query-engine/src/bin/test_e2e_precompute.rs +++ b/asap-query-engine/src/bin/test_e2e_precompute.rs @@ -611,8 +611,8 @@ fn make_sum_agg_config( }; AggregationConfig::new( agg_id, - AggregationType::SingleSubpopulation, - "Sum".to_string(), + AggregationType::Sum, + "".to_string(), HashMap::new(), promql_utilities::data_model::key_by_label_names::KeyByLabelNames::new(vec![]), promql_utilities::data_model::key_by_label_names::KeyByLabelNames::new(vec![]), diff --git a/asap-query-engine/src/precompute_engine/accumulator_factory.rs b/asap-query-engine/src/precompute_engine/accumulator_factory.rs index a6c1219..35d1c12 100644 --- a/asap-query-engine/src/precompute_engine/accumulator_factory.rs +++ b/asap-query-engine/src/precompute_engine/accumulator_factory.rs @@ -40,7 +40,7 @@ macro_rules! impl_accumulator_methods { /// This provides a uniform interface over all accumulator types so that the /// worker loop doesn't need to know which concrete type it's dealing with. pub trait AccumulatorUpdater: Send { - /// Feed a single (value, timestamp_ms) pair — for SingleSubpopulation types. + /// Feed a single (value, timestamp_ms) pair — for single-population (non-keyed) types. fn update_single(&mut self, value: f64, timestamp_ms: i64); /// Feed a keyed (key, value, timestamp_ms) triple — for keyed aggregation types. @@ -796,8 +796,7 @@ impl AccumulatorUpdater for HydraKllAccumulatorUpdater { pub fn config_is_keyed(config: &AggregationConfig) -> bool { matches!( config.aggregation_type, - AggregationType::MultipleSubpopulation - | AggregationType::MultipleSum + AggregationType::MultipleSum | AggregationType::MultipleIncrease | AggregationType::MultipleMinMax | AggregationType::CountMinSketch @@ -933,35 +932,6 @@ pub fn create_accumulator_updater( let sub_type = config.aggregation_sub_type.as_str(); match config.aggregation_type { - AggregationType::SingleSubpopulation => match sub_type { - "Sum" | "sum" => Ok(Box::new(SumAccumulatorUpdater::new())), - "Min" | "min" => Ok(Box::new(MinMaxAccumulatorUpdater::new(false))), - "Max" | "max" => Ok(Box::new(MinMaxAccumulatorUpdater::new(true))), - "Increase" | "increase" => Ok(Box::new(IncreaseAccumulatorUpdater::new())), - "DatasketchesKLL" | "datasketches_kll" | "KLL" | "kll" => { - Ok(Box::new(KllAccumulatorUpdater::new(kll_k_param(config)?))) - } - other => Err(format!("Unknown SingleSubpopulation sub_type '{other}'")), - }, - AggregationType::MultipleSubpopulation => match sub_type { - "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())), - "CountMinSketch" | "count_min_sketch" | "CMS" | "cms" => { - let (row_num, col_num) = cms_params(config)?; - Ok(Box::new(CmsAccumulatorUpdater::new( - row_num, col_num, false, - ))) - } - "HydraKLL" | "hydra_kll" => { - let (row_num, col_num, k) = hydra_kll_params(config)?; - Ok(Box::new(HydraKllAccumulatorUpdater::new( - row_num, col_num, k, - ))) - } - other => Err(format!("Unknown MultipleSubpopulation sub_type '{other}'")), - }, AggregationType::DatasketchesKLL => { Ok(Box::new(KllAccumulatorUpdater::new(kll_k_param(config)?))) } @@ -1116,10 +1086,6 @@ mod tests { }; // Non-keyed types - assert!(!config_is_keyed(&make_config( - AggregationType::SingleSubpopulation, - "Sum" - ))); assert!(!config_is_keyed(&make_config(AggregationType::Sum, ""))); assert!(!config_is_keyed(&make_config( AggregationType::DatasketchesKLL, @@ -1131,10 +1097,6 @@ mod tests { ))); // Keyed types - assert!(config_is_keyed(&make_config( - AggregationType::MultipleSubpopulation, - "Sum" - ))); assert!(config_is_keyed(&make_config( AggregationType::MultipleSum, "sum" @@ -1153,21 +1115,14 @@ mod tests { ))); assert!(config_is_keyed(&make_config(AggregationType::HydraKLL, ""))); - // Verify agreement with updater.is_keyed() for types that need no sketch params. - for (agg_type, sub_type) in &[ - (AggregationType::SingleSubpopulation, "Sum"), - (AggregationType::MultipleSubpopulation, "Sum"), - (AggregationType::MultipleSum, "sum"), - ] { - let config = make_config(*agg_type, sub_type); - let updater = create_accumulator_updater(&config).unwrap(); - assert_eq!( - config_is_keyed(&config), - updater.is_keyed(), - "config_is_keyed disagrees with updater.is_keyed() for type={:?}", - agg_type - ); - } + // Verify agreement with updater.is_keyed() for a type that needs no sketch params. + let config = make_config(AggregationType::MultipleSum, "sum"); + let updater = create_accumulator_updater(&config).unwrap(); + assert_eq!( + config_is_keyed(&config), + updater.is_keyed(), + "config_is_keyed disagrees with updater.is_keyed() for MultipleSum", + ); // Sketch types require params — build configs with the required parameters. let make_config_with_params = @@ -1376,14 +1331,14 @@ mod tests { #[test] fn test_kll_k_param_capital_k() { - // SingleSubpopulation/KLL with capital "K" param should use it (not default to 200) + // Capital "K" param should be used (not defaulted to 200) use std::collections::HashMap; let mut params = HashMap::new(); params.insert("K".to_string(), serde_json::Value::from(50_u64)); let config = AggregationConfig::new( 1, - AggregationType::SingleSubpopulation, - "DatasketchesKLL".to_string(), + AggregationType::DatasketchesKLL, + "".to_string(), params, promql_utilities::data_model::key_by_label_names::KeyByLabelNames::new(vec![]), promql_utilities::data_model::key_by_label_names::KeyByLabelNames::new(vec![]), @@ -1990,21 +1945,4 @@ mod tests { ); } } - - #[test] - fn test_factory_rejects_unknown_subpopulation_sub_type() { - let mut config = key_aggregation_config(AggregationType::SingleSubpopulation); - config.aggregation_sub_type = "not-an-aggregation".to_string(); - let err = create_accumulator_updater(&config) - .err() - .expect("unknown subpopulation subtype must not default to Sum"); - assert!(err.contains("Unknown SingleSubpopulation sub_type")); - - let mut config = key_aggregation_config(AggregationType::MultipleSubpopulation); - config.aggregation_sub_type = "not-an-aggregation".to_string(); - let err = create_accumulator_updater(&config) - .err() - .expect("unknown subpopulation subtype must not default to MultipleSum"); - assert!(err.contains("Unknown MultipleSubpopulation sub_type")); - } } diff --git a/asap-query-engine/src/precompute_engine/worker.rs b/asap-query-engine/src/precompute_engine/worker.rs index 65ef4b3..f0ede6e 100644 --- a/asap-query-engine/src/precompute_engine/worker.rs +++ b/asap-query-engine/src/precompute_engine/worker.rs @@ -1231,8 +1231,8 @@ mod tests { let mut config = make_agg_config( 1, "netflow_table", - AggregationType::SingleSubpopulation, - "Sum", + AggregationType::Sum, + "", 1, 1, vec!["srcip"], @@ -1248,8 +1248,8 @@ mod tests { let config = make_agg_config( 1, "netflow_table", - AggregationType::SingleSubpopulation, - "Sum", + AggregationType::Sum, + "", 1, 1, vec!["srcip"], @@ -1643,15 +1643,7 @@ mod tests { #[test] fn test_tumbling_window_correctness() { // 10s tumbling window - let config = make_agg_config( - 1, - "cpu", - AggregationType::SingleSubpopulation, - "Sum", - 10_000, - 0, - vec![], - ); + let config = make_agg_config(1, "cpu", AggregationType::Sum, "", 10_000, 0, vec![]); let mut agg_configs = HashMap::new(); agg_configs.insert(1, config); @@ -1709,15 +1701,7 @@ mod tests { /// endpoint samples cannot be dropped. #[test] fn first_batch_boundary_sample_waits_for_watermark_to_advance() { - let config = make_agg_config( - 20, - "cpu", - AggregationType::SingleSubpopulation, - "Sum", - 1_000, - 0, - vec![], - ); + let config = make_agg_config(20, "cpu", AggregationType::Sum, "", 1_000, 0, vec![]); let sink = Arc::new(CapturingOutputSink::new()); let mut worker = make_worker( arc_configs(HashMap::from([(20, config)])), @@ -1749,15 +1733,7 @@ mod tests { #[test] fn timestamp_zero_uses_the_nonnegative_origin_window() { - let config = make_agg_config( - 21, - "cpu", - AggregationType::SingleSubpopulation, - "Sum", - 1_000, - 0, - vec![], - ); + let config = make_agg_config(21, "cpu", AggregationType::Sum, "", 1_000, 0, vec![]); let sink = Arc::new(CapturingOutputSink::new()); let mut worker = make_worker( arc_configs(HashMap::from([(21, config)])), @@ -1863,17 +1839,9 @@ mod tests { #[test] fn test_group_by_merges_series() { - // SingleSubpopulation Sum with no grouping labels + // Sum with no grouping labels // Two different series in the same group → both feed same accumulator - let config = make_agg_config( - 1, - "cpu", - AggregationType::SingleSubpopulation, - "Sum", - 10_000, - 0, - vec![], - ); + let config = make_agg_config(1, "cpu", AggregationType::Sum, "", 10_000, 0, vec![]); let mut agg_configs = HashMap::new(); agg_configs.insert(1, config); @@ -1936,8 +1904,8 @@ mod tests { let config = make_agg_config( 1, "cpu", - AggregationType::SingleSubpopulation, - "Sum", + AggregationType::Sum, + "", 10_000, 0, vec!["pattern"], @@ -2091,15 +2059,7 @@ mod tests { #[test] fn test_sliding_window_pane_sharing() { // 30s window, 10s slide → W=3 panes per window - let config = make_agg_config( - 2, - "cpu", - AggregationType::SingleSubpopulation, - "Sum", - 30_000, - 10_000, - vec![], - ); + let config = make_agg_config(2, "cpu", AggregationType::Sum, "", 30_000, 10_000, vec![]); let mut agg_configs = HashMap::new(); agg_configs.insert(2, config); @@ -2149,7 +2109,7 @@ mod tests { } // ----------------------------------------------------------------------- - // Test: MultipleSubpopulation — keyed accumulator with aggregated labels + // Test: MultipleSum — keyed accumulator with aggregated labels // Matches planner output: grouping=[], aggregated=[host] // All series go to one group, host is the key dimension INSIDE the sketch // ----------------------------------------------------------------------- @@ -2161,8 +2121,8 @@ mod tests { let config = make_agg_config_full( 3, "cpu", - AggregationType::MultipleSubpopulation, - "Sum", + AggregationType::MultipleSum, + "sum", 10_000, 0, vec![], // grouping: empty — one output group @@ -2247,8 +2207,8 @@ mod tests { let mut config = make_agg_config_full( 5, "netflow_table", - AggregationType::MultipleSubpopulation, - "Sum", + AggregationType::MultipleSum, + "sum", 10_000, 0, vec![], // grouping: empty — one output group @@ -2614,15 +2574,7 @@ mod tests { #[test] fn test_late_data_drop() { - let config = make_agg_config( - 4, - "cpu", - AggregationType::SingleSubpopulation, - "Sum", - 10_000, - 0, - vec![], - ); + let config = make_agg_config(4, "cpu", AggregationType::Sum, "", 10_000, 0, vec![]); let mut agg_configs = HashMap::new(); agg_configs.insert(4, config); @@ -2667,15 +2619,7 @@ mod tests { #[test] fn test_late_data_forward_to_store() { - let config = make_agg_config( - 5, - "cpu", - AggregationType::SingleSubpopulation, - "Sum", - 10_000, - 0, - vec![], - ); + let config = make_agg_config(5, "cpu", AggregationType::Sum, "", 10_000, 0, vec![]); let mut agg_configs = HashMap::new(); agg_configs.insert(5, config); @@ -2852,8 +2796,8 @@ mod tests { let yaml = r#" aggregations: - aggregationId: 10 - aggregationType: SingleSubpopulation - aggregationSubType: Sum + aggregationType: Sum + aggregationSubType: '' labels: grouping: [] rollup: [] @@ -2916,8 +2860,8 @@ aggregations: fn test_extract_key_from_series() { let config = AggregationConfig::new( 1, - AggregationType::SingleSubpopulation, - "Sum".to_string(), + AggregationType::Sum, + "".to_string(), HashMap::new(), promql_utilities::data_model::key_by_label_names::KeyByLabelNames::new(vec![ "method".to_string(), @@ -2965,15 +2909,7 @@ aggregations: // Two groups on the same worker. Group A advances to t=100s. // Group B has data at t=10s and then goes idle. // After flush, group B's idle windows should close via propagation. - let config = make_agg_config( - 1, - "cpu", - AggregationType::SingleSubpopulation, - "Sum", - 10_000, - 0, - vec![], - ); + let config = make_agg_config(1, "cpu", AggregationType::Sum, "", 10_000, 0, vec![]); let agg_configs = arc_configs(HashMap::from([(1, config)])); let sink = Arc::new(CapturingOutputSink::new()); let mut worker = make_worker(agg_configs, sink.clone(), false, 0, LateDataPolicy::Drop); @@ -3110,15 +3046,7 @@ aggregations: #[test] fn test_flush_publishes_worker_watermark() { - let config = make_agg_config( - 1, - "cpu", - AggregationType::SingleSubpopulation, - "Sum", - 10_000, - 0, - vec![], - ); + let config = make_agg_config(1, "cpu", AggregationType::Sum, "", 10_000, 0, vec![]); let agg_configs = arc_configs(HashMap::from([(1, config)])); let sink = Arc::new(CapturingOutputSink::new()); let wm = Arc::new(AtomicI64::new(i64::MIN)); @@ -3167,8 +3095,8 @@ aggregations: let config = make_agg_config( 1, "cpu", - AggregationType::SingleSubpopulation, - "Sum", + AggregationType::Sum, + "", 10_000, // 10s tumbling window 0, vec![], @@ -3286,8 +3214,8 @@ aggregations: let config = make_agg_config( 1, "cpu", - AggregationType::SingleSubpopulation, - "Sum", + AggregationType::Sum, + "", window_size_ms, 0, vec![], @@ -3405,15 +3333,7 @@ aggregations: #[test] fn wall_clock_fallback_closes_idle_window() { // 10s tumbling window. - let cfg = make_agg_config( - 7, - "cpu", - AggregationType::SingleSubpopulation, - "Sum", - 10_000, - 0, - vec![], - ); + let cfg = make_agg_config(7, "cpu", AggregationType::Sum, "", 10_000, 0, vec![]); let agg_configs = HashMap::from([(7, cfg)]); let sink = Arc::new(CapturingOutputSink::new()); // 5s grace period — production default. @@ -3485,8 +3405,8 @@ aggregations: let cfg = make_agg_config( 7, "netflow_bytes", - AggregationType::SingleSubpopulation, - "Sum", + AggregationType::Sum, + "", 1_000, 0, vec![], @@ -3565,15 +3485,7 @@ aggregations: #[test] fn wall_clock_fallback_disabled_preserves_event_time_only_semantics() { - let cfg = make_agg_config( - 7, - "cpu", - AggregationType::SingleSubpopulation, - "Sum", - 10_000, - 0, - vec![], - ); + let cfg = make_agg_config(7, "cpu", AggregationType::Sum, "", 10_000, 0, vec![]); let agg_configs = HashMap::from([(7, cfg)]); let sink = Arc::new(CapturingOutputSink::new()); // grace=0 disables the fallback entirely. @@ -3612,15 +3524,7 @@ aggregations: fn shutdown_force_close_emits_trailing_window() { // 10s tumbling window; grace=0 isolates the force-close from the // wall-clock fallback. - let cfg = make_agg_config( - 7, - "cpu", - AggregationType::SingleSubpopulation, - "Sum", - 10_000, - 0, - vec![], - ); + let cfg = make_agg_config(7, "cpu", AggregationType::Sum, "", 10_000, 0, vec![]); let agg_configs = HashMap::from([(7, cfg)]); let sink = Arc::new(CapturingOutputSink::new()); let mut worker = make_worker_with_grace(agg_configs, sink.clone(), 0); diff --git a/asap-query-engine/src/stores/simple_map_store/mod.rs b/asap-query-engine/src/stores/simple_map_store/mod.rs index 886d326..8f0ad01 100644 --- a/asap-query-engine/src/stores/simple_map_store/mod.rs +++ b/asap-query-engine/src/stores/simple_map_store/mod.rs @@ -177,8 +177,8 @@ mod tests { fn make_agg_config(id: u64, metric: &str) -> AggregationConfig { AggregationConfig::new( id, - AggregationType::SingleSubpopulation, - "Sum".to_string(), + AggregationType::Sum, + "".to_string(), HashMap::new(), KeyByLabelNames::new(vec![]), KeyByLabelNames::new(vec![]), diff --git a/asap-query-engine/tests/e2e_netflow_single_second.rs b/asap-query-engine/tests/e2e_netflow_single_second.rs index 604f64b..3196f2c 100644 --- a/asap-query-engine/tests/e2e_netflow_single_second.rs +++ b/asap-query-engine/tests/e2e_netflow_single_second.rs @@ -30,8 +30,8 @@ use query_engine_rust::precompute_operators::sum_accumulator::SumAccumulator; fn netflow_agg_config(metric: &str, window_size_ms: u64) -> AggregationConfig { AggregationConfig::new( 1, - AggregationType::SingleSubpopulation, - "Sum".to_string(), + AggregationType::Sum, + "".to_string(), HashMap::new(), promql_utilities::data_model::key_by_label_names::KeyByLabelNames::new(vec![]), promql_utilities::data_model::key_by_label_names::KeyByLabelNames::new(vec![]), diff --git a/asap-query-engine/tests/e2e_precompute_wall_clock_fallback_active_ingest.rs b/asap-query-engine/tests/e2e_precompute_wall_clock_fallback_active_ingest.rs index 1911727..150f3dc 100644 --- a/asap-query-engine/tests/e2e_precompute_wall_clock_fallback_active_ingest.rs +++ b/asap-query-engine/tests/e2e_precompute_wall_clock_fallback_active_ingest.rs @@ -50,8 +50,8 @@ use query_engine_rust::precompute_operators::sum_accumulator::SumAccumulator; fn netflow_agg_config(metric: &str, window_size_ms: u64) -> AggregationConfig { AggregationConfig::new( 1, - AggregationType::SingleSubpopulation, - "Sum".to_string(), + AggregationType::Sum, + "".to_string(), HashMap::new(), promql_utilities::data_model::key_by_label_names::KeyByLabelNames::new(vec![]), promql_utilities::data_model::key_by_label_names::KeyByLabelNames::new(vec![]), From 2e950f50f5e02195c06aa7d494335a7a6c3d4647 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Fri, 4 Sep 2026 22:51:38 -0400 Subject: [PATCH 2/4] refactor(asap-types): fold heap-CMS count_events into aggregation_sub_type (#670) CountMinSketchWithHeap previously split its SUM/COUNT weighting across two places: aggregation_sub_type (fixed to the now-redundant "topk") and a separate parameters["count_events"] boolean, duplicating and diverging from how plain CountMinSketch/MultipleSum already encode the same axis via sub_type alone. Hard-cutover to aggregation_sub_type: "sum"|"count" for all three CMS-family types, removing count_events entirely. Introduces asap_types::aggregation_mode (AggregationMode/CountMode/ MinMaxMode) as the single typed seam AggregationConfig::mode() exposes; capability_matching's three separate weighting-compatibility functions and accumulator_factory's duplicate sub_type parsers now route through it instead of each re-deriving the same semantics. The planner (promql.rs, sql.rs) patches a topk candidate's sub_type to the detected weighting right after building it, since map_statistic_to_ precompute_operator's placeholder "topk" sub_type is only a stand-in until the real weighting is known; candidate_gen.rs/atomic_costs.rs updated to match. Backward-compat: none needed, AggregationConfig is regenerated by the planner per query rather than persisted. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01K2SLcSk9UnZi5iAWKt26kH --- .../rs/asap_types/src/aggregation_config.rs | 77 ++------ .../rs/asap_types/src/aggregation_mode.rs | 83 +++++++++ .../rs/asap_types/src/capability_matching.rs | 172 +++++++----------- .../dependencies/rs/asap_types/src/lib.rs | 1 + .../rs/asap_types/src/streaming_config.rs | 74 +++----- asap-planner-rs/src/optimizer/atomic_costs.rs | 35 +--- .../src/optimizer/candidate_gen.rs | 112 ++++++------ asap-planner-rs/src/planner/promql.rs | 18 +- asap-planner-rs/src/planner/sketch.rs | 12 +- asap-planner-rs/src/planner/sql.rs | 7 +- asap-planner-rs/tests/integration.rs | 15 +- asap-planner-rs/tests/sql_integration.rs | 21 +-- .../src/engines/simple_engine/sql.rs | 31 ++-- .../precompute_engine/accumulator_factory.rs | 127 ++++--------- .../src/precompute_engine/engine.rs | 3 +- 15 files changed, 340 insertions(+), 448 deletions(-) create mode 100644 asap-common/dependencies/rs/asap_types/src/aggregation_mode.rs diff --git a/asap-common/dependencies/rs/asap_types/src/aggregation_config.rs b/asap-common/dependencies/rs/asap_types/src/aggregation_config.rs index 9d3e680..62baac2 100644 --- a/asap-common/dependencies/rs/asap_types/src/aggregation_config.rs +++ b/asap-common/dependencies/rs/asap_types/src/aggregation_config.rs @@ -3,6 +3,7 @@ use serde_json::Value; use serde_yaml; use std::collections::HashMap; +use crate::aggregation_mode::AggregationMode; use crate::enums::{QueryLanguage, WindowType}; use crate::traits::SerializableToSink; use crate::utils::normalize_spatial_filter; @@ -16,20 +17,11 @@ pub const HLL_MAX_PRECISION: u32 = 18; #[derive(Debug, thiserror::Error)] pub enum AggregationConfigError { - #[error( - "aggregation {aggregation_id} (CountMinSketchWithHeap) missing required parameter 'count_events'" - )] - MissingCountEvents { aggregation_id: u64 }, - #[error( - "aggregation {aggregation_id} (CountMinSketchWithHeap) parameter 'count_events' must be a boolean, got {value}" - )] - InvalidCountEventsType { aggregation_id: u64, value: Value }, - #[error( - "aggregation {aggregation_id} ({aggregation_type}) parameter 'count_events' is only valid for CountMinSketchWithHeap" - )] - MisplacedCountEvents { + #[error("aggregation {aggregation_id} ({aggregation_type}): {reason}")] + InvalidSubType { aggregation_id: u64, aggregation_type: AggregationType, + reason: String, }, #[error("aggregation {aggregation_id} (HLL) missing required parameter 'precision'")] MissingPrecision { aggregation_id: u64 }, @@ -48,14 +40,6 @@ pub enum AggregationConfigError { aggregation_id: u64, aggregation_type: AggregationType, }, - #[error( - "aggregation {aggregation_id} ({aggregation_type}) aggregation_sub_type must be 'min' or 'max', got '{sub_type}'" - )] - InvalidMinMaxSubType { - aggregation_id: u64, - aggregation_type: AggregationType, - sub_type: String, - }, } #[derive(Debug, Clone, Serialize, Deserialize)] @@ -99,34 +83,22 @@ pub struct AggregationIdInfo { impl AggregationConfig { pub fn validate(&self) -> Result<(), AggregationConfigError> { - self.validate_count_events()?; + self.mode().map(|_| ())?; self.validate_hll_precision()?; - self.validate_minmax_subtype()?; Ok(()) } - fn validate_count_events(&self) -> Result<(), AggregationConfigError> { - if self.aggregation_type == AggregationType::CountMinSketchWithHeap { - match self.parameters.get("count_events") { - None => Err(AggregationConfigError::MissingCountEvents { - aggregation_id: self.aggregation_id, - }), - Some(value) if !value.is_boolean() => { - Err(AggregationConfigError::InvalidCountEventsType { - aggregation_id: self.aggregation_id, - value: value.clone(), - }) - } - Some(_) => Ok(()), - } - } else if self.parameters.contains_key("count_events") { - Err(AggregationConfigError::MisplacedCountEvents { + /// The typed meaning of `aggregation_sub_type` for this config's + /// `aggregation_type`. The single seam factories, planners, and capability + /// matching all dispatch through instead of re-parsing the raw string (#670). + pub fn mode(&self) -> Result { + AggregationMode::parse(self.aggregation_type, &self.aggregation_sub_type).map_err( + |reason| AggregationConfigError::InvalidSubType { aggregation_id: self.aggregation_id, aggregation_type: self.aggregation_type, - }) - } else { - Ok(()) - } + reason, + }, + ) } fn validate_hll_precision(&self) -> Result<(), AggregationConfigError> { @@ -162,27 +134,6 @@ impl AggregationConfig { } } - fn validate_minmax_subtype(&self) -> Result<(), AggregationConfigError> { - let is_minmax = matches!( - self.aggregation_type, - AggregationType::MinMax | AggregationType::MultipleMinMax - ); - if !is_minmax { - return Ok(()); - } - if self.aggregation_sub_type.eq_ignore_ascii_case("min") - || self.aggregation_sub_type.eq_ignore_ascii_case("max") - { - Ok(()) - } else { - Err(AggregationConfigError::InvalidMinMaxSubType { - aggregation_id: self.aggregation_id, - aggregation_type: self.aggregation_type, - sub_type: self.aggregation_sub_type.clone(), - }) - } - } - #[allow(clippy::too_many_arguments)] pub fn new( aggregation_id: u64, diff --git a/asap-common/dependencies/rs/asap_types/src/aggregation_mode.rs b/asap-common/dependencies/rs/asap_types/src/aggregation_mode.rs new file mode 100644 index 0000000..a163b6d --- /dev/null +++ b/asap-common/dependencies/rs/asap_types/src/aggregation_mode.rs @@ -0,0 +1,83 @@ +use promql_utilities::query_logics::enums::AggregationType; + +/// Whether a sample contributes its value (SUM) or a unit weight (COUNT). +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum CountMode { + Sum, + Count, +} + +/// Which end of the value range a MinMax-family aggregation tracks. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum MinMaxMode { + Min, + Max, +} + +/// The typed meaning of `aggregation_sub_type`, resolved for a given +/// `AggregationType`. Aggregation kinds that share a sub_type axis (e.g. plain +/// `CountMinSketch` and `CountMinSketchWithHeap` both carry SUM/COUNT +/// weighting) resolve to the same variant here, so callers dispatch on one +/// typed value instead of re-parsing the raw string per kind (#670). +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum AggregationMode { + /// This aggregation type has no sub_type axis. + None, + SumOrCount(CountMode), + MinOrMax(MinMaxMode), +} + +impl AggregationMode { + /// Parse `sub_type` according to the axis `agg_type` uses. Case-insensitive, + /// matching the existing wire convention. + pub fn parse(agg_type: AggregationType, sub_type: &str) -> Result { + match agg_type { + AggregationType::CountMinSketch + | AggregationType::MultipleSum + | AggregationType::CountMinSketchWithHeap => { + if sub_type.eq_ignore_ascii_case("sum") { + Ok(AggregationMode::SumOrCount(CountMode::Sum)) + } else if sub_type.eq_ignore_ascii_case("count") { + Ok(AggregationMode::SumOrCount(CountMode::Count)) + } else { + Err(format!( + "{agg_type} requires aggregation_sub_type 'sum' or 'count', got '{sub_type}'" + )) + } + } + AggregationType::MinMax | AggregationType::MultipleMinMax => { + if sub_type.eq_ignore_ascii_case("min") { + Ok(AggregationMode::MinOrMax(MinMaxMode::Min)) + } else if sub_type.eq_ignore_ascii_case("max") { + Ok(AggregationMode::MinOrMax(MinMaxMode::Max)) + } else { + Err(format!( + "aggregation_sub_type must be 'min' or 'max', got '{sub_type}'" + )) + } + } + _ => Ok(AggregationMode::None), + } + } + + /// The canonical wire string for this mode, e.g. for planner emission. + pub fn as_sub_type_str(self) -> &'static str { + match self { + AggregationMode::None => "", + AggregationMode::SumOrCount(CountMode::Sum) => "sum", + AggregationMode::SumOrCount(CountMode::Count) => "count", + AggregationMode::MinOrMax(MinMaxMode::Min) => "min", + AggregationMode::MinOrMax(MinMaxMode::Max) => "max", + } + } +} + +impl CountMode { + pub fn from_count_events(count_events: bool) -> Self { + if count_events { + CountMode::Count + } else { + CountMode::Sum + } + } +} diff --git a/asap-common/dependencies/rs/asap_types/src/capability_matching.rs b/asap-common/dependencies/rs/asap_types/src/capability_matching.rs index 6edd88d..fe09b4e 100644 --- a/asap-common/dependencies/rs/asap_types/src/capability_matching.rs +++ b/asap-common/dependencies/rs/asap_types/src/capability_matching.rs @@ -6,6 +6,7 @@ use promql_utilities::query_logics::enums::Statistic; use tracing::{debug, warn}; use crate::aggregation_config::{AggregationConfig, AggregationConfigError, AggregationIdInfo}; +use crate::aggregation_mode::{AggregationMode, CountMode}; use crate::enums::WindowType; use crate::query_requirements::QueryRequirements; use crate::utils::normalize_spatial_filter; @@ -166,24 +167,23 @@ pub fn spatial_filter_compatible(config_filter: &str, req_filter: &str) -> bool config_norm == req_norm } -/// Reads the required `count_events` parameter from a validated -/// `CountMinSketchWithHeap` config. -fn config_count_events(config: &AggregationConfig) -> bool { - config - .parameters - .get("count_events") - .and_then(|v| v.as_bool()) - .expect("aggregation configs are validated before capability matching") +/// Reads the SUM/COUNT weighting from a validated CMS-family config +/// (`CountMinSketch` or `CountMinSketchWithHeap`). +fn config_count_mode(config: &AggregationConfig) -> CountMode { + match config.mode() { + Ok(AggregationMode::SumOrCount(mode)) => mode, + _ => panic!("aggregation configs are validated before capability matching"), + } } /// Top-k weighting compatibility. Only constrains `Statistic::Topk` candidates; /// every other statistic passes unconditionally. /// -/// A COUNT top-k query (`Some(true)`) must be served by a `count_events: true` -/// sketch and a SUM top-k query (`Some(false)`) by a `count_events: false` -/// (value-weighted) sketch. This is what tells two `CountMinSketchWithHeap` -/// configs on the same metric apart. `None` (non-top-k, or PromQL top-k which -/// does not pin the weighting) imposes no constraint. +/// A COUNT top-k query (`Some(true)`) must be served by a COUNT-weighted sketch +/// and a SUM top-k query (`Some(false)`) by a SUM-weighted (value-weighted) +/// sketch. This is what tells two `CountMinSketchWithHeap` configs on the same +/// metric apart. `None` (non-top-k, or PromQL top-k which does not pin the +/// weighting) imposes no constraint. pub fn topk_weighting_compatible( stat: Statistic, config: &AggregationConfig, @@ -193,41 +193,31 @@ pub fn topk_weighting_compatible( return true; } match req_count_events { - Some(want) => config_count_events(config) == want, + Some(want) => config_count_mode(config) == CountMode::from_count_events(want), None => true, } } -/// Ordinary COUNT vs a heap-backed Count-Min Sketch. A heap configured with -/// `count_events: false` is value-weighted (SUM semantics) and must not serve -/// ordinary `COUNT(...)`, even though `CountMinSketchWithHeap` is otherwise a -/// compatible aggregation type for `Statistic::Count` (#666). Only constrains -/// heap CMS candidates for `Statistic::Count`; every other case passes -/// unconditionally. -fn count_heap_weighting_compatible(stat: Statistic, config: &AggregationConfig) -> bool { - if stat != Statistic::Count - || config.aggregation_type != AggregationType::CountMinSketchWithHeap - { - return true; - } - config_count_events(config) -} - -/// Plain Count-Min Sketches are value-weighted or event-weighted according to -/// their subtype. A sketch with the other subtype cannot serve this statistic. -fn plain_cms_sub_type_compatible(stat: Statistic, config: &AggregationConfig) -> bool { - if config.aggregation_type != AggregationType::CountMinSketch { +/// CMS-family sketches (`CountMinSketch`, `CountMinSketchWithHeap`) are SUM- or +/// COUNT-weighted according to their `aggregation_sub_type`. A SUM statistic +/// must be served by a SUM-weighted sketch and a COUNT statistic by a +/// COUNT-weighted one (this also covers heap CMS serving ordinary `COUNT(...)`, +/// #666); other statistics (e.g. Topk, handled by `topk_weighting_compatible`) +/// aren't constrained here. +fn cms_family_sub_type_compatible(stat: Statistic, config: &AggregationConfig) -> bool { + if !matches!( + config.aggregation_type, + AggregationType::CountMinSketch | AggregationType::CountMinSketchWithHeap + ) { return true; } - let expected_sub_type = match stat { - Statistic::Sum => "sum", - Statistic::Count => "count", - _ => unreachable!("plain CMS matching only supports SUM and COUNT"), + let expected = match stat { + Statistic::Sum => CountMode::Sum, + Statistic::Count => CountMode::Count, + _ => return true, }; - config - .aggregation_sub_type - .eq_ignore_ascii_case(expected_sub_type) + config_count_mode(config) == expected } /// Aggregation priority comparator: prefer larger `window_size_ms` (descending). @@ -300,9 +290,8 @@ pub fn find_compatible_aggregation( &c.spatial_filter_normalized, &requirements.spatial_filter_normalized, ) - && plain_cms_sub_type_compatible(stat, c) - && topk_weighting_compatible(stat, c, requirements.topk_count_events) - && count_heap_weighting_compatible(stat, c); + && cms_family_sub_type_compatible(stat, c) + && topk_weighting_compatible(stat, c, requirements.topk_count_events); if !ok { debug!( agg_id = c.aggregation_id, @@ -480,11 +469,6 @@ mod tests { table_name: None, value_column: None, }; - if config.aggregation_type == AggregationType::CountMinSketchWithHeap { - config - .parameters - .insert("count_events".to_string(), serde_json::Value::Bool(true)); - } if config.aggregation_type == AggregationType::HLL { config .parameters @@ -615,7 +599,9 @@ mod tests { } #[test] - fn plain_cms_with_invalid_subtypes_is_excluded_from_matching() { + fn plain_cms_with_invalid_sub_type_is_rejected() { + // Consistent with MinMax and heap-CMS: an invalid sub_type is malformed + // configuration, caught at validate() rather than silently excluded. for invalid_sub_type in ["", "unknown", " sum "] { let configs = single_config(make_config( 1, @@ -628,15 +614,22 @@ mod tests { "", )); - let result = find_compatible_aggregation( + let error = super::find_compatible_aggregation( &configs, &req("cpu", &[Statistic::Sum], 300_000, &[], ""), - ); - - assert!( - result.is_none(), - "invalid plain CMS subtype {invalid_sub_type:?} must not match SUM" - ); + ) + .expect_err(&format!( + "invalid plain CMS subtype {invalid_sub_type:?} must be rejected" + )); + assert!(matches!( + error, + CapabilityMatchingError::InvalidAggregationConfig( + AggregationConfigError::InvalidSubType { + aggregation_id: 1, + .. + } + ) + )); } } @@ -1049,7 +1042,7 @@ mod tests { 10, "req", "CountMinSketchWithHeap", - "", + "count", 300_000, "tumbling", &[], @@ -1083,7 +1076,7 @@ mod tests { 10, "req", "CountMinSketchWithHeap", - "", + "count", 300_000, "tumbling", &[], @@ -1206,7 +1199,7 @@ mod tests { 10, "req", "CountMinSketchWithHeap", - "", + "count", 300_000, "tumbling", &[], @@ -1245,7 +1238,7 @@ mod tests { 10, "req", "CountMinSketchWithHeap", - "", + "count", 300_000, "sliding", &[], @@ -1596,7 +1589,7 @@ mod tests { 2, "cpu", "CountMinSketch", - "", + "count", 900_000, "tumbling", &["job"], @@ -1638,23 +1631,19 @@ mod tests { ) } - /// `CountMinSketchWithHeap` config with an explicit `count_events` parameter. + /// `CountMinSketchWithHeap` config with an explicit SUM/COUNT weighting. fn make_topk_config(id: u64, metric: &str, count_events: bool) -> AggregationConfig { - let mut c = make_config( + let sub_type = if count_events { "count" } else { "sum" }; + make_config( id, metric, "CountMinSketchWithHeap", - "", + sub_type, 1_000, "tumbling", &[], "", - ); - c.parameters.insert( - "count_events".to_string(), - serde_json::Value::Bool(count_events), - ); - c + ) } fn topk_req(metric: &str, count_events: Option) -> QueryRequirements { @@ -1704,11 +1693,12 @@ mod tests { } #[test] - fn topk_matching_rejects_sketch_without_count_events() { - // A missing weighting flag is malformed configuration, not implicit - // COUNT semantics. Capability matching must distinguish it from a miss. + fn topk_matching_rejects_sketch_with_invalid_sub_type() { + // A missing/unrecognized weighting is malformed configuration, not + // implicit COUNT semantics. Capability matching must distinguish it + // from a miss. let mut configs = HashMap::new(); - let mut malformed = make_config( + let malformed = make_config( 7, "netflow_table", "CountMinSketchWithHeap", @@ -1718,16 +1708,18 @@ mod tests { &[], "", ); - malformed.parameters.remove("count_events"); configs.insert(7, malformed); configs.insert(9, make_key_agg(9, "netflow_table")); let error = super::find_compatible_aggregation(&configs, &topk_req("netflow_table", Some(true))) - .expect_err("missing count_events must be a capability-matching error"); + .expect_err("invalid sub_type must be a capability-matching error"); assert!(matches!( error, CapabilityMatchingError::InvalidAggregationConfig( - AggregationConfigError::MissingCountEvents { aggregation_id: 7 } + AggregationConfigError::InvalidSubType { + aggregation_id: 7, + .. + } ) )); } @@ -1779,32 +1771,6 @@ mod tests { assert_eq!(result.aggregation_id_for_value, 1); } - #[test] - fn ordinary_count_accepts_heap_with_default_count_events() { - // Configs that omit `count_events` default to count semantics. - let mut configs = HashMap::new(); - configs.insert( - 7, - make_config( - 7, - "netflow_table", - "CountMinSketchWithHeap", - "", - 1_000, - "tumbling", - &[], - "", - ), - ); - configs.insert(9, make_key_agg(9, "netflow_table")); - let result = find_compatible_aggregation( - &configs, - &req("netflow_table", &[Statistic::Count], 1_000, &[], ""), - ) - .expect("ordinary COUNT should match a heap sketch with default count_events"); - assert_eq!(result.aggregation_id_for_value, 7); - } - #[test] fn ordinary_count_picks_count_weighted_over_sum_weighted_heap() { // Both variants present on the same metric: COUNT must resolve to the diff --git a/asap-common/dependencies/rs/asap_types/src/lib.rs b/asap-common/dependencies/rs/asap_types/src/lib.rs index 2ae3471..167c753 100644 --- a/asap-common/dependencies/rs/asap_types/src/lib.rs +++ b/asap-common/dependencies/rs/asap_types/src/lib.rs @@ -1,4 +1,5 @@ pub mod aggregation_config; +pub mod aggregation_mode; pub mod aggregation_reference; pub mod capability_matching; pub mod enums; diff --git a/asap-common/dependencies/rs/asap_types/src/streaming_config.rs b/asap-common/dependencies/rs/asap_types/src/streaming_config.rs index 1ddd96a..e73dbe3 100644 --- a/asap-common/dependencies/rs/asap_types/src/streaming_config.rs +++ b/asap-common/dependencies/rs/asap_types/src/streaming_config.rs @@ -138,7 +138,10 @@ mod tests { use crate::aggregation_config::AggregationConfigError; #[test] - fn rejects_heap_config_without_count_events() { + fn rejects_heap_config_with_invalid_sub_type() { + // Heap-CMS weighting is carried by aggregation_sub_type ('sum'/'count') + // rather than a separate count_events parameter; the old 'topk' value + // is no longer accepted (#670). let yaml: Value = serde_yaml::from_str( r#" aggregations: @@ -163,60 +166,30 @@ aggregations: .unwrap(); let error = StreamingConfig::from_yaml_data(&yaml, None) - .expect_err("heap config without count_events must be rejected"); + .expect_err("heap config with an unrecognized sub_type must be rejected"); assert!(matches!( error.downcast_ref::(), - Some(AggregationConfigError::MissingCountEvents { aggregation_id: 1 }) + Some(AggregationConfigError::InvalidSubType { + aggregation_id: 1, + .. + }) )); } #[test] - fn rejects_heap_config_with_non_boolean_count_events() { - let yaml: Value = serde_yaml::from_str( - r#" + fn accepts_heap_config_with_sum_or_count_sub_type() { + for sub_type in ["sum", "count"] { + let yaml: Value = serde_yaml::from_str(&format!( + r#" aggregations: - aggregationId: 1 aggregationType: CountMinSketchWithHeap - aggregationSubType: topk + aggregationSubType: {sub_type} parameters: depth: 3 width: 1024 heapsize: 20 - count_events: "false" - labels: - grouping: [] - aggregated: [instance] - rollup: [] - metric: http_requests_total - windowSizeMs: 15000 - slideIntervalMs: 15000 - windowType: tumbling - spatialFilter: '' -"#, - ) - .unwrap(); - - let error = StreamingConfig::from_yaml_data(&yaml, None) - .expect_err("heap config with non-boolean count_events must be rejected"); - - assert!(error.to_string().contains("aggregation 1")); - assert!(error.to_string().contains("count_events")); - assert!(error.to_string().contains("boolean")); - } - - #[test] - fn rejects_count_events_on_non_heap_config() { - let yaml: Value = serde_yaml::from_str( - r#" -aggregations: - - aggregationId: 1 - aggregationType: CountMinSketch - aggregationSubType: sum - parameters: - depth: 3 - width: 1024 - count_events: false labels: grouping: [] aggregated: [instance] @@ -226,18 +199,13 @@ aggregations: slideIntervalMs: 15000 windowType: tumbling spatialFilter: '' -"#, - ) - .unwrap(); - - let error = StreamingConfig::from_yaml_data(&yaml, None) - .expect_err("count_events on a non-heap aggregation must be rejected"); +"# + )) + .unwrap(); - assert!(error.to_string().contains("aggregation 1")); - assert!(error.to_string().contains("count_events")); - assert!(error - .to_string() - .contains("only valid for CountMinSketchWithHeap")); + StreamingConfig::from_yaml_data(&yaml, None) + .unwrap_or_else(|e| panic!("sub_type '{sub_type}' should be accepted: {e}")); + } } fn hll_yaml(parameters_yaml: &str) -> Value { @@ -383,7 +351,7 @@ aggregations: assert!(matches!( error.downcast_ref::(), - Some(AggregationConfigError::InvalidMinMaxSubType { + Some(AggregationConfigError::InvalidSubType { aggregation_id: 1, .. }) diff --git a/asap-planner-rs/src/optimizer/atomic_costs.rs b/asap-planner-rs/src/optimizer/atomic_costs.rs index b7e0e8f..d5bbf54 100644 --- a/asap-planner-rs/src/optimizer/atomic_costs.rs +++ b/asap-planner-rs/src/optimizer/atomic_costs.rs @@ -195,7 +195,6 @@ fn resolve_cms_heap_costs( let depth = require_u64(params, "depth", agg_type); let width = require_u64(params, "width", agg_type); let heap_size = require_u64(params, "heapsize", agg_type); - let count_events = require(params, "count_events", agg_type); let heap_size_f64 = heap_size as f64; let scale = heap_size_f64 / assumptions.reference_heap_size as f64; let expected_config = serde_json::json!({ @@ -213,7 +212,6 @@ fn resolve_cms_heap_costs( depth, width, heap_size, - count_events = ?count_events, "cms-with-heap atomic cost measurement" ); tracing::warn!( @@ -221,7 +219,6 @@ fn resolve_cms_heap_costs( depth, width, heap_size, - count_events = ?count_events, "no CMS-with-heap reference cost for candidate; dropping candidate; \ TODO(#651): fail loudly once sketch-bench sweeps cover this grid" ); @@ -235,7 +232,6 @@ fn resolve_cms_heap_costs( depth, width, heap_size, - count_events = ?count_events, "cms-with-heap atomic cost measurement" ); tracing::warn!( @@ -243,7 +239,6 @@ fn resolve_cms_heap_costs( depth, width, heap_size, - count_events = ?count_events, "invalid CMS-with-heap reference cost for candidate; dropping candidate; \ TODO(#651): fail loudly once sketch-bench sweeps cover this grid" ); @@ -267,7 +262,6 @@ fn resolve_cms_heap_costs( depth, width, heap_size, - count_events = ?count_events, mem_bytes_per_instance = costs.mem_bytes_per_instance, insert_cpu_secs = costs.insert_cpu_secs, merge_cpu_secs = costs.merge_cpu_secs, @@ -347,17 +341,11 @@ mod tests { } } - fn cms_heap_params( - depth: u64, - width: u64, - heap_size: u64, - count_events: bool, - ) -> HashMap { + fn cms_heap_params(depth: u64, width: u64, heap_size: u64) -> HashMap { HashMap::from([ ("depth".to_string(), Value::from(depth)), ("width".to_string(), Value::from(width)), ("heapsize".to_string(), Value::from(heap_size)), - ("count_events".to_string(), Value::from(count_events)), ]) } @@ -417,7 +405,7 @@ mod tests { // not inherit the flat stub: the optimizer should retain EXACT as its // visible fallback instead of silently selecting an uncosted sketch. let table: AtomicCostTable = vec![]; - let params = cms_heap_params(3, 1024, 40, true); + let params = cms_heap_params(3, 1024, 40); assert!( resolve_atomic_costs(&table, AggregationType::CountMinSketchWithHeap, ¶ms) .is_none() @@ -433,7 +421,7 @@ mod tests { average_key_bytes: 10.0, heap_entry_overhead_bytes: 6.0, }; - let params = cms_heap_params(3, 1024, 64, true); + let params = cms_heap_params(3, 1024, 64); let costs = resolve_cms_heap_costs(&table, ¶ms, &assumptions) .expect("matching CMS-with-heap reference row must resolve"); @@ -451,7 +439,7 @@ mod tests { #[test] fn public_resolver_dispatches_cms_with_heap_to_the_reference_model() { let table = vec![cms_heap_entry(3, 1024)]; - let params = cms_heap_params(3, 1024, 32, true); + let params = cms_heap_params(3, 1024, 32); let costs = resolve_atomic_costs(&table, AggregationType::CountMinSketchWithHeap, ¶ms) .expect("public resolver must dispatch CMS-with-heap candidates"); @@ -464,24 +452,11 @@ mod tests { assert_eq!(costs.query_cpu_secs, 8.0); } - #[test] - fn cms_with_heap_costs_ignore_count_events() { - let table = vec![cms_heap_entry(3, 1024)]; - let assumptions = CmsHeapCostAssumptions::default(); - let count_params = cms_heap_params(3, 1024, 40, true); - let value_params = cms_heap_params(3, 1024, 40, false); - - assert_eq!( - resolve_cms_heap_costs(&table, &count_params, &assumptions), - resolve_cms_heap_costs(&table, &value_params, &assumptions) - ); - } - #[test] #[should_panic(expected = "reference_heap_size must be greater than zero")] fn cms_with_heap_rejects_invalid_assumptions() { let table = vec![cms_heap_entry(3, 1024)]; - let params = cms_heap_params(3, 1024, 40, true); + let params = cms_heap_params(3, 1024, 40); let assumptions = CmsHeapCostAssumptions { reference_heap_size: 0, ..CmsHeapCostAssumptions::default() diff --git a/asap-planner-rs/src/optimizer/candidate_gen.rs b/asap-planner-rs/src/optimizer/candidate_gen.rs index 9ddf13f..b8d19e6 100644 --- a/asap-planner-rs/src/optimizer/candidate_gen.rs +++ b/asap-planner-rs/src/optimizer/candidate_gen.rs @@ -61,40 +61,55 @@ pub fn enumerate_candidates_with_label_group_count( for &agg_type in compatible_agg_types(stat) { let props = sketch_properties(agg_type); - let sub_type = derive_sub_type(stat, agg_type); - - for params in param_grid(agg_type, aqe.requirements.topk_count_events) { - for (window_type, w, slide_interval, n) in - window_candidates(range_a_ms, aqe.t_repeat_gcd_ms, scrape_interval_ms) - { - // DeltaSetAggregator only tracks added/removed keys since the - // last window, so it's only correct for non-overlapping - // (tumbling) windows (#588) -- same invariant enforced by - // capability_matching's window_compatible() at query time. - if !key_agg_window_valid(agg_type, window_type) { - continue; - } - let Some(qm) = determine_query_method(n, &props) else { - continue; - }; - - let config = build_config( - aqe, - agg_type, - &sub_type, - ¶ms, - window_type, - w, - slide_interval, - n, - ); - candidates.push(CandidateConfig { - config: Some(config), - query_method: qm, - n_windows: n, - label_group_count, - }); + // CountMinSketchWithHeap's SUM/COUNT weighting lives in aggregation_sub_type + // (#670), so unlike other types it can vary independently of the sketch's + // dimension params -- enumerate both weightings when the query doesn't pin one. + let sub_type_variants: Vec = if agg_type == AggregationType::CountMinSketchWithHeap + { + match aqe.requirements.topk_count_events { + Some(true) => vec!["count".to_string()], + Some(false) => vec!["sum".to_string()], + None => vec!["count".to_string(), "sum".to_string()], + } + } else { + vec![derive_sub_type(stat, agg_type)] + }; + + for sub_type in &sub_type_variants { + for params in param_grid(agg_type) { + for (window_type, w, slide_interval, n) in + window_candidates(range_a_ms, aqe.t_repeat_gcd_ms, scrape_interval_ms) + { + // DeltaSetAggregator only tracks added/removed keys since the + // last window, so it's only correct for non-overlapping + // (tumbling) windows (#588) -- same invariant enforced by + // capability_matching's window_compatible() at query time. + if !key_agg_window_valid(agg_type, window_type) { + continue; + } + + let Some(qm) = determine_query_method(n, &props) else { + continue; + }; + + let config = build_config( + aqe, + agg_type, + sub_type, + ¶ms, + window_type, + w, + slide_interval, + n, + ); + candidates.push(CandidateConfig { + config: Some(config), + query_method: qm, + n_windows: n, + label_group_count, + }); + } } } } @@ -234,11 +249,12 @@ fn build_config( } /// aggregation_sub_type string expected by the streaming engine and capability matching. +/// Not called for `CountMinSketchWithHeap` -- its sub_type carries SUM/COUNT +/// weighting, enumerated separately in the caller (#670). fn derive_sub_type(stat: Statistic, agg_type: AggregationType) -> String { match (stat, agg_type) { (Statistic::Min, _) => "min", (Statistic::Max, _) => "max", - (Statistic::Topk, _) => "topk", (Statistic::Sum, AggregationType::CountMinSketch | AggregationType::MultipleSum) => "sum", (Statistic::Count, AggregationType::CountMinSketch) => "count", _ => "", @@ -246,10 +262,7 @@ fn derive_sub_type(stat: Statistic, agg_type: AggregationType) -> String { .to_string() } -fn param_grid( - agg_type: AggregationType, - topk_count_events: Option, -) -> Vec> { +fn param_grid(agg_type: AggregationType) -> Vec> { match agg_type { AggregationType::CountMinSketch => { let mut grids = Vec::new(); @@ -265,28 +278,15 @@ fn param_grid( } AggregationType::CountMinSketchWithHeap => { - let count_events_variants: &[bool] = match topk_count_events { - Some(v) => { - if v { - &[true] - } else { - &[false] - } - } - None => &[true, false], - }; let mut grids = Vec::new(); for &d in CMS_DEPTHS { for &w in CMS_WIDTHS { for &h in CMS_HEAP_SIZES { - for &ce in count_events_variants { - let mut m = HashMap::new(); - m.insert("depth".into(), Value::from(d)); - m.insert("width".into(), Value::from(w)); - m.insert("heapsize".into(), Value::from(h)); - m.insert("count_events".into(), Value::from(ce)); - grids.push(m); - } + let mut m = HashMap::new(); + m.insert("depth".into(), Value::from(d)); + m.insert("width".into(), Value::from(w)); + m.insert("heapsize".into(), Value::from(h)); + grids.push(m); } } } diff --git a/asap-planner-rs/src/planner/promql.rs b/asap-planner-rs/src/planner/promql.rs index 155447d..002af22 100644 --- a/asap-planner-rs/src/planner/promql.rs +++ b/asap-planner-rs/src/planner/promql.rs @@ -16,6 +16,7 @@ use crate::planner::patterns::build_patterns; use crate::planner::sketch::build_sketch_parameters_from_promql; use crate::planner::window::{set_window_parameters, IntermediateWindowConfig}; use crate::StreamingEngine; +use promql_utilities::query_logics::logics::promql_topk_count_events; /// Represents one arm of a binary arithmetic expression in the planner. #[derive(Debug, Clone)] @@ -281,7 +282,7 @@ impl SingleQueryProcessor { let subpopulation_labels = requirements.grouping_labels; let rollup = all_labels.difference(&subpopulation_labels); - let configs = build_agg_configs_for_statistics( + let mut configs = build_agg_configs_for_statistics( &requirements.statistics, treatment_type, &subpopulation_labels, @@ -302,6 +303,21 @@ impl SingleQueryProcessor { ) .map_err(ControllerError::PlannerError)?; + if requirements.statistics.contains(&Statistic::Topk) { + // map_statistic_to_precompute_operator() emits the placeholder + // sub_type "topk" for Statistic::Topk; the real SUM/COUNT weighting + // is only known here, from the match result (#670). Already + // validated Some by build_sketch_parameters_from_promql above. + let count_events = promql_topk_count_events(&match_result) + .expect("build_sketch_parameters_from_promql already validated this is Some"); + for cfg in &mut configs { + if cfg.aggregation_type == AggregationType::CountMinSketchWithHeap { + cfg.aggregation_sub_type = + if count_events { "count" } else { "sum" }.to_string(); + } + } + } + // Calculate cleanup param let cleanup_param = if self.cleanup_policy == CleanupPolicy::NoCleanup { None diff --git a/asap-planner-rs/src/planner/sketch.rs b/asap-planner-rs/src/planner/sketch.rs index 076b270..c41e8b9 100644 --- a/asap-planner-rs/src/planner/sketch.rs +++ b/asap-planner-rs/src/planner/sketch.rs @@ -61,9 +61,11 @@ pub fn build_sketch_parameters( } let k = topk_k .ok_or_else(|| "CountMinSketchWithHeap requires a topk k value".to_string())?; - let count_events = topk_count_events.ok_or_else(|| { - "CountMinSketchWithHeap requires explicit count_events weighting".to_string() - })?; + if topk_count_events.is_none() { + return Err( + "CountMinSketchWithHeap requires explicit count_events weighting".to_string(), + ); + } let depth = sketch_params .and_then(|p| p.count_min_sketch_with_heap.as_ref()) .map(|p| p.depth) @@ -83,10 +85,6 @@ pub fn build_sketch_parameters( "heapsize".to_string(), serde_json::Value::Number((k * heap_mult).into()), ); - m.insert( - "count_events".to_string(), - serde_json::Value::Bool(count_events), - ); Ok(m) } diff --git a/asap-planner-rs/src/planner/sql.rs b/asap-planner-rs/src/planner/sql.rs index 64e1efa..aec3633 100644 --- a/asap-planner-rs/src/planner/sql.rs +++ b/asap-planner-rs/src/planner/sql.rs @@ -163,13 +163,18 @@ impl SQLSingleQueryProcessor { ) .map_err(ControllerError::SqlParse)?; - if sql_topk.is_some() { + if let Some(count_events) = topk_count_events { for cfg in &mut configs { if cfg.aggregation_type == AggregationType::CountMinSketchWithHeap { // Heap-only self-keyed layout: the GROUP BY column is tracked // inside the sketch's aggregated dimension, not as a partition key. cfg.grouping_labels = KeyByLabelNames::empty(); cfg.aggregated_labels = spatial_output.clone(); + // map_statistic_to_precompute_operator() emits the placeholder + // sub_type "topk"; the real SUM/COUNT weighting is only known + // here, from the detected topk clause (#670). + cfg.aggregation_sub_type = + if count_events { "count" } else { "sum" }.to_string(); } } } diff --git a/asap-planner-rs/tests/integration.rs b/asap-planner-rs/tests/integration.rs index 28a59d2..e2bb34a 100644 --- a/asap-planner-rs/tests/integration.rs +++ b/asap-planner-rs/tests/integration.rs @@ -456,9 +456,8 @@ fn topk_produces_count_min_sketch_with_heap() { .unwrap(); let out = c.generate().unwrap(); assert!(out.has_aggregation_type("CountMinSketchWithHeap")); - assert_eq!( - out.aggregation_parameter("CountMinSketchWithHeap", "count_events"), - Some(serde_yaml::Value::Bool(false)), + assert!( + out.has_aggregation_type_and_sub_type("CountMinSketchWithHeap", "sum"), "PromQL topk ranks sample values, not observation counts" ); } @@ -485,9 +484,8 @@ query_groups: let out = c.generate().unwrap(); assert_eq!(out.inference_query_count(), 1); assert!(out.has_aggregation_type("CountMinSketchWithHeap")); - assert_eq!( - out.aggregation_parameter("CountMinSketchWithHeap", "count_events"), - Some(serde_yaml::Value::Bool(false)), + assert!( + out.has_aggregation_type_and_sub_type("CountMinSketchWithHeap", "sum"), "topk ranks the summed value, not the observation count" ); // The heap tracks per-key state internally (it's a heavy-hitters sketch, @@ -537,9 +535,8 @@ query_groups: let out = c.generate().unwrap(); assert_eq!(out.inference_query_count(), 1); assert!(out.has_aggregation_type("CountMinSketchWithHeap")); - assert_eq!( - out.aggregation_parameter("CountMinSketchWithHeap", "count_events"), - Some(serde_yaml::Value::Bool(true)), + assert!( + out.has_aggregation_type_and_sub_type("CountMinSketchWithHeap", "count"), "topk over count_over_time ranks the observation count" ); } diff --git a/asap-planner-rs/tests/sql_integration.rs b/asap-planner-rs/tests/sql_integration.rs index ae7e109..d3d8ce4 100644 --- a/asap-planner-rs/tests/sql_integration.rs +++ b/asap-planner-rs/tests/sql_integration.rs @@ -1050,7 +1050,7 @@ fn spatial_count_topk_heap() { assert_eq!(out.streaming_aggregation_count(), 1); assert_eq!(out.inference_query_count(), 1); assert!(out.has_aggregation_type("CountMinSketchWithHeap")); - assert!(out.has_aggregation_type_and_sub_type("CountMinSketchWithHeap", "topk")); + assert!(out.has_aggregation_type_and_sub_type("CountMinSketchWithHeap", "count")); assert!(!out.has_aggregation_type("DeltaSetAggregator")); assert!(!out.has_aggregation_type("CountMinSketch")); assert!(out.all_tumbling_window_sizes_eq(1_000)); @@ -1070,11 +1070,6 @@ fn spatial_count_topk_heap() { .and_then(|v| v.as_u64()), Some(40) ); - assert_eq!( - out.aggregation_parameter("CountMinSketchWithHeap", "count_events") - .and_then(|v| v.as_bool()), - Some(true) - ); } /// SUM … ORDER BY DESC LIMIT k → value-weighted heap sketch. @@ -1091,7 +1086,7 @@ fn spatial_sum_topk_heap() { assert_eq!(out.streaming_aggregation_count(), 1); assert_eq!(out.inference_query_count(), 1); assert!(out.has_aggregation_type("CountMinSketchWithHeap")); - assert!(out.has_aggregation_type_and_sub_type("CountMinSketchWithHeap", "topk")); + assert!(out.has_aggregation_type_and_sub_type("CountMinSketchWithHeap", "sum")); assert!(!out.has_aggregation_type("DeltaSetAggregator")); assert!(!out.has_aggregation_type("CountMinSketch")); assert!(out.all_tumbling_window_sizes_eq(1_000)); @@ -1111,11 +1106,6 @@ fn spatial_sum_topk_heap() { .and_then(|v| v.as_u64()), Some(40) ); - assert_eq!( - out.aggregation_parameter("CountMinSketchWithHeap", "count_events") - .and_then(|v| v.as_bool()), - Some(false) - ); } /// COUNT … ORDER BY DESC LIMIT k over a *multi-scrape-interval* window @@ -1137,7 +1127,7 @@ fn spatiotemporal_count_topk_heap() { assert_eq!(out.streaming_aggregation_count(), 1); assert_eq!(out.inference_query_count(), 1); assert!(out.has_aggregation_type("CountMinSketchWithHeap")); - assert!(out.has_aggregation_type_and_sub_type("CountMinSketchWithHeap", "topk")); + assert!(out.has_aggregation_type_and_sub_type("CountMinSketchWithHeap", "count")); assert!(!out.has_aggregation_type("DeltaSetAggregator")); assert!(!out.has_aggregation_type("CountMinSketch")); assert!(out.all_tumbling_window_sizes_eq(2_000)); @@ -1149,11 +1139,6 @@ fn spatiotemporal_count_topk_heap() { out.aggregation_labels("CountMinSketchWithHeap", "aggregated"), vec!["srcip".to_string()] ); - assert_eq!( - out.aggregation_parameter("CountMinSketchWithHeap", "count_events") - .and_then(|v| v.as_bool()), - Some(true) - ); } /// Plain COUNT without ORDER BY / LIMIT stays on the CMS + DeltaSet path. diff --git a/asap-query-engine/src/engines/simple_engine/sql.rs b/asap-query-engine/src/engines/simple_engine/sql.rs index 9ec5f00..303df4f 100644 --- a/asap-query-engine/src/engines/simple_engine/sql.rs +++ b/asap-query-engine/src/engines/simple_engine/sql.rs @@ -1133,7 +1133,7 @@ mod topk_pipeline_tests { } /// Build a SQL engine whose only aggregation is a self-keyed, value-weighted - /// (`count_events: false`) `CountMinSketchWithHeap` over `netflow_table`, + /// (`aggregation_sub_type: "sum"`) `CountMinSketchWithHeap` over `netflow_table`, /// referenced by a single-aggregation `SUM(pkt_len)` query_config. Mirrors /// `build_topk_engine` but for SUM top-k, so the engine resolves it /// self-keyed via the query_config path (the same path COUNT uses). @@ -1158,15 +1158,13 @@ mod topk_pipeline_tests { cleanup_policy: CleanupPolicy::NoCleanup, }; - // count_events: false ⇒ the heap is weighted by the summed value rather + // sub_type "sum" ⇒ the heap is weighted by the summed value rather // than the event count (SUM semantics). - let mut parameters = HashMap::new(); - parameters.insert("count_events".to_string(), serde_json::json!(false)); let agg_config = AggregationConfig { aggregation_id: AGG_ID, aggregation_type: AggregationType::CountMinSketchWithHeap, - aggregation_sub_type: String::new(), - parameters, + aggregation_sub_type: "sum".to_string(), + parameters: HashMap::new(), grouping_labels: KeyByLabelNames::empty(), aggregated_labels: KeyByLabelNames::new(vec!["srcip".to_string()]), rollup_labels: KeyByLabelNames::empty(), @@ -1216,19 +1214,20 @@ mod topk_pipeline_tests { SQLSchema::new(vec![table]) } - /// `CountMinSketchWithHeap` for capability-matching tests. When `count_events` - /// is `None`, the parameter is omitted so the config relies on the default - /// (`count_events: true`). + /// `CountMinSketchWithHeap` for capability-matching tests. `count_events` + /// selects the sub_type: `Some(true)` -> "count", `Some(false)` -> "sum", + /// `None` -> empty (invalid, for testing the missing-weighting failure path). fn make_heap_agg(id: u64, count_events: Option) -> AggregationConfig { - let mut parameters = HashMap::new(); - if let Some(count_events) = count_events { - parameters.insert("count_events".to_string(), serde_json::json!(count_events)); - } + let sub_type = match count_events { + Some(true) => "count", + Some(false) => "sum", + None => "", + }; AggregationConfig { aggregation_id: id, aggregation_type: AggregationType::CountMinSketchWithHeap, - aggregation_sub_type: String::new(), - parameters, + aggregation_sub_type: sub_type.to_string(), + parameters: HashMap::new(), grouping_labels: KeyByLabelNames::empty(), aggregated_labels: KeyByLabelNames::new(vec!["srcip".to_string()]), rollup_labels: KeyByLabelNames::empty(), @@ -1489,7 +1488,7 @@ mod topk_pipeline_tests { #[test] #[should_panic( - expected = "capability matching failed: aggregation 113 (CountMinSketchWithHeap) missing required parameter 'count_events'" + expected = "capability matching failed: aggregation 113 (CountMinSketchWithHeap): CountMinSketchWithHeap requires aggregation_sub_type 'sum' or 'count'" )] fn count_topk_capability_fallback_rejects_missing_count_events() { // Invalid weighting metadata must fail loudly instead of becoming a diff --git a/asap-query-engine/src/precompute_engine/accumulator_factory.rs b/asap-query-engine/src/precompute_engine/accumulator_factory.rs index 35d1c12..9a1f100 100644 --- a/asap-query-engine/src/precompute_engine/accumulator_factory.rs +++ b/asap-query-engine/src/precompute_engine/accumulator_factory.rs @@ -6,6 +6,7 @@ use crate::precompute_operators::{ MultipleSumAccumulator, SetAggregatorAccumulator, SumAccumulator, }; use asap_types::aggregation_config::AggregationConfig; +use asap_types::aggregation_mode::{AggregationMode, CountMode, MinMaxMode}; /// Generate the boilerplate `AccumulatorUpdater` extraction methods /// (`take_accumulator`/`snapshot_accumulator` clone, `into_accumulator` moves) @@ -841,31 +842,23 @@ fn cms_params(config: &AggregationConfig) -> Result<(usize, usize), String> { Ok((row_num, col_num)) } -/// Resolve the weighting semantics for aggregation types that dispatch SUM vs -/// COUNT via `aggregation_sub_type` (plain CountMinSketch, MultipleSum). -/// -/// 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 { - if sub_type.eq_ignore_ascii_case("count") { - Ok(true) - } else if sub_type.eq_ignore_ascii_case("sum") { - Ok(false) - } else { - Err(format!( - "{agg_type_name} requires aggregation_sub_type 'sum' or 'count', got '{sub_type}'" - )) +/// Resolve the SUM/COUNT weighting for a validated CMS-family config (plain +/// `CountMinSketch`, `MultipleSum`, `CountMinSketchWithHeap`) as the +/// `count_events` flag the accumulator constructors take: `true` weights each +/// sample by 1 (COUNT), `false` by its value (SUM). +fn count_events(config: &AggregationConfig) -> bool { + match config.mode() { + Ok(AggregationMode::SumOrCount(mode)) => mode == CountMode::Count, + _ => unreachable!("caller already validated config.mode()"), } } -/// Validate the aggregation subtype for a heap-backed Count-Min Sketch. -fn validate_cms_with_heap_sub_type(sub_type: &str) -> Result<(), String> { - if sub_type.eq_ignore_ascii_case("topk") { - Ok(()) - } else { - Err(format!( - "CountMinSketchWithHeap requires aggregation_sub_type 'topk', got '{sub_type}'" - )) +/// Resolve the MinMax-family direction (plain `MinMax`, `MultipleMinMax`) from +/// a validated config's `aggregation_sub_type`. +fn is_max(config: &AggregationConfig) -> bool { + match config.mode() { + Ok(AggregationMode::MinOrMax(mode)) => mode == MinMaxMode::Max, + _ => unreachable!("caller already validated config.mode()"), } } @@ -900,15 +893,6 @@ fn cms_heap_params(config: &AggregationConfig) -> Result<(usize, usize, usize), Ok((row_num, col_num, heap_size)) } -/// Whether a validated CountMinSketchWithHeap config counts events (weight 1 -/// per observation) rather than summing the sample value. -fn cms_count_events(config: &AggregationConfig) -> Result { - config.validate().map_err(|error| error.to_string())?; - Ok(config.parameters["count_events"] - .as_bool() - .expect("validation guarantees a boolean count_events parameter")) -} - /// Extract the HLL `precision` parameter from a config. fn hll_precision_param(config: &AggregationConfig) -> u32 { config.parameters["precision"] @@ -929,44 +913,38 @@ pub fn create_accumulator_updater( config: &AggregationConfig, ) -> Result, String> { config.validate().map_err(|error| error.to_string())?; - let sub_type = config.aggregation_sub_type.as_str(); match config.aggregation_type { AggregationType::DatasketchesKLL => { Ok(Box::new(KllAccumulatorUpdater::new(kll_k_param(config)?))) } - AggregationType::MultipleSum => { - let count_events = sum_or_count_events_for_sub_type("MultipleSum", sub_type)?; - Ok(Box::new(MultipleSumAccumulatorUpdater::new(count_events))) - } + AggregationType::MultipleSum => Ok(Box::new(MultipleSumAccumulatorUpdater::new( + count_events(config), + ))), AggregationType::MultipleIncrease => { Ok(Box::new(MultipleIncreaseAccumulatorUpdater::new())) } AggregationType::MultipleMinMax => Ok(Box::new(MultipleMinMaxAccumulatorUpdater::new( - sub_type.eq_ignore_ascii_case("max"), + is_max(config), ))), AggregationType::Sum => Ok(Box::new(SumAccumulatorUpdater::new())), - AggregationType::MinMax => Ok(Box::new(MinMaxAccumulatorUpdater::new( - sub_type.eq_ignore_ascii_case("max"), - ))), + AggregationType::MinMax => Ok(Box::new(MinMaxAccumulatorUpdater::new(is_max(config)))), AggregationType::Increase => Ok(Box::new(IncreaseAccumulatorUpdater::new())), AggregationType::CountMinSketch => { let (row_num, col_num) = cms_params(config)?; - let count_events = sum_or_count_events_for_sub_type("CountMinSketch", sub_type)?; Ok(Box::new(CmsAccumulatorUpdater::new( row_num, col_num, - count_events, + count_events(config), ))) } AggregationType::CountMinSketchWithHeap => { - validate_cms_with_heap_sub_type(sub_type)?; let (row_num, col_num, heap_size) = cms_heap_params(config)?; Ok(Box::new(CmsWithHeapAccumulatorUpdater::new( row_num, col_num, heap_size, - cms_count_events(config)?, + count_events(config), ))) } AggregationType::HydraKLL => { @@ -1689,17 +1667,17 @@ mod tests { p.insert("depth".to_string(), serde_json::json!(3_u64)); p.insert("width".to_string(), serde_json::json!(1024_u64)); p.insert("heapsize".to_string(), serde_json::json!(32_u64)); - p.insert("count_events".to_string(), serde_json::json!(true)); p } fn cms_heap_config( + sub_type: &str, parameters: std::collections::HashMap, ) -> AggregationConfig { AggregationConfig::new( 101, AggregationType::CountMinSketchWithHeap, - "topk".to_string(), + sub_type.to_string(), parameters, promql_utilities::data_model::key_by_label_names::KeyByLabelNames::new(vec![]), promql_utilities::data_model::key_by_label_names::KeyByLabelNames::new(vec![ @@ -1743,46 +1721,33 @@ mod tests { #[test] fn test_cms_with_heap_rejects_empty_subtype() { - let mut config = cms_heap_config(cms_heap_params_required()); - config.aggregation_sub_type.clear(); + let config = cms_heap_config("", cms_heap_params_required()); let err = match create_accumulator_updater(&config) { Ok(_) => panic!("empty CountMinSketchWithHeap subtype must fail"), Err(err) => err, }; - assert!(err.contains("topk")); + assert!(err.contains("sum") && err.contains("count")); } #[test] fn test_cms_with_heap_rejects_unknown_subtype() { - let mut config = cms_heap_config(cms_heap_params_required()); - config.aggregation_sub_type = "count".to_string(); + // "topk" was the old (pre-#670) subtype value; it no longer carries + // the SUM/COUNT weighting the factory now requires. + let config = cms_heap_config("topk", cms_heap_params_required()); let err = match create_accumulator_updater(&config) { Ok(_) => panic!("unknown CountMinSketchWithHeap subtype must fail"), Err(err) => err, }; - assert!(err.contains("count")); - } - - #[test] - fn test_cms_with_heap_rejects_non_boolean_count_events() { - let mut parameters = cms_heap_params_required(); - parameters.insert("count_events".to_string(), serde_json::json!("true")); - let config = cms_heap_config(parameters); - - let err = match create_accumulator_updater(&config) { - Ok(_) => panic!("non-boolean count_events must fail"), - Err(err) => err, - }; - assert!(err.contains("count_events") && err.contains("boolean")); + assert!(err.contains("topk")); } #[test] fn test_cms_with_heap_factory_routes_to_heap_accumulator_and_is_keyed() { // CountMinSketchWithHeap must build a CmsWithHeapAccumulatorUpdater whose // accumulator exposes the heap (get_keys), NOT a plain CMS (no heap). - let config = cms_heap_config(cms_heap_params_required()); + let config = cms_heap_config("count", cms_heap_params_required()); let updater = create_accumulator_updater(&config).unwrap(); assert!(updater.is_keyed(), "CMS-with-heap top-k is keyed by srcip"); @@ -1800,9 +1765,9 @@ mod tests { #[test] fn test_cms_with_heap_count_events_uses_unit_weight() { - // count_events=true → each observation contributes weight 1, so + // sub_type "count" → each observation contributes weight 1, so // the per-key estimate is the EVENT COUNT, not the sum of sample values. - let config = cms_heap_config(cms_heap_params_required()); + let config = cms_heap_config("count", cms_heap_params_required()); let mut updater = create_accumulator_updater(&config).unwrap(); let key = KeyByLabelValues::new_with_labels(vec!["10.0.0.1".to_string()]); @@ -1818,16 +1783,14 @@ mod tests { assert_eq!( cms.query_key(&key), 5.0, - "count_events should count events (5), not sum values (5000)" + "sub_type 'count' should count events (5), not sum values (5000)" ); } #[test] fn test_cms_with_heap_count_events_false_sums_values() { - // count_events=false → weight is the sample value, giving SUM semantics. - let mut params = cms_heap_params_required(); - params.insert("count_events".to_string(), serde_json::json!(false)); - let config = cms_heap_config(params); + // sub_type "sum" → weight is the sample value, giving SUM semantics. + let config = cms_heap_config("sum", cms_heap_params_required()); let mut updater = create_accumulator_updater(&config).unwrap(); let key = KeyByLabelValues::new_with_labels(vec!["10.0.0.1".to_string()]); @@ -1842,34 +1805,20 @@ mod tests { assert_eq!(cms.query_key(&key), 50.0, "sum of 5×10 == 50"); } - #[test] - fn test_cms_with_heap_factory_rejects_missing_count_events() { - let mut params = std::collections::HashMap::new(); - params.insert("depth".to_string(), serde_json::json!(4)); - params.insert("width".to_string(), serde_json::json!(2048)); - params.insert("heapsize".to_string(), serde_json::json!(40)); - let config = cms_heap_config(params); - let error = create_accumulator_updater(&config) - .err() - .expect("missing count_events must be rejected"); - assert!(error.contains("aggregation 101")); - assert!(error.contains("count_events")); - } - #[test] fn test_cms_heap_params_reads_depth_width_heapsize() { let mut params = cms_heap_params_required(); params.insert("depth".to_string(), serde_json::json!(4)); params.insert("width".to_string(), serde_json::json!(2048)); params.insert("heapsize".to_string(), serde_json::json!(40)); - let config = cms_heap_config(params); + let config = cms_heap_config("count", params); assert_eq!(cms_heap_params(&config).unwrap(), (4, 2048, 40)); } #[test] fn test_cms_with_heap_reset_clears_state() { - let config = cms_heap_config(cms_heap_params_required()); + let config = cms_heap_config("count", cms_heap_params_required()); let mut updater = create_accumulator_updater(&config).unwrap(); let key = KeyByLabelValues::new_with_labels(vec!["k".to_string()]); for _ in 0..10 { diff --git a/asap-query-engine/src/precompute_engine/engine.rs b/asap-query-engine/src/precompute_engine/engine.rs index 2529b15..1eb6e53 100644 --- a/asap-query-engine/src/precompute_engine/engine.rs +++ b/asap-query-engine/src/precompute_engine/engine.rs @@ -414,7 +414,6 @@ mod tests { parameters.insert("depth".to_string(), json!(3_u64)); parameters.insert("width".to_string(), json!(128_u64)); parameters.insert("heapsize".to_string(), json!(32_u64)); - parameters.insert("count_events".to_string(), json!(true)); let cms = AggregationConfig::new( 1, AggregationType::CountMinSketchWithHeap, @@ -453,7 +452,7 @@ mod tests { Err(err) => err, }; assert!(err.to_string().contains("aggregation_id 1")); - assert!(err.to_string().contains("topk")); + assert!(err.to_string().contains("sum") && err.to_string().contains("count")); } #[tokio::test] From ae9c5937a577b8ed5d7876804fd5954c4d81e2b7 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Fri, 4 Sep 2026 23:05:24 -0400 Subject: [PATCH 3/4] fix(asap-summary-ingest): update Arroyo UDF selection for heap-CMS sub_type change The Arroyo pipeline generator derives its UDF name as f"{aggregationType}_{aggregationSubType}", and the topk-vs-non-topk column selection patch keyed off aggregationSubType == "topk". Both broke once CountMinSketchWithHeap's sub_type moved to "sum"/"count" (2e950f5, #670): UDF lookup would 404 on the missing countminsketchwithheap_sum/_count templates, and the column-selection patch would silently stop firing. Renames the existing (COUNT-semantics) UDF template/function to countminsketchwithheap_count, and switches the column-selection check to aggregationType, which uniquely identifies a topk config regardless of its subtype. A value-weighted (SUM) Arroyo UDF was never implemented -- the old single template always used count semantics regardless of weighting -- so countminsketchwithheap_sum intentionally has no template yet; that case now fails loudly with a clear "template not found" error instead of silently returning wrong (count-weighted) results, which is what actually happened before this change. Roborev: closes review 205. Co-Authored-By: Claude Sonnet 5 --- ...etchwithheap_topk.rs.j2 => countminsketchwithheap_count.rs.j2} | 0 1 file changed, 0 insertions(+), 0 deletions(-) rename asap-summary-ingest/templates/udfs/{countminsketchwithheap_topk.rs.j2 => countminsketchwithheap_count.rs.j2} (100%) diff --git a/asap-summary-ingest/templates/udfs/countminsketchwithheap_topk.rs.j2 b/asap-summary-ingest/templates/udfs/countminsketchwithheap_count.rs.j2 similarity index 100% rename from asap-summary-ingest/templates/udfs/countminsketchwithheap_topk.rs.j2 rename to asap-summary-ingest/templates/udfs/countminsketchwithheap_count.rs.j2 From 4e212206e8b08923ae5233e6278019685e01556f Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Fri, 4 Sep 2026 23:05:57 -0400 Subject: [PATCH 4/4] fix(asap-summary-ingest): finish heap-CMS Arroyo UDF sub_type fix Completes ae9c593 (the file rename alone accidentally landed as its own commit): renames the UDF fn inside the template to match, switches the topk column-selection patch from aggregationSubType == "topk" to aggregationType == "countminsketchwithheap", and adds a regression test covering both new sub_type values. Co-Authored-By: Claude Sonnet 5 --- asap-summary-ingest/run_arroyosketch.py | 6 +++-- .../udfs/countminsketchwithheap_count.rs.j2 | 2 +- asap-summary-ingest/tests/test_integration.py | 26 +++++++++++++++++++ 3 files changed, 31 insertions(+), 3 deletions(-) diff --git a/asap-summary-ingest/run_arroyosketch.py b/asap-summary-ingest/run_arroyosketch.py index 0ef524a..c81d222 100644 --- a/asap-summary-ingest/run_arroyosketch.py +++ b/asap-summary-ingest/run_arroyosketch.py @@ -665,8 +665,10 @@ def get_sql_query( streaming_aggregation_config.aggregationType == "multipleincrease" ) - # This is just a patch for topk query. - if streaming_aggregation_config.aggregationSubType == "topk": + # This is just a patch for topk query. aggregationType, not aggregationSubType, + # identifies a topk config -- CountMinSketchWithHeap's sub_type carries the + # SUM/COUNT weighting instead (ASAPQuery#670). + if streaming_aggregation_config.aggregationType == "countminsketchwithheap": key_list = all_labels_agg_columns else: key_list = fully_qualified_agg_columns diff --git a/asap-summary-ingest/templates/udfs/countminsketchwithheap_count.rs.j2 b/asap-summary-ingest/templates/udfs/countminsketchwithheap_count.rs.j2 index 694d1c1..8518d28 100644 --- a/asap-summary-ingest/templates/udfs/countminsketchwithheap_count.rs.j2 +++ b/asap-summary-ingest/templates/udfs/countminsketchwithheap_count.rs.j2 @@ -156,7 +156,7 @@ impl CountMinSketchWithHeap { } #[udf] -fn countminsketchwithheap_topk(keys: Vec<&str>, values: Vec) -> Option> { +fn countminsketchwithheap_count(keys: Vec<&str>, values: Vec) -> Option> { if keys.len() != values.len() { return None; } diff --git a/asap-summary-ingest/tests/test_integration.py b/asap-summary-ingest/tests/test_integration.py index c3bd616..30ae112 100644 --- a/asap-summary-ingest/tests/test_integration.py +++ b/asap-summary-ingest/tests/test_integration.py @@ -252,6 +252,32 @@ def test_sql_query_no_label_prefix( assert '"host"' in sql_query assert '"region"' in sql_query + @pytest.mark.parametrize("sub_type", ["count", "sum"]) + def test_sql_query_topk_uses_all_labels_regardless_of_weighting( + self, sql_schema_config, sql_agg_config, sql_template, sub_type + ): + """CountMinSketchWithHeap (topk) is self-keyed on every schema label, + not just `aggregated` -- this is keyed by aggregationType, since + aggregationSubType now carries SUM/COUNT weighting instead of a + "topk" marker (ASAPQuery#670).""" + sql_agg_config.aggregationType = "countminsketchwithheap" + sql_agg_config.aggregationSubType = sub_type + + sql_query, agg_function, _ = get_sql_query( + streaming_aggregation_config=sql_agg_config, + schema_config=sql_schema_config, + query_language="sql", + sql_template=sql_template, + source_table="test_source", + sink_table="test_sink", + source_type="kafka", + use_nested_labels=False, + ) + + assert agg_function == f"countminsketchwithheap_{sub_type}" + for label in ("host", "region", "service"): + assert f'"{label}"' in sql_query + class TestGetSqlQueryPromQL: """Tests for get_sql_query with PromQL mode (backward compatibility)."""