fix: reject invalid HLL precision and Min/Max subtype

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

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

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");
+    }
 }