refactor: split the MinMax aggregation family into Min and Max

Cargo.toml

@@ -20,10 +20,10 @@ asap_sketchlib = { git = "https://github.com/ProjectASAP/asap_sketchlib", branch
 [workspace.dependencies]
 # Keep Planner frontends, selection, and IR on the same immutable revision (current-series Planner PR).
 # Alias upstream asap-types because this workspace also defines asap_types.
-planner-types = { package = "asap-types", git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "b8b5d705215200361b6a2c3b389d513d554077ca" }
-asap-aware-mapping = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "b8b5d705215200361b6a2c3b389d513d554077ca" }
-asap-frontend-promql = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "b8b5d705215200361b6a2c3b389d513d554077ca" }
-asap-frontend-sql = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "b8b5d705215200361b6a2c3b389d513d554077ca" }
+planner-types = { package = "asap-types", git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "029ff2fe041172c94c2d32c90b185bc83c5e8a57" }
+asap-aware-mapping = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "029ff2fe041172c94c2d32c90b185bc83c5e8a57" }
+asap-frontend-promql = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "029ff2fe041172c94c2d32c90b185bc83c5e8a57" }
+asap-frontend-sql = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "029ff2fe041172c94c2d32c90b185bc83c5e8a57" }
 
 # Shared external deps (used by 2+ crates)
 serde = { version = "1.0", features = ["derive"] }
@@ -46,3 +46,4 @@ reqwest = { version = "0.12", default-features = false, features = ["json", "rus
 asap_types = { path = "crates/asap_types" }
 asap_otel_proto = { path = "crates/asap_otel_proto" }
 indexmap = { version = "2.0", features = ["serde"] }
+

control_plane/examples/audit_clickhouse_corpus.rs

@@ -26,7 +26,7 @@ struct Row {
 
 fn publication_inputs(schema: &Schema, sql: String) -> ClickHouseSqlWorkload {
     let mut materialization = PrecomputeMaterialization::new(
-        AggregationType::MinMax,
+        AggregationType::Max,
         String::new(),
         std::collections::HashMap::from([("variant".into(), json!(2))]),
         KeyByLabelNames::new(vec!["labels".into()]),

control_plane/src/clickhouse.rs

@@ -1224,7 +1224,7 @@ mod tests {
             ("variant", serde_json::json!(1)),
         );
         let count_60 = materialization(
-            AggregationType::MinMax,
+            AggregationType::Max,
             "requests",
             60,
             10,

control_plane/src/emit/backend_wire.rs

@@ -230,9 +230,8 @@ pub(crate) fn build_backend_aggregation_json(agg: &BackendAggregation) -> JsonVa
             match kind {
                 ExactKind::Sum => "Sum",
                 ExactKind::Count => "Count",
-                // Typed `Min` shares the MinMax accumulator on the wire; the
-                // readout direction is carried by the aggregation sub-type.
-                ExactKind::MinMax | ExactKind::Min => "MinMax",
+                ExactKind::Min => "Min",
+                ExactKind::Max => "Max",
                 ExactKind::Increase => "Increase",
                 ExactKind::Rate => "Rate",
                 ExactKind::IRate => "IRate",
@@ -286,20 +285,7 @@ pub(crate) fn build_backend_aggregation_json(agg: &BackendAggregation) -> JsonVa
         clamp_window_secs(Some(agg.window_secs)).expect("clamp_window_secs preserves Some");
     json!({
         "aggregationType": aggregation_type,
-        // `MinMax` and the typed `Min` share one accumulator on the wire, so the
-        // readout direction has to travel in the sub-type or the backend cannot
-        // tell which end of the window the query wants.
-        "aggregationSubType": match &agg.family {
-            planner_types::post_asap::SummaryFamilyType::ExactAggregate(
-                planner_types::post_asap::ExactKind::MinMax,
-                _,
-            ) => "max",
-            planner_types::post_asap::SummaryFamilyType::ExactAggregate(
-                planner_types::post_asap::ExactKind::Min,
-                _,
-            ) => "min",
-            _ => "",
-        },
+        "aggregationSubType": "",
         "metric": agg.metric_name,
         "labels": {
             "grouping": agg.grouping,

control_plane/src/physical/compiler.rs

@@ -727,7 +727,7 @@ fn has_unsafe_raw_entity_leaf(
                 let preserves_series_state = matches!(
                     family,
                     SummaryFamilyType::ExactAggregate(
-                        ExactKind::Increase | ExactKind::Rate | ExactKind::MinMax | ExactKind::Min,
+                        ExactKind::Increase | ExactKind::Rate | ExactKind::Min | ExactKind::Max,
                         _
                     )
                 );
@@ -1063,7 +1063,7 @@ impl PhysicalPlanCompiler {
                         && (!matches!(
                             state.family,
                             SummaryFamilyType::ExactAggregate(
-                                planner_types::post_asap::ExactKind::MinMax,
+                                planner_types::post_asap::ExactKind::Max,
                                 _
                             )
                         ) || crate::query_plan::residual::selected_range_max_materialization(
@@ -2694,10 +2694,12 @@ fn retained_state_bytes(materialization: &asap_types::PrecomputeMaterialization)
         A::DDSketch => 64 * 1024,
         A::Sum
         | A::Increase
-        | A::MinMax
+        | A::Min
+        | A::Max
         | A::MultipleSum
         | A::MultipleIncrease
-        | A::MultipleMinMax
+        | A::MultipleMin
+        | A::MultipleMax
         | A::SingleSubpopulation
         | A::MultipleSubpopulation => 256,
     }
@@ -2715,7 +2717,7 @@ fn retained_partition_count(
     if materialization.partitioning == Some(asap_types::sds::PopulationPartitioning::PerEntity)
         || matches!(
             materialization.aggregation_type,
-            A::Increase | A::MultipleIncrease | A::MinMax | A::MultipleMinMax
+            A::Increase | A::MultipleIncrease | A::Min | A::Max | A::MultipleMin | A::MultipleMax
         )
         || !materialization.grouping_labels.names().is_empty()
     {
@@ -3874,8 +3876,8 @@ pub(crate) mod tests {
         let plan = PhysicalPlanCompiler.compile_promql(input, env).unwrap();
         assert_eq!(plan.precompute_plan.materializations.len(), 1);
         assert_eq!(
-            plan.precompute_plan.materializations[0].aggregation_sub_type,
-            "min"
+            plan.precompute_plan.materializations[0].aggregation_type,
+            asap_types::AggregationType::Min
         );
         assert!(plan
             .query_plan

control_plane/src/physical/post_asap/matcher.rs

@@ -189,8 +189,8 @@ mod tests {
         match kind {
             ExactKind::Sum => ExactParams::Sum,
             ExactKind::Count => ExactParams::Count,
-            ExactKind::MinMax => ExactParams::MinMax,
             ExactKind::Min => ExactParams::Min,
+            ExactKind::Max => ExactParams::Max,
             ExactKind::Increase => ExactParams::Increase,
             ExactKind::Rate => ExactParams::Rate,
             ExactKind::IRate => ExactParams::IRate,
@@ -285,10 +285,7 @@ mod tests {
     fn exact_accumulator_requires_the_exact_same_kind() {
         let m = SummaryFamilyMatcher;
         assert!(m.is_satisfied_by(&accumulator(ExactKind::Sum), &accumulator(ExactKind::Sum)));
-        assert!(!m.is_satisfied_by(
-            &accumulator(ExactKind::Sum),
-            &accumulator(ExactKind::MinMax)
-        ));
+        assert!(!m.is_satisfied_by(&accumulator(ExactKind::Sum), &accumulator(ExactKind::Max)));
         assert!(!m.is_satisfied_by(
             &accumulator(ExactKind::Increase),
             &accumulator(ExactKind::Rate)

control_plane/src/physical/runtime_capability.rs

@@ -63,7 +63,7 @@ pub enum Capability {
     /// wire format can answer this. `Any` required matches either
     /// `CmsWithHeap` or `CountSketchWithHeap`.
     FrequencyTopk(Option<SketchAlgorithm>),
-    /// Exact-aggregation ASAP-tier state — Sum / Count / MinMax / Avg /
+    /// Exact-aggregation ASAP-tier state — Sum / Count / Min / Max / Avg /
     /// Rate / Increase / SetAggregator etc. Backed by a per-accumulator
     /// payload (`AggPayload::ExactAgg` in the data plane). One variant
     /// per [`AggregationType`] — the inner enum names the concrete
@@ -351,16 +351,17 @@ fn sketch_algorithms_compatible(
 /// True when `available` is the multi-population equivalent of
 /// `required`'s single-population variant — i.e. a `MultipleSum`
 /// policy can serve a `Sum` query (via re-aggregation across keys),
-/// `MultipleIncrease` can serve `Increase`, `MultipleMinMax` can
-/// serve `MinMax`. Asymmetric: this returns `false` for the reverse
+/// `MultipleIncrease` can serve `Increase`, `MultipleMax` can
+/// serve `Max`. Asymmetric: this returns `false` for the reverse
 /// direction (single-pop can't recover keys that have been collapsed
 /// away).
 fn multi_pop_satisfies_single(required: AggregationType, available: AggregationType) -> bool {
     matches!(
         (required, available),
         (AggregationType::Sum, AggregationType::MultipleSum)
             | (AggregationType::Increase, AggregationType::MultipleIncrease)
-            | (AggregationType::MinMax, AggregationType::MultipleMinMax)
+            | (AggregationType::Min, AggregationType::MultipleMin)
+            | (AggregationType::Max, AggregationType::MultipleMax)
     )
 }
 
@@ -394,9 +395,12 @@ pub fn capability_for(intent: &AggIntent) -> Option<Capability> {
     }
     match intent {
         AggIntent::Sum { .. } => Some(Capability::ExactAgg(AggregationType::Sum)),
-        AggIntent::Min { .. } | AggIntent::Max { .. } => {
-            Some(Capability::ExactAgg(AggregationType::MinMax))
-        }
+        // Direction is part of the capability: a stored minimum cannot
+        // answer `max_over_time` and vice versa, so these must not
+        // collapse onto one `ExactAgg` the way they did while Planner
+        // had a single `MinMax` accumulator.
+        AggIntent::Min { .. } => Some(Capability::ExactAgg(AggregationType::Min)),
+        AggIntent::Max { .. } => Some(Capability::ExactAgg(AggregationType::Max)),
         AggIntent::Increase | AggIntent::Rate => {
             Some(Capability::ExactAgg(AggregationType::Increase))
         }
@@ -537,24 +541,36 @@ mod tests {
     }
 
     #[test]
-    fn capability_for_min_returns_exact_agg_minmax() {
+    fn capability_for_min_returns_exact_agg_min() {
         // Min/Max are exact, mergeable accumulators -- no approximation
         // needed at all -- matching ASAPController's own
         // `crates/plan/src/boundary.rs` treatment.
         assert_eq!(
             capability_for(&AggIntent::Min { col: None }),
-            Some(Capability::ExactAgg(AggregationType::MinMax))
+            Some(Capability::ExactAgg(AggregationType::Min))
         );
     }
 
     #[test]
-    fn capability_for_max_returns_exact_agg_minmax() {
+    fn capability_for_max_returns_exact_agg_max() {
         assert_eq!(
             capability_for(&AggIntent::Max { col: None }),
-            Some(Capability::ExactAgg(AggregationType::MinMax))
+            Some(Capability::ExactAgg(AggregationType::Max))
         );
     }
 
+    #[test]
+    fn exact_agg_min_and_max_do_not_satisfy_each_other() {
+        // The whole point of splitting the family: a deployed minimum
+        // sid must never be routed a `max_over_time` read.
+        assert!(!Capability::ExactAgg(AggregationType::Min)
+            .is_satisfied_by(&Capability::ExactAgg(AggregationType::Max)));
+        assert!(!Capability::ExactAgg(AggregationType::Max)
+            .is_satisfied_by(&Capability::ExactAgg(AggregationType::Min)));
+        assert!(!Capability::ExactAgg(AggregationType::Min)
+            .is_satisfied_by(&Capability::ExactAgg(AggregationType::MultipleMax)));
+    }
+
     #[test]
     fn capability_for_rate_increase_route_to_exact_agg_increase() {
         // PR-6 follow-up: Rate and Increase route to ASAP-tier
@@ -762,10 +778,10 @@ mod tests {
 
     #[test]
     fn is_satisfied_by_exact_agg_different_types_do_not_match() {
-        // Sum required, MinMax indexed → no match. No wildcard for
+        // Sum required, Max indexed → no match. No wildcard for
         // ExactAgg — every agg_type stands on its own.
         let required = Capability::ExactAgg(AggregationType::Sum);
-        let indexed = Capability::ExactAgg(AggregationType::MinMax);
+        let indexed = Capability::ExactAgg(AggregationType::Max);
         assert!(!required.is_satisfied_by(&indexed));
     }
 
@@ -805,11 +821,13 @@ mod tests {
         let cases = [
             AggregationType::Sum,
             AggregationType::Increase,
-            AggregationType::MinMax,
+            AggregationType::Min,
+            AggregationType::Max,
             AggregationType::DatasketchesKLL,
             AggregationType::MultipleSum,
             AggregationType::MultipleIncrease,
-            AggregationType::MultipleMinMax,
+            AggregationType::MultipleMin,
+            AggregationType::MultipleMax,
             AggregationType::HydraKLL,
             AggregationType::CountMinSketch,
             AggregationType::CountMinSketchWithHeap,