feat(sid): keyed ExactAgg — MultipleSum / MultipleIncrease / MultipleMinMax

control_plane/src/asap_tier_analysis.rs

@@ -447,16 +447,29 @@ pub fn policy_capability(cfg: &asap_types::AggregationConfig) -> Option<Capabili
         AggregationType::CountMinSketchWithHeap => {
             Some(Capability::FrequencyTopk(SketchKindHandle::CmsWithHeap))
         }
-        // Keyed-multi-population variants and legacy wrappers — no
-        // standalone ASAP-tier capability today. The L4 binder doesn't
-        // yet emit `PhysicalExpr::ExactAgg` for keyed `MultipleSum` /
-        // `MultipleIncrease` shapes (the matching capability doesn't
-        // exist either). When the keyed-ExactAgg follow-up lands, this
-        // function gets the corresponding arms.
-        AggregationType::MultipleSum
-        | AggregationType::MultipleIncrease
-        | AggregationType::MultipleMinMax
-        | AggregationType::HydraKLL
+        // Keyed-multi-population variants. The capability the policy
+        // *provides* is the multi-pop variant itself; the matching
+        // predicate (`Capability::is_satisfied_by`) recognises that
+        // a multi-pop indexed capability satisfies a single-pop
+        // required capability through `multi_pop_satisfies_single`.
+        // So a candidate's `ExactAgg(Sum)` matches a policy whose
+        // `policy_capability` returns `ExactAgg(MultipleSum)`.
+        AggregationType::MultipleSum => Some(Capability::ExactAgg(AggregationType::MultipleSum)),
+        AggregationType::MultipleIncrease => {
+            Some(Capability::ExactAgg(AggregationType::MultipleIncrease))
+        }
+        AggregationType::MultipleMinMax => {
+            Some(Capability::ExactAgg(AggregationType::MultipleMinMax))
+        }
+        // No ASAP-tier capability today. HydraKLL is a keyed-quantile
+        // family that needs its own QuantileApprox arm (with a
+        // multi-pop equivalent rule) — separate follow-up. SetAggregator
+        // and DeltaSetAggregator are exact-set-membership primitives;
+        // they serve `count(distinct ...)` queries through a different
+        // routing path (not via candidate capability matching).
+        // `Single/MultipleSubpopulation` are legacy enum wrappers from
+        // the pre-refactor config schema and have no semantic shape.
+        AggregationType::HydraKLL
         | AggregationType::SetAggregator
         | AggregationType::DeltaSetAggregator
         | AggregationType::SingleSubpopulation
@@ -927,9 +940,73 @@ mod tests {
         }
 
         #[test]
-        fn policy_capability_returns_none_for_multi_pop_variants() {
+        fn policy_capability_maps_multiple_sum_to_exact_agg_multiple_sum() {
             let c = cfg("m", AggregationType::MultipleSum, vec!["zone"], 60, "");
-            assert!(policy_capability(&c).is_none());
+            assert_eq!(
+                policy_capability(&c),
+                Some(Capability::ExactAgg(AggregationType::MultipleSum))
+            );
+        }
+
+        #[test]
+        fn multiple_sum_policy_satisfies_unkeyed_sum_query() {
+            // MultipleSum policy keeps per-zone state; an unkeyed Sum
+            // query can re-aggregate across zones. The is_satisfied_by
+            // multi-pop-satisfies-single rule + the group_by ⊆ policy
+            // grouping check let it through.
+            let policies = vec![cfg(
+                "http_lat",
+                AggregationType::MultipleSum,
+                vec!["zone"],
+                60,
+                "",
+            )];
+            let registry = PolicyRegistry::from_configs(policies);
+            let cand = candidate(
+                "http_lat",
+                &[],
+                Capability::ExactAgg(AggregationType::Sum),
+                60,
+            );
+            assert_eq!(find_matching_policies(&registry, &cand).len(), 1);
+        }
+
+        #[test]
+        fn multiple_increase_policy_satisfies_keyed_increase_query() {
+            let policies = vec![cfg(
+                "http_requests_total",
+                AggregationType::MultipleIncrease,
+                vec!["zone", "service"],
+                60,
+                "",
+            )];
+            let registry = PolicyRegistry::from_configs(policies);
+            // Query asks for per-zone increase; policy keeps {zone,
+            // service} (superset).
+            let cand = candidate(
+                "http_requests_total",
+                &["zone"],
+                Capability::ExactAgg(AggregationType::Increase),
+                60,
+            );
+            assert_eq!(find_matching_policies(&registry, &cand).len(), 1);
+        }
+
+        #[test]
+        fn single_pop_policy_does_not_satisfy_keyed_query() {
+            // Unkeyed Sum policy can't answer per-zone Sum — keys
+            // already collapsed. Group_by ⊆ policy_grouping_labels
+            // check rejects this even though capabilities would
+            // structurally satisfy.
+            let policies = vec![cfg("http_lat", AggregationType::Sum, vec![], 60, "")];
+            let registry = PolicyRegistry::from_configs(policies);
+            let cand = candidate(
+                "http_lat",
+                &["zone"],
+                Capability::ExactAgg(AggregationType::Sum),
+                60,
+            );
+            assert!(find_matching_policies(&registry, &cand).is_empty());
         }
 
         #[test]
@@ -1156,12 +1233,13 @@ mod tests {
         }
 
         #[test]
-        fn ignores_multi_pop_policies() {
-            // Multi-population policies have `policy_capability == None`;
-            // matching skips them even when other fields would line up.
+        fn ignores_unsupported_multi_pop_variants() {
+            // `HydraKLL`, `SetAggregator`, etc. have
+            // `policy_capability == None` because no Capability variant
+            // covers their shape today. Matching skips them.
             let policies = vec![cfg(
                 "http_lat",
-                AggregationType::MultipleSum,
+                AggregationType::SetAggregator,
                 vec!["zone"],
                 60,
                 "",

control_plane/src/sketch_algebra/capability.rs

@@ -175,12 +175,21 @@ impl Capability {
             (Capability::FrequencyEstimate(req), Capability::FrequencyTopk(have)) => {
                 is_heap_bearing(*have) && handles_compatible(*req, *have)
             }
-            // Exact-aggregation family: the agg_type must match exactly.
-            // There is no `Any` wildcard for ExactAgg — a Sum sid does
-            // not satisfy a MinMax requirement and vice versa. If a
-            // future PR introduces a wildcard semantic (e.g. "any
-            // single-population accumulator"), extend the match here.
-            (Capability::ExactAgg(req), Capability::ExactAgg(have)) => req == have,
+            // Exact-aggregation family: the agg_type must match
+            // exactly OR be the single-pop ⇆ multi-pop equivalent. A
+            // `MultipleSum` policy can serve a `Sum` query by
+            // re-aggregating across keys; the `find_matching_policies`
+            // group_by ⊆ policy_grouping_labels check is what
+            // ultimately decides whether the re-aggregation is
+            // semantically valid. The reverse direction (single-pop
+            // serving multi-pop) is NOT allowed — the single-pop
+            // policy has lost the key dimension and can't recover it.
+            //
+            // Cross-family ExactAgg combos (Sum vs MinMax, etc.)
+            // remain non-satisfiable: they're different operations.
+            (Capability::ExactAgg(req), Capability::ExactAgg(have)) => {
+                req == have || multi_pop_satisfies_single(*req, *have)
+            }
             _ => false,
         }
     }
@@ -223,6 +232,22 @@ fn is_frequency_family(h: SketchKindHandle) -> bool {
     )
 }
 
+/// 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
+/// 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)
+    )
+}
+
 // ── AggIntent → Capability bridge ────────────────────────────────────────────
 
 /// Map a semantic [`AggIntent`] to the ASAP-tier [`Capability`] that can

control_plane/src/sketch_algebra/rules/bind_exact_agg.rs

@@ -77,15 +77,25 @@ impl Rule for BindExactAgg {
     }
 
     fn apply(&self, expr: &QueryExpr, _accuracy: &AccuracyTarget) -> Option<PhysicalExpr> {
-        let (intent, child) = match expr {
+        let (intent, child, keyed) = match expr {
             QueryExpr::Aggregate {
                 aggs, child, by, ..
-            } if aggs.len() == 1 && by.is_empty() => (&aggs[0], child),
+            } if aggs.len() == 1 => (&aggs[0], child, !by.is_empty()),
             _ => return None,
         };
 
+        // Keyed aggregations (non-empty `by`) lower to the multi-pop
+        // accumulator variant; the data plane stores per-key state so
+        // the query can fan results out over the surviving labels.
+        // Unkeyed aggregations stay on the single-pop variant.
         let agg_type = match intent {
-            AggIntent::Sum => AggregationType::Sum,
+            AggIntent::Sum => {
+                if keyed {
+                    AggregationType::MultipleSum
+                } else {
+                    AggregationType::Sum
+                }
+            }
             AggIntent::Rate { window } | AggIntent::Increase { window } => {
                 // The window is informational here — the data plane keys
                 // the policy on (metric, attrs, agg_kind, filter) plus
@@ -95,11 +105,24 @@ impl Rule for BindExactAgg {
                 if *window == Duration::ZERO {
                     return None;
                 }
-                AggregationType::Increase
+                if keyed {
+                    AggregationType::MultipleIncrease
+                } else {
+                    AggregationType::Increase
+                }
             }
             AggIntent::Count {
                 accuracy: AccuracyTarget::Exact,
-            } => AggregationType::Sum,
+            } => {
+                // count_over_time = sum-of-1s, so it lowers through the
+                // Sum/MultipleSum accumulator family — same as
+                // `AggIntent::Sum` above.
+                if keyed {
+                    AggregationType::MultipleSum
+                } else {
+                    AggregationType::Sum
+                }
+            }
             _ => return None,
         };
 
@@ -213,20 +236,66 @@ mod tests {
         assert!(BindExactAgg.apply(&expr, &AccuracyTarget::Exact).is_none());
     }
 
-    #[test]
-    fn does_not_bind_when_by_clause_present() {
-        // Group-by-bearing intents are L3-canonical-form-only here;
-        // this rule mirrors the existing bind_ddsketch_quantile shape
-        // and rejects `by`-bearing inputs. A keyed ExactAgg follow-up
-        // would emit `MultipleSum` / `MultipleIncrease` instead — see
-        // the AggregationType enum.
-        let expr = QueryExpr::Aggregate {
-            aggs: vec![AggIntent::Sum],
-            child: Box::new(scan("test_metric")),
-            by: vec![0],
+    fn agg_over_with_by(intent: AggIntent, metric: &str, by: Vec<usize>) -> QueryExpr {
+        QueryExpr::Aggregate {
+            aggs: vec![intent],
+            child: Box::new(scan(metric)),
+            by,
             having: None,
-        };
-        assert!(BindExactAgg.apply(&expr, &AccuracyTarget::Exact).is_none());
+        }
+    }
+
+    fn check_keyed_binds(intent: AggIntent, expected: AggregationType) {
+        let expr = agg_over_with_by(intent, "test_metric", vec![0]);
+        let bound = BindExactAgg
+            .apply(&expr, &AccuracyTarget::Exact)
+            .unwrap_or_else(|| panic!("rule didn't fire on keyed {expected:?}"));
+        match bound {
+            PhysicalExpr::ExactAgg { agg_type, .. } => assert_eq!(agg_type, expected),
+            other => panic!("expected ExactAgg, got {other:?}"),
+        }
+    }
+
+    #[test]
+    fn keyed_sum_binds_to_multiple_sum() {
+        check_keyed_binds(AggIntent::Sum, AggregationType::MultipleSum);
+    }
+
+    #[test]
+    fn keyed_rate_binds_to_multiple_increase() {
+        check_keyed_binds(
+            AggIntent::Rate {
+                window: Duration::from_secs(60),
+            },
+            AggregationType::MultipleIncrease,
+        );
+    }
+
+    #[test]
+    fn keyed_increase_binds_to_multiple_increase() {
+        check_keyed_binds(
+            AggIntent::Increase {
+                window: Duration::from_secs(300),
+            },
+            AggregationType::MultipleIncrease,
+        );
+    }
+
+    #[test]
+    fn keyed_count_exact_binds_to_multiple_sum() {
+        check_keyed_binds(
+            AggIntent::Count {
+                accuracy: AccuracyTarget::Exact,
+            },
+            AggregationType::MultipleSum,
+        );
+    }
+
+    #[test]
+    fn unkeyed_sum_still_binds_to_single_pop_sum() {
+        // Regression guard: the keyed/unkeyed branch must still
+        // dispatch correctly on `by.is_empty()`.
+        check_binds(AggIntent::Sum, AggregationType::Sum);
     }
 
     #[test]

control_plane/src/sketch_algebra/tests.rs

@@ -449,11 +449,13 @@ fn phase_b_pattern_only_temporal_sum_binds_to_exact_agg() {
 }
 
 /// `ONLY_SPATIAL` — `sum by (host) (m)`.
-/// Control plane path: `Aggregate{Sum, by=[host]}` over a bare `Scan` (no
-/// `Window`). Sum is exact → Logical pass-through. The point of the test
-/// is the by-clause survives binding intact.
+/// Control plane path: `Aggregate{Sum, by=[host]}` over a bare `Scan`.
+/// Post keyed-ExactAgg follow-up, `BindExactAgg` lowers this to
+/// `PhysicalExpr::ExactAgg { agg_type: MultipleSum, .. }` — the
+/// multi-pop accumulator family the data plane uses for keyed
+/// per-group sums.
 #[test]
-fn phase_b_pattern_only_spatial_aggregate_preserves_by_clause() {
+fn phase_b_pattern_only_spatial_aggregate_binds_to_multiple_sum() {
     let expr = QueryExpr::Aggregate {
         by: vec![1], // service column
         aggs: vec![AggIntent::Sum],
@@ -462,19 +464,19 @@ fn phase_b_pattern_only_spatial_aggregate_preserves_by_clause() {
     };
     let bound = bind_query_expr(&expr, AccuracyTarget::Epsilon(0.01)).unwrap();
     match bound {
-        PhysicalExpr::Logical(QueryExpr::Aggregate { by, .. }) => {
-            assert_eq!(by, vec![1]);
-        }
-        other => panic!("expected Logical(Aggregate), got {other:?}"),
+        PhysicalExpr::ExactAgg { agg_type, .. } => assert_eq!(
+            agg_type,
+            promql_utilities::query_logics::enums::AggregationType::MultipleSum,
+        ),
+        other => panic!("expected ExactAgg(MultipleSum), got {other:?}"),
     }
 }
 
 /// `ONE_TEMPORAL_ONE_SPATIAL` — `sum by (host) (rate(m[5m]))`.
-/// Control plane path: combined `Aggregate{Sum, by=[host]}` over `Window` —
-/// the L3 algebra captures both axes natively without needing the legacy
-/// pattern's `One*One*` enum.
+/// Post keyed-ExactAgg follow-up: `Rate` keyed by `host` lowers to
+/// `MultipleIncrease`.
 #[test]
-fn phase_b_pattern_temporal_and_spatial_combined() {
+fn phase_b_pattern_temporal_and_spatial_combined_binds_to_multiple_increase() {
     let expr = QueryExpr::Aggregate {
         by: vec![1],
         aggs: vec![AggIntent::Rate {
@@ -484,8 +486,13 @@ fn phase_b_pattern_temporal_and_spatial_combined() {
         child: Box::new(windowed_scan()),
     };
     let bound = bind_query_expr(&expr, AccuracyTarget::Epsilon(0.01)).unwrap();
-    // Rate has no ASAP-tier sketch family today — expect Logical.
-    assert!(matches!(bound, PhysicalExpr::Logical(_)));
+    match bound {
+        PhysicalExpr::ExactAgg { agg_type, .. } => assert_eq!(
+            agg_type,
+            promql_utilities::query_logics::enums::AggregationType::MultipleIncrease,
+        ),
+        other => panic!("expected ExactAgg(MultipleIncrease), got {other:?}"),
+    }
 }
 
 /// Phase β archive-only intent: any of the no-ASAP-tier-family entries