fix(query-engine,asap-types): reject invalid HLL precision and Min/Max subtype

HLL precision out-of-range/non-numeric values silently clamped to the
default (14) instead of erroring, and unrecognized Min/Max subtypes
silently resolved to Min. Both let a config typo change aggregation
semantics without failing loudly (#674).

precision is now a required, range-checked HLL parameter, and
MinMax/MultipleMinMax subtypes must case-insensitively be "min" or
"max" — both enforced in AggregationConfig::validate() alongside the
existing count_events check.

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

Author: milindsrivastava1997

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)]

asap-common/dependencies/rs/asap_types/src/capability_matching.rs

@@ -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
     }
 

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::<AggregationConfigError>(),
+            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::<AggregationConfigError>(),
+            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::<AggregationConfigError>(),
+            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");
+    }
 }

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<bool, String> {
         .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![]),