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 f4c1bf1..9d3e680 100644 --- a/asap-common/dependencies/rs/asap_types/src/aggregation_config.rs +++ b/asap-common/dependencies/rs/asap_types/src/aggregation_config.rs @@ -9,6 +9,11 @@ use crate::utils::normalize_spatial_filter; use promql_utilities::data_model::KeyByLabelNames; use promql_utilities::query_logics::enums::AggregationType; +/// Valid range for the HLL `precision` parameter, per the underlying +/// `HyperLogLogPlus` storage (`datafusion_summary_library::physical::hll`). +pub const HLL_MIN_PRECISION: u32 = 4; +pub const HLL_MAX_PRECISION: u32 = 18; + #[derive(Debug, thiserror::Error)] pub enum AggregationConfigError { #[error( @@ -26,6 +31,31 @@ pub enum AggregationConfigError { aggregation_id: u64, aggregation_type: AggregationType, }, + #[error("aggregation {aggregation_id} (HLL) missing required parameter 'precision'")] + MissingPrecision { aggregation_id: u64 }, + #[error( + "aggregation {aggregation_id} (HLL) parameter 'precision' must be an integer, got {value}" + )] + InvalidPrecisionType { aggregation_id: u64, value: Value }, + #[error( + "aggregation {aggregation_id} (HLL) parameter 'precision' must be between {HLL_MIN_PRECISION} and {HLL_MAX_PRECISION}, got {value}" + )] + PrecisionOutOfRange { aggregation_id: u64, value: u64 }, + #[error( + "aggregation {aggregation_id} ({aggregation_type}) parameter 'precision' is only valid for HLL" + )] + MisplacedPrecision { + 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)] @@ -69,28 +99,88 @@ pub struct AggregationIdInfo { impl AggregationConfig { pub fn validate(&self) -> Result<(), AggregationConfigError> { + self.validate_count_events()?; + 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 => { - return Err(AggregationConfigError::MissingCountEvents { - aggregation_id: self.aggregation_id, - }) - } + None => Err(AggregationConfigError::MissingCountEvents { + aggregation_id: self.aggregation_id, + }), Some(value) if !value.is_boolean() => { - return Err(AggregationConfigError::InvalidCountEventsType { + Err(AggregationConfigError::InvalidCountEventsType { aggregation_id: self.aggregation_id, value: value.clone(), }) } - Some(_) => {} + Some(_) => Ok(()), } } else if self.parameters.contains_key("count_events") { - return Err(AggregationConfigError::MisplacedCountEvents { + Err(AggregationConfigError::MisplacedCountEvents { aggregation_id: self.aggregation_id, aggregation_type: self.aggregation_type, - }); + }) + } else { + Ok(()) + } + } + + fn validate_hll_precision(&self) -> Result<(), AggregationConfigError> { + if self.aggregation_type == AggregationType::HLL { + match self.parameters.get("precision") { + None => Err(AggregationConfigError::MissingPrecision { + aggregation_id: self.aggregation_id, + }), + Some(value) => match value.as_u64() { + None => Err(AggregationConfigError::InvalidPrecisionType { + aggregation_id: self.aggregation_id, + value: value.clone(), + }), + Some(precision) + if precision < HLL_MIN_PRECISION as u64 + || precision > HLL_MAX_PRECISION as u64 => + { + Err(AggregationConfigError::PrecisionOutOfRange { + aggregation_id: self.aggregation_id, + value: precision, + }) + } + Some(_) => Ok(()), + }, + } + } else if self.parameters.contains_key("precision") { + Err(AggregationConfigError::MisplacedPrecision { + aggregation_id: self.aggregation_id, + aggregation_type: self.aggregation_type, + }) + } else { + Ok(()) + } + } + + 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(), + }) } - Ok(()) } #[allow(clippy::too_many_arguments)] 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 7185ffe..6edd88d 100644 --- a/asap-common/dependencies/rs/asap_types/src/capability_matching.rs +++ b/asap-common/dependencies/rs/asap_types/src/capability_matching.rs @@ -293,7 +293,7 @@ pub fn find_compatible_aggregation( .filter(|c| { let ok = c.metric == requirements.metric && types.contains(&c.aggregation_type) - && sub_type.is_none_or(|st| c.aggregation_sub_type == st) + && sub_type.is_none_or(|st| c.aggregation_sub_type.eq_ignore_ascii_case(st)) && window_compatible(c, requirements.data_range_ms) && labels_compatible(&c.grouping_labels, &requirements.grouping_labels) && spatial_filter_compatible( @@ -485,6 +485,11 @@ mod tests { .parameters .insert("count_events".to_string(), serde_json::Value::Bool(true)); } + if config.aggregation_type == AggregationType::HLL { + config + .parameters + .insert("precision".to_string(), serde_json::Value::from(14)); + } config } @@ -1010,6 +1015,29 @@ mod tests { assert!(result.is_none()); } + #[test] + fn sub_type_matching_is_case_insensitive() { + // AggregationConfig::validate() accepts "MAX" (case-insensitive), so + // capability matching must recognize it too, or an accepted config + // silently becomes unreachable for Max queries (roborev #715/194). + let configs = single_config(make_config( + 1, + "cpu", + "MinMax", + "MAX", + 300_000, + "tumbling", + &[], + "", + )); + let result = + find_compatible_aggregation(&configs, &req("cpu", &[Statistic::Max], 300_000, &[], "")); + assert!( + result.is_some(), + "uppercase 'MAX' subtype must still match a Max query" + ); + } + // --- multi-population --- #[test] 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 6c6467c..1ddd96a 100644 --- a/asap-common/dependencies/rs/asap_types/src/streaming_config.rs +++ b/asap-common/dependencies/rs/asap_types/src/streaming_config.rs @@ -239,4 +239,162 @@ aggregations: .to_string() .contains("only valid for CountMinSketchWithHeap")); } + + fn hll_yaml(parameters_yaml: &str) -> Value { + serde_yaml::from_str(&format!( + r#" +aggregations: + - aggregationId: 1 + aggregationType: HLL + aggregationSubType: distinct + parameters: + {parameters_yaml} + labels: + grouping: [] + aggregated: [instance] + rollup: [] + metric: http_requests_total + windowSizeMs: 15000 + slideIntervalMs: 15000 + windowType: tumbling + spatialFilter: '' +"# + )) + .unwrap() + } + + #[test] + fn rejects_hll_config_without_precision() { + let yaml = hll_yaml("{}"); + + let error = StreamingConfig::from_yaml_data(&yaml, None) + .expect_err("HLL config without precision must be rejected"); + + assert!(matches!( + error.downcast_ref::(), + Some(AggregationConfigError::MissingPrecision { aggregation_id: 1 }) + )); + } + + #[test] + fn rejects_hll_config_with_non_integer_precision() { + let yaml = hll_yaml(r#"precision: "fourteen""#); + + let error = StreamingConfig::from_yaml_data(&yaml, None) + .expect_err("HLL config with non-integer precision must be rejected"); + + assert!(error.to_string().contains("aggregation 1")); + assert!(error.to_string().contains("precision")); + assert!(error.to_string().contains("integer")); + } + + #[test] + fn rejects_hll_config_with_out_of_range_precision() { + // Issue #674: a typo'd precision (e.g. 20) must not silently clamp to + // the default (14) — it must fail configuration. + let yaml = hll_yaml("precision: 20"); + + let error = StreamingConfig::from_yaml_data(&yaml, None) + .expect_err("HLL config with out-of-range precision must be rejected"); + + assert!(matches!( + error.downcast_ref::(), + Some(AggregationConfigError::PrecisionOutOfRange { + aggregation_id: 1, + value: 20, + .. + }) + )); + } + + #[test] + fn rejects_precision_on_non_hll_config() { + let yaml: Value = serde_yaml::from_str( + r#" +aggregations: + - aggregationId: 1 + aggregationType: Sum + aggregationSubType: sum + parameters: + precision: 14 + 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("precision on a non-HLL aggregation must be rejected"); + + assert!(error.to_string().contains("aggregation 1")); + assert!(error.to_string().contains("precision")); + assert!(error.to_string().contains("only valid for HLL")); + } + + fn minmax_yaml(aggregation_type: &str, sub_type: &str) -> Value { + serde_yaml::from_str(&format!( + r#" +aggregations: + - aggregationId: 1 + aggregationType: {aggregation_type} + aggregationSubType: {sub_type} + parameters: {{}} + labels: + grouping: [] + aggregated: [instance] + rollup: [] + metric: http_requests_total + windowSizeMs: 15000 + slideIntervalMs: 15000 + windowType: tumbling + spatialFilter: '' +"# + )) + .unwrap() + } + + #[test] + fn rejects_minmax_config_with_misspelled_subtype() { + // Issue #674: a typo'd subtype (e.g. "Mxa") must not silently be + // interpreted as "min" — it must fail configuration. + let yaml = minmax_yaml("MinMax", "Mxa"); + + let error = StreamingConfig::from_yaml_data(&yaml, None) + .expect_err("MinMax config with a misspelled subtype must be rejected"); + + assert!(error.to_string().contains("aggregation 1")); + assert!(error.to_string().contains("Mxa")); + assert!(error.to_string().contains("min") || error.to_string().contains("max")); + } + + #[test] + fn rejects_multiple_minmax_config_with_misspelled_subtype() { + let yaml = minmax_yaml("MultipleMinMax", "Mxa"); + + let error = StreamingConfig::from_yaml_data(&yaml, None) + .expect_err("MultipleMinMax config with a misspelled subtype must be rejected"); + + assert!(matches!( + error.downcast_ref::(), + Some(AggregationConfigError::InvalidMinMaxSubType { + aggregation_id: 1, + .. + }) + )); + } + + #[test] + fn accepts_minmax_config_with_case_insensitive_subtype() { + let yaml = minmax_yaml("MinMax", "MAX"); + + StreamingConfig::from_yaml_data(&yaml, None) + .expect("MinMax config with 'MAX' subtype must be accepted"); + } } diff --git a/asap-query-engine/src/precompute_engine/accumulator_factory.rs b/asap-query-engine/src/precompute_engine/accumulator_factory.rs index a99c4fb..6bad3c3 100644 --- a/asap-query-engine/src/precompute_engine/accumulator_factory.rs +++ b/asap-query-engine/src/precompute_engine/accumulator_factory.rs @@ -3,7 +3,7 @@ use crate::precompute_operators::{ CountMinSketchAccumulator, CountMinSketchWithHeapAccumulator, DatasketchesKLLAccumulator, DeltaSetAggregatorAccumulator, HllAccumulator, HydraKllSketchAccumulator, IncreaseAccumulator, MinMaxAccumulator, MultipleIncreaseAccumulator, MultipleMinMaxAccumulator, - MultipleSumAccumulator, SetAggregatorAccumulator, SumAccumulator, DEFAULT_HLL_PRECISION, + MultipleSumAccumulator, SetAggregatorAccumulator, SumAccumulator, }; use asap_types::aggregation_config::AggregationConfig; @@ -914,26 +914,11 @@ fn cms_count_events(config: &AggregationConfig) -> Result { .expect("validation guarantees a boolean count_events parameter")) } -/// Extract the HLL `precision` parameter from a config. Falls back to -/// `DEFAULT_HLL_PRECISION` (14) when absent or non-numeric. The valid range is -/// 4..=18 per the underlying `HllSketch` storage; out-of-range values are -/// clamped and warned about so a typo doesn't crash the streaming worker. +/// Extract the HLL `precision` parameter from a config. fn hll_precision_param(config: &AggregationConfig) -> u32 { - let raw = config - .parameters - .get("precision") - .and_then(|v| v.as_u64()) - .map(|v| v as u32); - match raw { - Some(p) if (4..=18).contains(&p) => p, - Some(p) => { - tracing::warn!( - "HLL precision {p} is out of range (4..=18); using default {DEFAULT_HLL_PRECISION}" - ); - DEFAULT_HLL_PRECISION - } - None => DEFAULT_HLL_PRECISION, - } + config.parameters["precision"] + .as_u64() + .expect("validation guarantees an in-range integer precision parameter") as u32 } // --------------------------------------------------------------------------- @@ -1238,7 +1223,7 @@ mod tests { 42, AggregationType::HLL, String::new(), - HashMap::new(), + HashMap::from([("precision".to_string(), serde_json::json!(14))]), 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![]), @@ -1318,15 +1303,16 @@ mod tests { } #[test] - fn test_hll_updater_default_precision_is_14() { - // When no `precision` parameter is supplied, the factory must use the - // documented default (14) — not whatever the type default resolves to. + fn test_hll_updater_honors_explicit_precision() { + // `precision` is a required parameter (issue #674: a missing/invalid + // precision must not silently fall back to a default — see + // AggregationConfig::validate() regression tests in asap_types). use std::collections::HashMap; let config = AggregationConfig::new( 7, AggregationType::HLL, String::new(), - HashMap::new(), + HashMap::from([("precision".to_string(), serde_json::json!(14))]), 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![]), @@ -1360,7 +1346,7 @@ mod tests { 7, AggregationType::HLL, String::new(), - HashMap::new(), + HashMap::from([("precision".to_string(), serde_json::json!(14))]), 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![]), diff --git a/asap-query-engine/src/precompute_engine/worker.rs b/asap-query-engine/src/precompute_engine/worker.rs index b7e4985..d5c344d 100644 --- a/asap-query-engine/src/precompute_engine/worker.rs +++ b/asap-query-engine/src/precompute_engine/worker.rs @@ -1294,6 +1294,9 @@ mod tests { vec!["srcip"], ); config.value_column = Some("dstip".to_string()); + config + .parameters + .insert("precision".to_string(), serde_json::json!(14)); let mut agg_configs = HashMap::new(); agg_configs.insert(4, config);