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 2a6f3de..215e5be 100644 --- a/asap-common/dependencies/rs/asap_types/src/capability_matching.rs +++ b/asap-common/dependencies/rs/asap_types/src/capability_matching.rs @@ -18,7 +18,11 @@ use promql_utilities::query_logics::enums::AggregationType; /// Returns the aggregation types that can serve this statistic. pub fn compatible_agg_types(stat: Statistic) -> &'static [AggregationType] { match stat { - Statistic::Sum => &[AggregationType::Sum, AggregationType::MultipleSum], + Statistic::Sum => &[ + AggregationType::Sum, + AggregationType::MultipleSum, + AggregationType::CountMinSketch, + ], Statistic::Count => &[ AggregationType::CountMinSketch, AggregationType::CountMinSketchWithHeap, @@ -187,6 +191,23 @@ pub fn topk_weighting_compatible( } } +/// 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 { + return true; + } + + let expected_sub_type = match stat { + Statistic::Sum => "sum", + Statistic::Count => "count", + _ => unreachable!("plain CMS matching only supports SUM and COUNT"), + }; + config + .aggregation_sub_type + .eq_ignore_ascii_case(expected_sub_type) +} + /// Aggregation priority comparator: prefer larger `window_size_ms` (descending). /// This is a separate function so callers can swap the policy without touching matching logic. pub fn aggregation_priority(a: &AggregationConfig, b: &AggregationConfig) -> Ordering { @@ -244,6 +265,7 @@ 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); if !ok { debug!( @@ -466,6 +488,88 @@ mod tests { assert_eq!(result.unwrap().aggregation_id_for_value, 1); } + #[test] + fn plain_cms_matching_respects_sum_and_count_subtypes() { + let mut configs = HashMap::new(); + configs.insert( + 1, + make_config( + 1, + "cpu", + "CountMinSketch", + "sum", + 300_000, + "tumbling", + &[], + "", + ), + ); + configs.insert( + 2, + make_config( + 2, + "cpu", + "CountMinSketch", + "count", + 300_000, + "tumbling", + &[], + "", + ), + ); + configs.insert( + 9, + make_config( + 9, + "cpu", + "DeltaSetAggregator", + "", + 300_000, + "tumbling", + &[], + "", + ), + ); + + let sum = + find_compatible_aggregation(&configs, &req("cpu", &[Statistic::Sum], 300_000, &[], "")) + .expect("SUM should select the value-weighted sketch"); + assert_eq!(sum.aggregation_id_for_value, 1); + + let count = find_compatible_aggregation( + &configs, + &req("cpu", &[Statistic::Count], 300_000, &[], ""), + ) + .expect("COUNT should select the event-weighted sketch"); + assert_eq!(count.aggregation_id_for_value, 2); + } + + #[test] + fn plain_cms_with_invalid_subtypes_is_excluded_from_matching() { + for invalid_sub_type in ["", "unknown", " sum "] { + let configs = single_config(make_config( + 1, + "cpu", + "CountMinSketch", + invalid_sub_type, + 300_000, + "tumbling", + &[], + "", + )); + + let result = 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" + ); + } + } + #[test] fn quantile_any_value_finds_kll() { let configs = single_config(make_config( @@ -1076,7 +1180,16 @@ mod tests { #[test] fn multi_pop_rejects_tumbling_delta_set_that_cannot_partition_sliding_value_window() { - let mut value = make_config(10, "req", "CountMinSketch", "", 6_000, "sliding", &[], ""); + let mut value = make_config( + 10, + "req", + "CountMinSketch", + "count", + 6_000, + "sliding", + &[], + "", + ); value.slide_interval_ms = 1_000; let delta_keys = make_config( 11, @@ -1102,7 +1215,16 @@ mod tests { #[test] fn multi_pop_accepts_tumbling_delta_set_that_partitions_sliding_value_grid() { - let mut value = make_config(10, "req", "CountMinSketch", "", 6_000, "sliding", &[], ""); + let mut value = make_config( + 10, + "req", + "CountMinSketch", + "count", + 6_000, + "sliding", + &[], + "", + ); value.slide_interval_ms = 1_000; let delta_keys = make_config( 11, @@ -1127,7 +1249,16 @@ mod tests { #[test] fn multi_pop_rejects_tumbling_set_key_on_mismatched_nonzero_grid_step() { - let value = make_config(10, "req", "CountMinSketch", "", 5_000, "tumbling", &[], ""); + let value = make_config( + 10, + "req", + "CountMinSketch", + "count", + 5_000, + "tumbling", + &[], + "", + ); let mut keys = make_config(11, "req", "SetAggregator", "", 5_000, "tumbling", &[], ""); keys.slide_interval_ms = 1_000; let configs = HashMap::from([(10, value), (11, keys)]); @@ -1141,7 +1272,16 @@ mod tests { #[test] fn tumbling_set_pairing_normalizes_zero_slide_to_window_size() { - let mut value = make_config(10, "req", "CountMinSketch", "", 5_000, "tumbling", &[], ""); + let mut value = make_config( + 10, + "req", + "CountMinSketch", + "count", + 5_000, + "tumbling", + &[], + "", + ); let mut key = make_config(11, "req", "SetAggregator", "", 5_000, "tumbling", &[], ""); value.slide_interval_ms = 0; key.slide_interval_ms = 0; @@ -1152,7 +1292,16 @@ mod tests { #[test] fn set_pairing_rejects_each_grid_mismatch_dimension() { - let value = make_config(10, "req", "CountMinSketch", "", 5_000, "sliding", &[], ""); + let value = make_config( + 10, + "req", + "CountMinSketch", + "count", + 5_000, + "sliding", + &[], + "", + ); let mut key = make_config(11, "req", "SetAggregator", "", 5_000, "sliding", &[], ""); key.slide_interval_ms = 1_000; assert!(!key_agg_compatible_with_value(&value, &key)); @@ -1166,7 +1315,16 @@ mod tests { #[test] fn delta_set_pairing_truth_table_checks_both_divisors() { - let mut value = make_config(10, "req", "CountMinSketch", "", 6_000, "sliding", &[], ""); + let mut value = make_config( + 10, + "req", + "CountMinSketch", + "count", + 6_000, + "sliding", + &[], + "", + ); value.slide_interval_ms = 2_000; let key_valid = make_config( 11, @@ -1212,7 +1370,16 @@ mod tests { #[test] fn delta_set_pairing_for_tumbling_values_does_not_apply_sliding_rules() { - let value = make_config(10, "req", "CountMinSketch", "", 6_000, "tumbling", &[], ""); + let value = make_config( + 10, + "req", + "CountMinSketch", + "count", + 6_000, + "tumbling", + &[], + "", + ); let key = make_config( 11, "req", @@ -1228,7 +1395,16 @@ mod tests { #[test] fn matching_skips_incompatible_key_candidate_and_selects_compatible_one() { - let value = make_config(10, "req", "CountMinSketch", "", 6_000, "sliding", &[], ""); + let value = make_config( + 10, + "req", + "CountMinSketch", + "count", + 6_000, + "sliding", + &[], + "", + ); let mut incompatible = make_config(11, "req", "SetAggregator", "", 6_000, "sliding", &[], ""); incompatible.slide_interval_ms = 2_000; @@ -1255,7 +1431,7 @@ mod tests { 2, "cpu", "CountMinSketch", - "", + "count", 300_000, "tumbling", &["job"], diff --git a/asap-query-engine/src/precompute_engine/accumulator_factory.rs b/asap-query-engine/src/precompute_engine/accumulator_factory.rs index 7044cd2..569e87f 100644 --- a/asap-query-engine/src/precompute_engine/accumulator_factory.rs +++ b/asap-query-engine/src/precompute_engine/accumulator_factory.rs @@ -524,14 +524,16 @@ pub struct CmsAccumulatorUpdater { acc: CountMinSketchAccumulator, row_num: usize, col_num: usize, + count_events: bool, } impl CmsAccumulatorUpdater { - pub fn new(row_num: usize, col_num: usize) -> Self { + pub fn new(row_num: usize, col_num: usize, count_events: bool) -> Self { Self { acc: CountMinSketchAccumulator::new(row_num, col_num), row_num, col_num, + count_events, } } } @@ -545,7 +547,8 @@ impl AccumulatorUpdater for CmsAccumulatorUpdater { } fn update_keyed(&mut self, key: &KeyByLabelValues, value: f64, _timestamp_ms: i64) { - self.acc.inner.update(&key.to_semicolon_str(), value); + let weight = if self.count_events { 1.0 } else { value }; + self.acc.inner.update(&key.to_semicolon_str(), weight); } impl_accumulator_methods!(acc); @@ -732,6 +735,35 @@ fn cms_params(config: &AggregationConfig) -> Result<(usize, usize), String> { Ok((row_num, col_num)) } +/// Resolve the weighting semantics for a plain Count-Min Sketch. +/// +/// 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 { + 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}'" + )) + } +} + +/// 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}'" + )) + } +} + /// Extract `(row_num, col_num, k)` for HydraKLL configs. fn hydra_kll_params(config: &AggregationConfig) -> Result<(usize, usize, u16), String> { let (row_num, col_num) = cms_params(config)?; @@ -766,12 +798,15 @@ fn cms_heap_params(config: &AggregationConfig) -> Result<(usize, usize, usize), /// Whether a CountMinSketchWithHeap config should count events (weight 1 per /// observation, COUNT semantics) rather than summing the sample value. /// Defaults to `true` so `COUNT(...)` top-k works out of the box. -fn cms_count_events(config: &AggregationConfig) -> bool { - config - .parameters - .get("count_events") - .and_then(|v| v.as_bool()) - .unwrap_or(true) +fn cms_count_events(config: &AggregationConfig) -> Result { + match config.parameters.get("count_events") { + None => Ok(true), + Some(value) => value.as_bool().ok_or_else(|| { + format!( + "CountMinSketchWithHeap parameter 'count_events' must be a boolean, got {value}" + ) + }), + } } /// Extract the HLL `precision` parameter from a config. Falls back to @@ -834,7 +869,9 @@ pub fn create_accumulator_updater( "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))) + Ok(Box::new(CmsAccumulatorUpdater::new( + row_num, col_num, false, + ))) } "HydraKLL" | "hydra_kll" => { let (row_num, col_num, k) = hydra_kll_params(config)?; @@ -867,15 +904,21 @@ pub fn create_accumulator_updater( AggregationType::Increase => Ok(Box::new(IncreaseAccumulatorUpdater::new())), AggregationType::CountMinSketch => { let (row_num, col_num) = cms_params(config)?; - Ok(Box::new(CmsAccumulatorUpdater::new(row_num, col_num))) + let count_events = cms_count_events_for_sub_type(sub_type)?; + Ok(Box::new(CmsAccumulatorUpdater::new( + row_num, + col_num, + count_events, + ))) } 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), + cms_count_events(config)?, ))) } AggregationType::HydraKLL => { @@ -1032,7 +1075,7 @@ mod tests { ))); assert!(config_is_keyed(&make_config( AggregationType::CountMinSketch, - "" + "sum" ))); assert!(config_is_keyed(&make_config(AggregationType::HydraKLL, ""))); @@ -1079,7 +1122,11 @@ mod tests { }; for (agg_type, sub_type, params) in [ (AggregationType::DatasketchesKLL, "", kll_params_required()), - (AggregationType::CountMinSketch, "", cms_params_required()), + ( + AggregationType::CountMinSketch, + "sum", + cms_params_required(), + ), ] { let config = make_config_with_params(agg_type, sub_type, params); let updater = create_accumulator_updater(&config).unwrap(); @@ -1323,7 +1370,7 @@ mod tests { let config = AggregationConfig::new( 1, AggregationType::CountMinSketch, - String::new(), + "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![]), @@ -1359,7 +1406,7 @@ mod tests { let config = AggregationConfig::new( 21, AggregationType::CountMinSketch, - String::new(), + "sum".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![]), @@ -1397,6 +1444,102 @@ mod tests { p } + fn cms_config(sub_type: &str) -> AggregationConfig { + AggregationConfig::new( + 100, + AggregationType::CountMinSketch, + sub_type.to_string(), + cms_params_required(), + 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_cms_count_subtype_uses_unit_weight() { + let config = cms_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 cms = acc + .as_any() + .downcast_ref::() + .expect("CountMinSketch accumulator"); + assert_eq!(cms.query_key(&key), 5.0); + } + + #[test] + fn test_cms_sum_subtype_uses_sample_weight() { + let config = cms_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 cms = acc + .as_any() + .downcast_ref::() + .expect("CountMinSketch accumulator"); + assert_eq!(cms.query_key(&key), 50.0); + } + + #[test] + fn test_cms_rejects_empty_subtype() { + let config = cms_config(""); + let err = match create_accumulator_updater(&config) { + Ok(_) => panic!("empty CountMinSketch subtype must fail"), + Err(err) => err, + }; + assert!(err.contains("sum") && err.contains("count")); + } + + #[test] + fn test_cms_rejects_unknown_subtype() { + let config = cms_config("frequency"); + let err = match create_accumulator_updater(&config) { + Ok(_) => panic!("unknown CountMinSketch subtype must fail"), + Err(err) => err, + }; + assert!(err.contains("frequency")); + } + + #[test] + fn test_cms_accepts_case_insensitive_subtype() { + for sub_type in ["COUNT", "SuM"] { + create_accumulator_updater(&cms_config(sub_type)) + .unwrap_or_else(|err| panic!("subtype '{sub_type}' should be accepted: {err}")); + } + } + + #[test] + fn test_cms_rejects_whitespace_padded_subtype() { + let config = cms_config(" count "); + let err = match create_accumulator_updater(&config) { + Ok(_) => panic!("whitespace-padded CountMinSketch subtype must fail"), + Err(err) => err, + }; + assert!(err.contains(" count ")); + } + fn cms_heap_params_required() -> std::collections::HashMap { let mut p = std::collections::HashMap::new(); p.insert("depth".to_string(), serde_json::json!(3_u64)); @@ -1431,6 +1574,43 @@ 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 err = match create_accumulator_updater(&config) { + Ok(_) => panic!("empty CountMinSketchWithHeap subtype must fail"), + Err(err) => err, + }; + assert!(err.contains("topk")); + } + + #[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(); + + 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")); + } + #[test] fn test_cms_with_heap_factory_routes_to_heap_accumulator_and_is_keyed() { // CountMinSketchWithHeap must build a CmsWithHeapAccumulatorUpdater whose @@ -1503,7 +1683,10 @@ mod tests { params.insert("heapsize".to_string(), serde_json::json!(40)); let config = cms_heap_config(params); assert_eq!(cms_heap_params(&config).unwrap(), (4, 2048, 40)); - assert!(cms_count_events(&config), "count_events defaults to true"); + assert!( + cms_count_events(&config).unwrap(), + "count_events defaults to true" + ); } #[test] diff --git a/asap-query-engine/src/precompute_engine/engine.rs b/asap-query-engine/src/precompute_engine/engine.rs index 7fbc9ce..e3818c5 100644 --- a/asap-query-engine/src/precompute_engine/engine.rs +++ b/asap-query-engine/src/precompute_engine/engine.rs @@ -1,4 +1,5 @@ -use crate::data_model::StreamingConfig; +use crate::data_model::{AggregationType, StreamingConfig}; +use crate::precompute_engine::accumulator_factory::create_accumulator_updater; use crate::precompute_engine::config::PrecomputeEngineConfig; use crate::precompute_engine::ingest_source::{IngestContext, IngestSource}; use crate::precompute_engine::output_sink::OutputSink; @@ -154,6 +155,8 @@ impl PrecomputeEngine { /// Start the precompute engine. This spawns worker tasks and all registered /// ingest sources, then blocks until shutdown. pub async fn run(mut self) -> Result<(), Box> { + validate_startup_aggregation_configs(&self.streaming_config)?; + let num_workers = self.config.num_workers; let receivers = self @@ -252,3 +255,239 @@ impl PrecomputeEngine { Ok(()) } } + +/// 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. +fn validate_startup_aggregation_configs( + streaming_config: &StreamingConfig, +) -> Result<(), Box> { + let mut errors = Vec::new(); + + for (&aggregation_id, config) in streaming_config.get_all_aggregation_configs() { + if !matches!( + config.aggregation_type, + AggregationType::CountMinSketch | AggregationType::CountMinSketchWithHeap + ) { + continue; + } + + if let Err(err) = create_accumulator_updater(config) { + errors.push((aggregation_id, err)); + } + } + + if !errors.is_empty() { + errors.sort_by_key(|(aggregation_id, _)| *aggregation_id); + let details = errors + .into_iter() + .map(|(aggregation_id, err)| { + format!("invalid aggregation config for aggregation_id {aggregation_id}: {err}") + }) + .collect::>() + .join("; "); + return Err(details.into()); + } + + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::data_model::{AggregationType, StreamingConfig, WindowType}; + use crate::precompute_engine::config::LateDataPolicy; + use crate::precompute_engine::ingest_source::{IngestContext, IngestSource}; + use crate::precompute_engine::output_sink::NoopOutputSink; + use async_trait::async_trait; + use serde_json::json; + + struct ShutdownSource; + + #[async_trait] + impl IngestSource for ShutdownSource { + async fn run( + self: Box, + ctx: IngestContext, + ) -> Result<(), Box> { + ctx.router.broadcast_shutdown().await + } + } + + #[tokio::test] + async fn run_rejects_invalid_cms_subtype_before_starting_workers() { + let mut parameters = HashMap::new(); + parameters.insert("depth".to_string(), json!(3_u64)); + parameters.insert("width".to_string(), json!(128_u64)); + let cms = AggregationConfig::new( + 1, + AggregationType::CountMinSketch, + String::new(), + parameters, + 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, cms)]))), + Arc::new(NoopOutputSink::new()), + vec![Box::new(ShutdownSource)], + ); + + let result = engine.run().await; + let err = match result { + Ok(()) => panic!("invalid CMS 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(); + parameters.insert("depth".to_string(), json!(3_u64)); + parameters.insert("width".to_string(), json!(128_u64)); + parameters.insert("heapsize".to_string(), json!(32_u64)); + let cms = AggregationConfig::new( + 1, + AggregationType::CountMinSketchWithHeap, + String::new(), + parameters, + 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, cms)]))), + Arc::new(NoopOutputSink::new()), + vec![Box::new(ShutdownSource)], + ); + + let result = engine.run().await; + let err = match result { + Ok(()) => panic!("invalid heap CMS subtype must fail before startup"), + Err(err) => err, + }; + assert!(err.to_string().contains("aggregation_id 1")); + assert!(err.to_string().contains("topk")); + } + + #[tokio::test] + async fn run_reports_all_invalid_cms_configs_in_aggregation_id_order() { + let cms_config = |id, aggregation_type, aggregation_sub_type, parameters| { + AggregationConfig::new( + id, + aggregation_type, + aggregation_sub_type, + parameters, + 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 mut heap_parameters = HashMap::new(); + heap_parameters.insert("depth".to_string(), json!(3_u64)); + heap_parameters.insert("width".to_string(), json!(128_u64)); + heap_parameters.insert("heapsize".to_string(), json!(32_u64)); + + let configs = HashMap::from([ + ( + 20, + cms_config( + 20, + AggregationType::CountMinSketchWithHeap, + String::new(), + heap_parameters, + ), + ), + ( + 3, + cms_config( + 3, + AggregationType::CountMinSketch, + String::new(), + HashMap::from([ + ("depth".to_string(), json!(3_u64)), + ("width".to_string(), json!(128_u64)), + ]), + ), + ), + ]); + let engine = PrecomputeEngine::new( + PrecomputeEngineConfig { + num_workers: 1, + late_data_policy: LateDataPolicy::Drop, + ..PrecomputeEngineConfig::default() + }, + Arc::new(StreamingConfig::new(configs)), + Arc::new(NoopOutputSink::new()), + vec![Box::new(ShutdownSource)], + ); + + let result = engine.run().await; + let err = match result { + Ok(()) => panic!("invalid CMS configs must fail before startup"), + Err(err) => err.to_string(), + }; + let id_3 = err + .find("aggregation_id 3") + .expect("CMS error should be reported"); + let id_20 = err + .find("aggregation_id 20") + .expect("heap CMS error should be reported"); + assert!( + id_3 < id_20, + "errors should be ordered by aggregation ID: {err}" + ); + } +} diff --git a/asap-query-engine/src/precompute_engine/worker.rs b/asap-query-engine/src/precompute_engine/worker.rs index f2e151b..b59a14e 100644 --- a/asap-query-engine/src/precompute_engine/worker.rs +++ b/asap-query-engine/src/precompute_engine/worker.rs @@ -1242,6 +1242,7 @@ mod tests { use crate::precompute_operators::datasketches_kll_accumulator::DatasketchesKLLAccumulator; use crate::precompute_operators::multiple_sum_accumulator::MultipleSumAccumulator; use crate::precompute_operators::sum_accumulator::SumAccumulator; + use crate::precompute_operators::CountMinSketchAccumulator; use asap_sketchlib::KllSketch; use asap_types::enums::{AggregationType, WindowType}; @@ -1349,6 +1350,66 @@ mod tests { .collect() } + #[test] + fn test_count_min_sketch_count_subtype_counts_events_through_worker() { + let mut config = make_agg_config_full( + 6, + "requests_total", + AggregationType::CountMinSketch, + "count", + 1_000, + 1_000, + vec![], + vec!["host"], + ); + config + .parameters + .insert("depth".to_string(), serde_json::json!(3_u64)); + config + .parameters + .insert("width".to_string(), serde_json::json!(128_u64)); + + let sink = Arc::new(CapturingOutputSink::new()); + let mut worker = make_worker( + arc_configs(HashMap::from([(6, config)])), + sink.clone(), + false, + 0, + LateDataPolicy::Drop, + ); + + worker + .process_group_samples( + 6, + "", + vec![ + ("requests_total{host=\"A\"}".to_string(), 100, 100.0), + ("requests_total{host=\"A\"}".to_string(), 200, 200.0), + ], + ) + .unwrap(); + + worker + .process_group_samples( + 6, + "", + group_samples("requests_total{host=\"A\"}", vec![(5_000, 1.0)]), + ) + .unwrap(); + + let captured = sink.drain(); + let (_output, acc) = captured + .iter() + .find(|(output, _)| output.start_timestamp == 0) + .expect("worker should emit the closed [0, 1000) window"); + let cms = acc + .as_any() + .downcast_ref::() + .expect("worker should emit a CountMinSketch accumulator"); + let key = KeyByLabelValues::new_with_labels(vec!["A".to_string()]); + assert_eq!(cms.query_key(&key), 2.0); + } + // ----------------------------------------------------------------------- // Test: raw mode — each sample forwarded as SumAccumulator with sum==value // -----------------------------------------------------------------------