refactor: fold heap-CMS count_events into aggregation_sub_type

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, AggregationConfigError> {
+        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,

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<Self, String> {
+        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
+        }
+    }
+}