refactor: align backend with merged Planner contracts for migration

control_plane/Cargo.toml

@@ -93,16 +93,16 @@ asap_types.workspace = true
 # scaffolding, unaware that `data_plane`'s `summary_executor.rs` in *this*
 # repo is a real one. Vendored locally instead of chased upstream -- see
 # `data_plane/src/query_engines/asap_query_engine/summary_exec.rs`.
-planner-types = { package = "asap-types", git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "cb50219c582d43f53ab77d3a595bd1ea4a9aa119" }
-asap-aware-mapping = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "cb50219c582d43f53ab77d3a595bd1ea4a9aa119" }
+planner-types = { package = "asap-types", git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "d0701dd4d4f0acb267b004c42ba30b2c9547ff7c" }
+asap-aware-mapping = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "d0701dd4d4f0acb267b004c42ba30b2c9547ff7c" }
 
 # L1 adoption (design-target-architecture.md Part B): the PromQL front
 # end itself, replacing control_plane's own query_parser/promql.rs.
 # Pinned via `rev`, not a floating branch reference. Same rev as
 # `planner-types`/`asap-aware-mapping` above -- these three MUST move
 # together (two revs of the same upstream repo's types in one workspace
 # resolve to distinct Rust types that won't unify).
-asap-frontend-promql = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "cb50219c582d43f53ab77d3a595bd1ea4a9aa119" }
+asap-frontend-promql = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "d0701dd4d4f0acb267b004c42ba30b2c9547ff7c" }
 
 [dev-dependencies]
 tokio = { version = "1", features = ["full", "test-util"] }

control_plane/src/asap_tier_implement.rs

@@ -134,7 +134,7 @@ fn collect_aggregate_roots<'a>(expr: &'a QueryExpr, out: &mut Vec<&'a QueryExpr>
         | QueryExpr::Sort { child, .. }
         | QueryExpr::Limit { child, .. }
         | QueryExpr::PromqlSubquery { child, .. } => collect_aggregate_roots(child, out),
-        QueryExpr::Concat { children } => {
+        QueryExpr::Concat { children, .. } => {
             for c in children {
                 collect_aggregate_roots(c, out);
             }

control_plane/src/emit/mod.rs

@@ -321,7 +321,8 @@ fn extract_from_node(node: &Rc<SummaryNode>) -> Option<SketchAlgorithm> {
         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 `deployment_expr.rs`'s module docs).
-        SummaryExpr::SummaryJoin { .. }
+        SummaryExpr::BinaryOp { .. }
+        | SummaryExpr::SummaryJoin { .. }
         | SummaryExpr::SummarySubtract { .. }
         | SummaryExpr::SummaryDelete { .. }
         | SummaryExpr::KeepPreAsap(_) => None,

control_plane/src/physical/allocator.rs

@@ -263,7 +263,10 @@ impl SketchAllocator {
             }
 
             // ── Merge — Backend ───────────────────────────────────────────
-            QueryExpr::Concat { children: inputs } => {
+            QueryExpr::Concat {
+                children: inputs,
+                discriminator_unique_key,
+            } => {
                 let children: Vec<PlanNode> = inputs
                     .into_iter()
                     .map(|inp| self.alloc_node(inp, budget))
@@ -272,6 +275,7 @@ impl SketchAllocator {
                 PlanNode {
                     expr: QueryExpr::Concat {
                         children: children.iter().map(|c| c.expr.clone()).collect(),
+                        discriminator_unique_key,
                     },
                     stage: PipelineStage::Backend,
                     mode: ExecutionMode::Passthrough,
@@ -952,6 +956,7 @@ mod tests {
     fn merge_goes_to_backend() {
         let expr = QueryExpr::Concat {
             children: vec![scan("a"), scan("b")],
+            discriminator_unique_key: None,
         };
         let node = alloc(unlimited(), expr);
         assert_eq!(node.stage, PipelineStage::Backend);

control_plane/src/physical/colored_dag/allocator.rs

@@ -177,6 +177,13 @@ impl ThreeStageWalker {
             // (scrape locality); `Ref` resolves through the lexical
             // scope map.
             SummaryExpr::KeepPreAsap(qe) => self.colour_logical(qe)?,
+            SummaryExpr::BinaryOp { lhs, rhs, .. } => {
+                for child in [lhs, rhs] {
+                    let (cid, _) = self.visit_l4node(child)?;
+                    self.dag.edges.push((id, cid));
+                }
+                StageId::Backend
+            }
 
             // ── SummaryAgg: always edge per design.md §6 batched-queries
             // table — true for both approximate sketches (the old

control_plane/src/physical/colored_dag/emitter.rs

@@ -113,7 +113,8 @@ fn classify(expr: &PhysicalExpr) -> NodeKind<'_> {
             SummaryExpr::SummaryAgg { .. } => NodeKind::Other,
             SummaryExpr::SummaryEstimate { query, .. } => NodeKind::SketchEstimate { query },
             SummaryExpr::SummaryMerge { .. } => NodeKind::SketchMerge,
-            SummaryExpr::SummaryJoin { .. }
+            SummaryExpr::BinaryOp { .. }
+            | SummaryExpr::SummaryJoin { .. }
             | SummaryExpr::SummarySubtract { .. }
             | SummaryExpr::SummaryDelete { .. } => NodeKind::Other,
         },
@@ -1125,7 +1126,7 @@ fn extract_edge_facts(qe: &planner_types::pre_asap::QueryExpr, edge: &mut EdgeSt
         | QE::Sort { child, .. }
         | QE::Limit { child, .. }
         | QE::PromqlSubquery { child, .. } => extract_edge_facts(child, edge),
-        QE::Concat { children } => {
+        QE::Concat { children, .. } => {
             for c in children {
                 extract_edge_facts(c, edge);
             }

control_plane/src/physical/colored_dag/tests.rs

@@ -109,7 +109,10 @@ fn sketch_agg_l4(
                 SketchKind::new(kind, params),
                 GroupingStrategy::default(),
             ),
-            col: ColumnRef::SampleValue,
+            input: planner_types::post_asap::SummaryUpdate {
+                item: None,
+                weight: planner_types::post_asap::SummaryInputExpr::Column(ColumnRef::SampleValue),
+            },
             reduction: Reduction::by(vec![]),
             grouping: GroupingStrategy::default(),
         },

control_plane/src/physical/compiler.rs

@@ -42,7 +42,7 @@ use crate::query_plan::{
 use crate::types_v2::AccuracyTarget;
 use planner_types::pre_asap::Source;
 
-pub const PLANNER_REVISION: &str = "cb50219c582d43f53ab77d3a595bd1ea4a9aa119";
+pub const PLANNER_REVISION: &str = "d0701dd4d4f0acb267b004c42ba30b2c9547ff7c";
 
 #[derive(Debug, Clone)]
 pub struct PlanningQuery {
@@ -2009,6 +2009,10 @@ fn summary_agg_metric(node: &SummaryNode) -> Option<String> {
                 walk(right, metrics);
             }
             SummaryExpr::SummaryDelete { summary_input, .. } => walk(summary_input, metrics),
+            SummaryExpr::BinaryOp { lhs, rhs, .. } => {
+                walk(lhs, metrics);
+                walk(rhs, metrics);
+            }
         }
     }
     let mut metrics = BTreeSet::new();
@@ -2366,7 +2370,8 @@ fn collect_selected_materializations(
                     parameters: Value::Object(Default::default()),
                 });
             }
-            SummaryExpr::KeepPreAsap(_)
+            SummaryExpr::BinaryOp { .. }
+            | SummaryExpr::KeepPreAsap(_)
             | SummaryExpr::SummaryAgg { .. }
             | SummaryExpr::SummaryJoin { .. }
             | SummaryExpr::SummarySubtract { .. }

control_plane/src/physical/planner.rs

@@ -351,7 +351,7 @@ fn plan_node(expr: &QueryExpr, config: &PhysicalPlannerConfig) -> PhysicalNode {
         // (`intent_algebra::lower`), so the `HashAggregate { keys }` this
         // arm used to build now comes straight out of the `Aggregate`
         // arm above.
-        QueryExpr::Concat { children } => {
+        QueryExpr::Concat { children, .. } => {
             let children: Vec<PhysicalNode> =
                 children.iter().map(|c| plan_node(c, config)).collect();
             let sketch_type = children
@@ -774,6 +774,7 @@ mod tests {
             having: None,
             child: QueryExpr::Concat {
                 children: vec![windowed_agg(default_frequency(), 60, "requests")],
+                discriminator_unique_key: None,
             }
             .into(),
         };

control_plane/src/physical/post_asap/cost_model.rs

@@ -252,6 +252,7 @@ impl CostModel for ControlPlaneCostModel {
             .min_by(|left, right| left.1 .0.total_cmp(&right.1 .0))
             .map(
                 |(framework, physical_cost)| CompleteSummaryCandidateEstimate {
+                    physical_plan_id: None,
                     cost: Cost(lifecycle_cost + physical_cost.0),
                     window_frameworks: vec![Some(framework.clone()); deployments.len()],
                     window_accuracy_guarantee: Some(

control_plane/src/physical/post_asap/tests.rs

@@ -119,6 +119,7 @@ fn node_is_archive(node: &Rc<SummaryNode>) -> bool {
             node_is_archive(left) || node_is_archive(right)
         }
         SummaryExpr::SummaryDelete { summary_input, .. } => node_is_archive(summary_input),
+        SummaryExpr::BinaryOp { lhs, rhs, .. } => node_is_archive(lhs) || node_is_archive(rhs),
     }
 }
 
@@ -774,7 +775,16 @@ fn phase_b_e2e_topk_well_formed() {
     let accuracy = AccuracyTarget::Epsilon(0.05);
     let expr = crate::query_parser::parse_query_expr_canonical(query, accuracy.clone())
         .expect("TopK parses");
-    assert!(bind_query_expr(&expr, accuracy).is_err());
+    let bound = bind_query_expr(&expr, accuracy);
+    assert!(
+        bound.is_err()
+            || matches!(
+                &bound,
+                Ok(PhysicalExpr::Committed(PostAsapPlan::Summary(node)))
+                    if matches!(node.expr, SummaryExpr::KeepPreAsap(_))
+            ),
+        "unevidenced TopK must remain exact or unavailable: {bound:?}"
+    );
 }
 
 /// Archive-only routing through the full L1→L3→L4 pipeline. Asserts the

control_plane/src/query_parser/mod.rs

@@ -147,7 +147,7 @@ fn root_scan_schema(qe: &QueryExpr) -> Option<&planner_types::pre_asap::Schema>
         | QueryExpr::Sort { child, .. }
         | QueryExpr::Limit { child, .. }
         | QueryExpr::PromqlSubquery { child, .. } => root_scan_schema(child),
-        QueryExpr::Concat { children } => children.iter().find_map(root_scan_schema),
+        QueryExpr::Concat { children, .. } => children.iter().find_map(root_scan_schema),
         QueryExpr::Join { left, right, .. }
         | QueryExpr::SetOp { left, right, .. }
         | QueryExpr::BinaryOp {
@@ -257,7 +257,7 @@ impl QeCollector {
                 self.visit(child, schema);
             }
             QueryExpr::Dedup { child, .. } => self.visit(child, schema),
-            QueryExpr::Concat { children } => {
+            QueryExpr::Concat { children, .. } => {
                 for c in children {
                     self.visit(c, schema);
                 }

control_plane/src/query_plan.rs

@@ -331,9 +331,11 @@ where
         self.next_id += 1;
         self.seen.insert(identity, id);
         let physical = match &node.expr {
-            SummaryExpr::KeepPreAsap(_) => QueryPlanNode::ExactFallback {
-                reason: "post-ASAP node requires exact execution".into(),
-            },
+            SummaryExpr::BinaryOp { .. } | SummaryExpr::KeepPreAsap(_) => {
+                QueryPlanNode::ExactFallback {
+                    reason: "post-ASAP node requires exact execution".into(),
+                }
+            }
             SummaryExpr::SummaryAgg {
                 family,
                 reduction,

control_plane/src/workload.rs

@@ -194,7 +194,7 @@ fn collect_agg_intents(expr: &planner_types::pre_asap::QueryExpr, out: &mut Vec<
         | QueryExpr::TimeRange { child, .. }
         | QueryExpr::TimeShift { child, .. }
         | QueryExpr::SQLWindowFunc { child, .. } => collect_agg_intents(child, out),
-        QueryExpr::Concat { children } => {
+        QueryExpr::Concat { children, .. } => {
             for child in children {
                 collect_agg_intents(child, out);
             }

crates/asap_types/Cargo.toml

@@ -30,4 +30,4 @@ xxhash-rust = { version = "0.8", features = ["xxh64"] }
 # exactly (`control_plane/Cargo.toml`) -- two different revs of the same
 # git dependency in one workspace resolve to two distinct Rust types that
 # won't unify.
-planner-types = { package = "asap-types", git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "cb50219c582d43f53ab77d3a595bd1ea4a9aa119" }
+planner-types = { package = "asap-types", git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "d0701dd4d4f0acb267b004c42ba30b2c9547ff7c" }

data_plane/Cargo.toml

@@ -38,7 +38,7 @@ control_plane = { path = "../control_plane" }
 # reduction: Reduction, .. }`) are `pre_asap` types, in the same crate now
 # (not a separate `asap-ir` import). Query serving consumes the compiled
 # QueryPlan; these types are used at physical-plan compilation boundaries.
-planner-types = { package = "asap-types", git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "cb50219c582d43f53ab77d3a595bd1ea4a9aa119" }
+planner-types = { package = "asap-types", git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "d0701dd4d4f0acb267b004c42ba30b2c9547ff7c" }
 
 # Shared external (workspace)
 serde.workspace = true

data_plane/src/query_engines/asap_query_engine/post_asap_planner.rs

@@ -386,7 +386,7 @@ fn query_expr_contains_time_range(qe: &planner_types::pre_asap::QueryExpr) -> bo
         | QueryExpr::PromqlSubquery { child, .. }
         | QueryExpr::TimeShift { child, .. }
         | QueryExpr::SQLWindowFunc { child, .. } => query_expr_contains_time_range(child),
-        QueryExpr::Concat { children } => children.iter().any(query_expr_contains_time_range),
+        QueryExpr::Concat { children, .. } => children.iter().any(query_expr_contains_time_range),
         QueryExpr::Join { left, right, .. }
         | QueryExpr::SetOp { left, right, .. }
         | QueryExpr::BinaryOp {
@@ -415,7 +415,7 @@ fn query_expr_contains_rate(qe: &planner_types::pre_asap::QueryExpr) -> bool {
         | QueryExpr::TimeRange { child, .. }
         | QueryExpr::TimeShift { child, .. }
         | QueryExpr::SQLWindowFunc { child, .. } => query_expr_contains_rate(child),
-        QueryExpr::Concat { children } => children.iter().any(query_expr_contains_rate),
+        QueryExpr::Concat { children, .. } => children.iter().any(query_expr_contains_rate),
         QueryExpr::Join { left, right, .. }
         | QueryExpr::SetOp { left, right, .. }
         | QueryExpr::BinaryOp {
@@ -441,7 +441,7 @@ fn query_expr_has_filter(qe: &planner_types::pre_asap::QueryExpr) -> bool {
         | QueryExpr::TimeRange { child, .. }
         | QueryExpr::TimeShift { child, .. }
         | QueryExpr::SQLWindowFunc { child, .. } => query_expr_has_filter(child),
-        QueryExpr::Concat { children } => children.iter().any(query_expr_has_filter),
+        QueryExpr::Concat { children, .. } => children.iter().any(query_expr_has_filter),
         QueryExpr::Join { left, right, .. }
         | QueryExpr::SetOp { left, right, .. }
         | QueryExpr::BinaryOp {
@@ -472,7 +472,7 @@ fn query_expr_max_time_range_ms(qe: &planner_types::pre_asap::QueryExpr) -> Opti
         | QueryExpr::Limit { child, .. }
         | QueryExpr::TimeShift { child, .. }
         | QueryExpr::SQLWindowFunc { child, .. } => child_max(child),
-        QueryExpr::Concat { children } => children
+        QueryExpr::Concat { children, .. } => children
             .iter()
             .filter_map(query_expr_max_time_range_ms)
             .max(),

data_plane/src/query_engines/asap_query_engine/summary_exec.rs

@@ -154,11 +154,18 @@ pub fn execute<E: SummaryExecutor>(
         SummaryExpr::SummaryAgg {
             child,
             family,
-            col,
+            input,
             reduction,
             ..
         } => {
-            let tagged = exec.find_candidates(family, col, reduction, child)?;
+            let col = match input.item.as_ref().unwrap_or(&input.weight) {
+                planner_types::post_asap::SummaryInputExpr::Column(col) => col.clone(),
+                planner_types::post_asap::SummaryInputExpr::Constant(value) if *value == 1.0 => {
+                    ColumnRef::SampleValue
+                }
+                _ => return Err(ExecError::NotYetSupported("summary update expression")),
+            };
+            let tagged = exec.find_candidates(family, &col, reduction, child)?;
             if tagged.is_empty() {
                 return Err(ExecError::NoCandidates);
             }
@@ -233,6 +240,7 @@ pub fn execute<E: SummaryExecutor>(
         }
 
         SummaryExpr::SummaryJoin { .. } => Err(ExecError::NotYetSupported("SummaryJoin")),
+        SummaryExpr::BinaryOp { .. } => Err(ExecError::NotYetSupported("BinaryOp")),
         SummaryExpr::SummarySubtract { .. } => Err(ExecError::NotYetSupported("SummarySubtract")),
         SummaryExpr::SummaryDelete { .. } => Err(ExecError::NotYetSupported("SummaryDelete")),
     }
@@ -325,7 +333,12 @@ mod tests {
             expr: SummaryExpr::SummaryAgg {
                 child,
                 family,
-                col: ColumnRef::SampleValue,
+                input: planner_types::post_asap::SummaryUpdate {
+                    item: None,
+                    weight: planner_types::post_asap::SummaryInputExpr::Column(
+                        ColumnRef::SampleValue,
+                    ),
+                },
                 reduction,
                 grouping: planner_types::post_asap::GroupingStrategy::default(),
             },