Execute ClickHouse joins in the shared query DAG

control_plane/Cargo.toml

@@ -91,18 +91,18 @@ 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 = "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/ProjectASAP/ASAPPlanner", rev = "26e580710cf9b44b282598719d6c5ea9a9cc62be" }
-asap-aware-mapping = { git = "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/ProjectASAP/ASAPPlanner", rev = "26e580710cf9b44b282598719d6c5ea9a9cc62be" }
+planner-types = { package = "asap-types", git = "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/ProjectASAP/ASAPPlanner", rev = "f99237d" }
+asap-aware-mapping = { git = "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/ProjectASAP/ASAPPlanner", rev = "f99237d" }
 
 # 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 = "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/ProjectASAP/ASAPPlanner", rev = "26e580710cf9b44b282598719d6c5ea9a9cc62be" }
-asap-frontend-metricsql = { git = "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/ProjectASAP/ASAPPlanner", rev = "26e580710cf9b44b282598719d6c5ea9a9cc62be" }
-asap-frontend-sql = { git = "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/ProjectASAP/ASAPPlanner", rev = "26e580710cf9b44b282598719d6c5ea9a9cc62be" }
+asap-frontend-promql = { git = "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/ProjectASAP/ASAPPlanner", rev = "f99237d" }
+asap-frontend-metricsql = { git = "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/ProjectASAP/ASAPPlanner", rev = "f99237d" }
+asap-frontend-sql = { git = "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/ProjectASAP/ASAPPlanner", rev = "f99237d" }
 
 [dev-dependencies]
 tokio = { version = "1", features = ["full", "test-util"] }

control_plane/examples/calibration_candidates.rs

@@ -74,6 +74,16 @@ fn planner_forest(queries: &[control_plane::physical::compiler::PlanningQuery])
                 vec![outer, inner],
                 json!({"key_debug":format!("{key:?}"),"family_debug":format!("{family:?}")}),
             ),
+            SummaryExpr::RelationalJoin {
+                left,
+                right,
+                kind,
+                pred,
+            } => (
+                "RelationalJoin",
+                vec![left, right],
+                json!({"kind_debug":format!("{kind:?}"),"predicate_debug":format!("{pred:?}")}),
+            ),
             SummaryExpr::SummarySubtract { left, right } => {
                 ("SummarySubtract", vec![left, right], json!({}))
             }

control_plane/examples/offline_planner_replay.rs

@@ -73,6 +73,11 @@ fn inspect(
             left: lhs,
             right: rhs,
         }
+        | SummaryExpr::RelationalJoin {
+            left: lhs,
+            right: rhs,
+            ..
+        }
         | SummaryExpr::BinaryOp { lhs, rhs, .. } => {
             inspect(lhs, model, seen, states, raw);
             inspect(rhs, model, seen, states, raw);

control_plane/src/emit/mod.rs

@@ -326,6 +326,7 @@ fn extract_from_node(node: &Rc<SummaryNode>) -> Option<SketchAlgorithm> {
         // Not surfaced by any `Bind*` path yet (gated on rules that
         // haven't landed — see `deployment_expr.rs`'s module docs).
         SummaryExpr::BinaryOp { .. }
+        | SummaryExpr::RelationalJoin { .. }
         | SummaryExpr::SummaryJoin { .. }
         | SummaryExpr::SummarySubtract { .. }
         | SummaryExpr::SummaryDelete { .. }

control_plane/src/physical/colored_dag/allocator.rs

@@ -184,6 +184,13 @@ impl ThreeStageWalker {
                 }
                 StageId::Backend
             }
+            SummaryExpr::RelationalJoin { left, right, .. } => {
+                for child in [left, right] {
+                    let (cid, _) = self.visit_l4node(child)?;
+                    self.dag.edges.push((id, cid));
+                }
+                StageId::Backend
+            }
             SummaryExpr::ValueOperation { child, timing, .. } => {
                 let (cid, child_stage) = self.visit_l4node(child)?;
                 self.dag.edges.push((id, cid));

control_plane/src/physical/colored_dag/emitter.rs

@@ -117,6 +117,7 @@ fn classify(expr: &PhysicalExpr) -> NodeKind<'_> {
             SummaryExpr::SummaryEstimate { query, .. } => NodeKind::SketchEstimate { query },
             SummaryExpr::SummaryMerge { .. } => NodeKind::SketchMerge,
             SummaryExpr::BinaryOp { .. }
+            | SummaryExpr::RelationalJoin { .. }
             | SummaryExpr::CandidateTopK { .. }
             | SummaryExpr::ValueOperation { .. }
             | SummaryExpr::SummaryJoin { .. }

control_plane/src/physical/compiler.rs

@@ -2885,6 +2885,7 @@ fn summary_agg_metric(node: &SummaryNode) -> Option<String> {
                 inner: right,
                 ..
             }
+            | SummaryExpr::RelationalJoin { left, right, .. }
             | SummaryExpr::SummarySubtract { left, right }
             | SummaryExpr::BinaryOp {
                 lhs: left,
@@ -3032,6 +3033,7 @@ fn requires_exact_erp_fallback(
                 walk(inner, out);
             }
             SummaryExpr::SummarySubtract { left, right }
+            | SummaryExpr::RelationalJoin { left, right, .. }
             | SummaryExpr::BinaryOp {
                 lhs: left,
                 rhs: right,
@@ -3474,12 +3476,12 @@ fn select_lifecycle(
             reason: "latest ASAPPlanner selected no window framework from the supplied physical evidence".into(),
         })?;
     Ok(PlannerPhysicalSelection {
-        window_implementation_id: plan.selected_physical_plan_id.clone().ok_or_else(|| {
-            CompileError::Lifecycle {
+        window_implementation_id: plan.selected_window_implementation_id.clone().ok_or_else(
+            || CompileError::Lifecycle {
                 query_id: query.query_id.clone(),
                 reason: "Planner returned no concrete window implementation identity".into(),
-            }
-        })?,
+            },
+        )?,
         expected_reads: plan.expected_reads.ok_or_else(|| CompileError::Lifecycle {
             query_id: query.query_id.clone(),
             reason: "missing joint read demand".into(),
@@ -3845,6 +3847,10 @@ fn collect_selected_materializations(
             SummaryExpr::ValueOperation { child, .. } => {
                 walk(child, readout, composable, grouping.clone(), selected)?;
             }
+            SummaryExpr::RelationalJoin { left, right, .. } => {
+                walk(left, readout, composable, grouping.clone(), selected)?;
+                walk(right, readout, composable, grouping.clone(), selected)?;
+            }
             SummaryExpr::BinaryOp { lhs, rhs, .. }
                 if composable || crate::query_plan::exact_value_executable(node) =>
             {

control_plane/src/physical/post_asap/tests.rs

@@ -120,6 +120,7 @@ fn node_is_archive(node: &Rc<SummaryNode>) -> bool {
             candidates, values, ..
         } => node_is_archive(candidates) || node_is_archive(values),
         SummaryExpr::SummarySubtract { left, right }
+        | SummaryExpr::RelationalJoin { left, right, .. }
         | SummaryExpr::BinaryOp {
             lhs: left,
             rhs: right,

control_plane/src/query_plan.rs

@@ -636,6 +636,14 @@ pub struct ExternalExactRequest {
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
 #[serde(tag = "op", rename_all = "snake_case", deny_unknown_fields)]
 pub enum QueryPlanNode {
+    RelationalJoin {
+        inputs: [QueryNodeId; 2],
+        join_kind: planner_types::pre_asap::JoinKind,
+        pred: serde_json::Value,
+        left_schema: planner_types::post_asap::SummarySchema,
+        right_schema: planner_types::post_asap::SummarySchema,
+        output_schema: planner_types::post_asap::SummarySchema,
+    },
     Relational {
         input: QueryNodeId,
         /// Serialized planner-owned operation. Keeping the wire form here makes
@@ -699,7 +707,7 @@ impl QueryPlanNode {
             Self::Scalar { .. } | Self::ReadMaterialization { .. } | Self::ExactFallback { .. } => {
                 &[]
             }
-            Self::Binary { inputs, .. } => inputs,
+            Self::Binary { inputs, .. } | Self::RelationalJoin { inputs, .. } => inputs,
             Self::ReduceSum { input, .. }
             | Self::Relational { input, .. }
             | Self::SummaryEstimate { input, .. }
@@ -806,7 +814,8 @@ where
                     }
                 }
                 QueryPlanNode::CandidateTopK { inputs, .. }
-                | QueryPlanNode::Binary { inputs, .. } => {
+                | QueryPlanNode::Binary { inputs, .. }
+                | QueryPlanNode::RelationalJoin { inputs, .. } => {
                     for input in inputs {
                         *input = remap[input];
                     }
@@ -872,6 +881,26 @@ where
         }
 
         let physical = match &node.expr {
+            SummaryExpr::RelationalJoin {
+                left,
+                right,
+                kind,
+                pred,
+            } if self.preserve_relational => QueryPlanNode::RelationalJoin {
+                inputs: [self.lower(left)?, self.lower(right)?],
+                join_kind: kind.clone(),
+                pred: serde_json::to_value(pred).map_err(|error| {
+                    QueryPlanError::Invalid(format!(
+                        "cannot serialize relational join predicate: {error}"
+                    ))
+                })?,
+                left_schema: left.schema.clone(),
+                right_schema: right.schema.clone(),
+                output_schema: node.schema.clone(),
+            },
+            SummaryExpr::RelationalJoin { .. } => QueryPlanNode::ExactFallback {
+                reason: "read-time relational join requires the relational compiler".into(),
+            },
             SummaryExpr::ValueOperation {
                 child, operation, ..
             } if self.preserve_relational

control_plane/src/query_plan/logical.rs

@@ -1289,6 +1289,10 @@ pub fn materialization_candidate_keys(
                 visit(original, lhs, keys)?;
                 visit(original, rhs, keys)?;
             }
+            SummaryExpr::RelationalJoin { left, right, .. } => {
+                visit(original, left, keys)?;
+                visit(original, right, keys)?;
+            }
             SummaryExpr::CandidateTopK {
                 candidates, values, ..
             } => {

crates/asap_types/Cargo.toml

@@ -32,4 +32,4 @@ sha2 = "0.10"
 # 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 = "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/ProjectASAP/ASAPPlanner", rev = "26e580710cf9b44b282598719d6c5ea9a9cc62be" }
+planner-types = { package = "asap-types", git = "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/ProjectASAP/ASAPPlanner", rev = "f99237d" }

data_plane/Cargo.toml

@@ -38,8 +38,8 @@ 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 = "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/ProjectASAP/ASAPPlanner", rev = "26e580710cf9b44b282598719d6c5ea9a9cc62be" }
-asap-frontend-metricsql = { git = "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/ProjectASAP/ASAPPlanner", rev = "26e580710cf9b44b282598719d6c5ea9a9cc62be" }
+planner-types = { package = "asap-types", git = "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/ProjectASAP/ASAPPlanner", rev = "f99237d" }
+asap-frontend-metricsql = { git = "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/ProjectASAP/ASAPPlanner", rev = "f99237d" }
 
 # Shared external (workspace)
 serde.workspace = true
@@ -130,7 +130,7 @@ crc32fast = "1.4"
 # none of them.
 
 [dev-dependencies]
-asap-aware-mapping = { git = "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/ProjectASAP/ASAPPlanner", rev = "26e580710cf9b44b282598719d6c5ea9a9cc62be" }
+asap-aware-mapping = { git = "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/ProjectASAP/ASAPPlanner", rev = "f99237d" }
 tempfile = "3.20.0"
 criterion = { version = "0.5", features = ["html_reports"] }
 tokio-tungstenite = "0.21"