feat(engine): shadow-mode SummaryExecutor comparison (no serving change)

control_plane/Cargo.toml

@@ -68,10 +68,10 @@ asap_types.workspace = true
 # `AggIntent::Extension` hook this repo's `Extension{"frequency"}` intent
 # needs (ASAPController#150). 64df20d is a strict descendant of d4c1756
 # (the previous pin), so nothing this repo already consumes moves.
-asap-ir = { git = "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/ProjectASAP/ASAPController", rev = "64df20d90c3ddd519c726dea45c05e1fe5225ce6" }
-asap-l2 = { git = "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/ProjectASAP/ASAPController", rev = "64df20d90c3ddd519c726dea45c05e1fe5225ce6" }
-asap-sketch = { git = "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/ProjectASAP/ASAPController", rev = "64df20d90c3ddd519c726dea45c05e1fe5225ce6" }
-asap-plan = { git = "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/ProjectASAP/ASAPController", rev = "64df20d90c3ddd519c726dea45c05e1fe5225ce6" }
+asap-ir = { git = "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/ProjectASAP/ASAPController", rev = "cc18c9872bbaf0cadf43566892c3974ec9948eba" }
+asap-l2 = { git = "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/ProjectASAP/ASAPController", rev = "cc18c9872bbaf0cadf43566892c3974ec9948eba" }
+asap-sketch = { git = "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/ProjectASAP/ASAPController", rev = "cc18c9872bbaf0cadf43566892c3974ec9948eba" }
+asap-plan = { git = "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/ProjectASAP/ASAPController", rev = "cc18c9872bbaf0cadf43566892c3974ec9948eba" }
 
 [dev-dependencies]
 tokio = { version = "1", features = ["full", "test-util"] }

control_plane/src/emit/mod.rs

@@ -328,14 +328,14 @@ fn is_exact_accumulator(kind: &SummaryKind) -> bool {
 
 fn extract_from_node(node: &Rc<L4Node>) -> Option<SummaryKind> {
     match &node.expr {
-        SummaryExpr::SummaryAgg { sketch, .. } if !is_exact_accumulator(sketch) => {
-            Some(sketch.clone())
+        SummaryExpr::SummaryAgg { summary, .. } if !is_exact_accumulator(summary) => {
+            Some(summary.clone())
         }
         // An exact accumulator has no sketch family beneath it (its own
         // child is always a plain `Logical` leaf) — same as the old
         // `ExactAgg` case.
         SummaryExpr::SummaryAgg { .. } => None,
-        SummaryExpr::SummaryEstimate { sketch_input, .. } => extract_from_node(sketch_input),
+        SummaryExpr::SummaryEstimate { summary_input, .. } => extract_from_node(summary_input),
         SummaryExpr::SummaryMerge { children } => children.iter().find_map(extract_from_node),
         // Not surfaced by any `Bind*` path yet (gated on rules that
         // haven't landed — see `physical_expr.rs`'s module docs).

control_plane/src/intent_algebra/lower.rs

@@ -121,7 +121,7 @@ use crate::intent_algebra::column_resolution::{
 };
 use crate::intent_algebra::query_expr::{
     GroupKeys, L3Scalar, Predicate, ProjectItem as CProjectItem, QueryExpr as CQueryExpr,
-    QueryExprError, SortKey as CSortKey, Source, WindowKind as CWindowKind,
+    QueryExprError, Reduction, SortKey as CSortKey, Source, WindowKind as CWindowKind,
 };
 use crate::intent_algebra::relational::{AggFunc, QueryExpr as LQueryExpr};
 use crate::intent_algebra::schema::Schema;
@@ -270,6 +270,21 @@ pub fn convert(legacy: &LQueryExpr, schema: &Schema) -> Result<CQueryExpr, Conve
                 let item = &aggs[0];
                 let intents = agg_func_to_intents(&item.func, frequency_trigger);
                 if !intents.is_empty() {
+                    // Per-entity (issue ASAPController#163/#165): a single
+                    // intent, no `by`/`without` at all, whose intent is
+                    // inherently per-series or whose child is a range
+                    // window (`windowed`, computed above -- this repo's L3
+                    // shape keeps `Window` *above* the `Aggregate` rather
+                    // than a `TimeRange` child inside it, unlike
+                    // ASAPController's current canonical shape, but it's
+                    // the same structural marker for "this is a bare range
+                    // reduction with no grouping syntax to begin with").
+                    // `by.is_without()` is never true here -- `without` PromQL
+                    // grouping always resolves via the `without` bool above,
+                    // which only ever produces `GroupKeys::without` with a
+                    // concrete (possibly empty) exclusion list, and that
+                    // path already implies a genuine reduction regardless.
+                    let per_entity_base = by.is_empty() && !by.is_without();
                     let nodes: Vec<CQueryExpr> = match input.as_ref() {
                         LQueryExpr::Window {
                             duration,
@@ -284,30 +299,48 @@ pub fn convert(legacy: &LQueryExpr, schema: &Schema) -> Result<CQueryExpr, Conve
                             let win_child = convert(win_input, schema)?;
                             intents
                                 .into_iter()
-                                .map(|intent| CQueryExpr::Window {
-                                    kind: kind.clone(),
-                                    size: *duration,
-                                    slide: *slide,
-                                    child: Box::new(CQueryExpr::Aggregate {
-                                        by: by.clone(),
-                                        aggs: vec![intent],
-                                        output_names: Vec::new(),
-                                        having: None,
-                                        child: Box::new(win_child.clone()),
-                                    }),
+                                .map(|intent| {
+                                    let reduction = if per_entity_base
+                                        && (intent.is_per_series() || windowed)
+                                    {
+                                        Reduction::PerEntity
+                                    } else {
+                                        Reduction::Reduce(by.clone())
+                                    };
+                                    CQueryExpr::Window {
+                                        kind: kind.clone(),
+                                        size: *duration,
+                                        slide: *slide,
+                                        child: Box::new(CQueryExpr::Aggregate {
+                                            reduction,
+                                            aggs: vec![intent],
+                                            output_names: Vec::new(),
+                                            having: None,
+                                            child: Box::new(win_child.clone()),
+                                        }),
+                                    }
                                 })
                                 .collect()
                         }
                         other => {
                             let child = convert(other, schema)?;
                             intents
                                 .into_iter()
-                                .map(|intent| CQueryExpr::Aggregate {
-                                    by: by.clone(),
-                                    aggs: vec![intent],
-                                    output_names: Vec::new(),
-                                    having: None,
-                                    child: Box::new(child.clone()),
+                                .map(|intent| {
+                                    let reduction = if per_entity_base
+                                        && (intent.is_per_series() || windowed)
+                                    {
+                                        Reduction::PerEntity
+                                    } else {
+                                        Reduction::Reduce(by.clone())
+                                    };
+                                    CQueryExpr::Aggregate {
+                                        reduction,
+                                        aggs: vec![intent],
+                                        output_names: Vec::new(),
+                                        having: None,
+                                        child: Box::new(child.clone()),
+                                    }
                                 })
                                 .collect()
                         }
@@ -338,7 +371,11 @@ pub fn convert(legacy: &LQueryExpr, schema: &Schema) -> Result<CQueryExpr, Conve
                 .map(|h| resolve_expr(h, schema).map(Predicate))
                 .transpose()?;
             CQueryExpr::Aggregate {
-                by,
+                // Always a genuine reduction: this branch is multi-intent,
+                // HAVING-bearing, or a (structurally unreachable) unmapped
+                // AggFunc -- per-entity reductions are always the single,
+                // HAVING-less intent handled above.
+                reduction: Reduction::Reduce(by),
                 aggs: intents,
                 output_names: Vec::new(),
                 having,
@@ -372,7 +409,9 @@ pub fn convert(legacy: &LQueryExpr, schema: &Schema) -> Result<CQueryExpr, Conve
             // see `topk_bridge` module doc.
             let by: GroupKeys = resolve_group_keys_promql(by, schema)?.into();
             CQueryExpr::Aggregate {
-                by,
+                // A ranking always reduces (empty by ranks the whole
+                // input, never per-entity).
+                reduction: Reduction::Reduce(by),
                 aggs: vec![AggIntent::TopK {
                     k: *k as usize,
                     accuracy: AccuracyTarget::Epsilon(0.05),
@@ -745,8 +784,13 @@ mod tests {
     fn single_aggregate_folds_to_canonical_aggregate() {
         let legacy = agg(vec![], false, vec![agg_item("s", AggFunc::Sum)], src("m"));
         match convert_root(&legacy).unwrap() {
-            CQueryExpr::Aggregate { by, aggs, .. } => {
-                assert!(by.is_empty(), "no GROUP BY → empty `by`: {by:?}");
+            CQueryExpr::Aggregate {
+                reduction, aggs, ..
+            } => {
+                assert!(
+                    matches!(&reduction, Reduction::Reduce(k) if k.is_empty()),
+                    "no GROUP BY, unwindowed, non-per-series intent → global reduce: {reduction:?}"
+                );
                 assert!(matches!(aggs.as_slice(), [AggIntent::Sum { col: None }]));
             }
             other => panic!("expected Aggregate, got {other:?}"),
@@ -822,7 +866,9 @@ mod tests {
             src("trades"),
         );
         match convert_root(&legacy).unwrap() {
-            CQueryExpr::Aggregate { by, .. } => assert!(by.is_empty()),
+            CQueryExpr::Aggregate { reduction, .. } => {
+                assert!(matches!(&reduction, Reduction::Reduce(k) if k.is_empty()))
+            }
             other => panic!("expected Aggregate, got {other:?}"),
         }
     }
@@ -844,8 +890,8 @@ mod tests {
                 assert_eq!(kind, CWindowKind::Tumbling);
                 assert!(matches!(
                     *child,
-                    CQueryExpr::Aggregate { ref by, ref aggs, .. }
-                        if by.is_empty()
+                    CQueryExpr::Aggregate { ref reduction, ref aggs, .. }
+                        if matches!(reduction, Reduction::PerEntity)
                             && matches!(aggs.as_slice(), [AggIntent::Quantile { .. }])
                 ));
             }
@@ -939,8 +985,10 @@ mod tests {
             input: Box::new(src("m")),
         };
         match convert_root(&legacy).unwrap() {
-            CQueryExpr::Aggregate { by, aggs, .. } => {
-                assert!(by.is_empty());
+            CQueryExpr::Aggregate {
+                reduction, aggs, ..
+            } => {
+                assert!(matches!(&reduction, Reduction::Reduce(k) if k.is_empty()));
                 assert!(matches!(aggs.as_slice(), [AggIntent::TopK { k: 5, .. }]));
             }
             other => panic!("expected Aggregate, got {other:?}"),

control_plane/src/intent_algebra/mod.rs

@@ -113,8 +113,8 @@ pub use expr_ir::{ArithOp, ColumnRef, CompareOp, Expr, L2Expr, L3Expr, L3Scalar}
 pub use query_expr::{
     aggregate_output_schema, between, conjoin, label_filter_to_predicate, AtModifier, BinaryOpKind,
     BindingScope, DataModel, GroupKeys, GroupSide, InfoMatcher, JoinKind, LabelFilter, Predicate,
-    ProjectItem, QueryExpr, QueryExprError, SampleKind, SetOpKind, SortKey, Source, TimeShift,
-    VectorGrouping, VectorMatch, VectorMatchKind, WindowFuncKind, WindowKind,
+    ProjectItem, QueryExpr, QueryExprError, Reduction, SampleKind, SetOpKind, SortKey, Source,
+    TimeShift, VectorGrouping, VectorMatch, VectorMatchKind, WindowFuncKind, WindowKind,
 };
 pub use schema::{cse_reuse_is_legal, Column, ColumnId, CseError, DataType, Schema};
 

control_plane/src/intent_algebra/query_expr.rs

@@ -61,7 +61,7 @@ use asap_ir::intent_algebra::schema::ColumnId;
 pub use asap_ir::intent_algebra::{
     aggregate_output_schema, AtModifier, BinaryOpKind, BindingScope, DataModel, GroupKeys,
     GroupSide, InfoMatcher, JoinKind, Predicate, ProjectItem, QueryExpr, QueryExprError,
-    SampleKind, SetOpKind, SortKey, Source, TimeShift, VectorGrouping, VectorMatch,
+    Reduction, SampleKind, SetOpKind, SortKey, Source, TimeShift, VectorGrouping, VectorMatch,
     VectorMatchKind, WindowFuncKind, WindowKind,
 };
 pub use asap_ir::intent_algebra::{ArithOp, ColumnRef, CompareOp, Expr, L3Scalar};