From 3dda28032c2a2badb64bf5358e0f40fef9918227 Mon Sep 17 00:00:00 2001 From: Zeying Zhu <50204836+zzylol@users.noreply.github.com> Date: Thu, 10 Sep 2026 13:41:57 -0400 Subject: [PATCH 1/3] feat(sql): plan joins over summary children (#370) * Plan relational joins over summary children * Test relational join safety boundaries * Validate relational join input boundaries * Classify relational joins in DAG viewer * fix: finalize exact state as readable values * fix: export validated edge data states --- crates/asap-aware-mapping/src/replacement.rs | 136 ++++++++++++++-- .../src/summary_maintenance_cost/estimator.rs | 17 +- .../src/summary_maintenance_cost/model.rs | 3 + .../src/summary_maintenance_cost/window.rs | 1 + .../src/summary_maintenance_dag_export.rs | 5 + .../src/summary_maintenance_lifecycle.rs | 5 + .../tests/sql_to_post_asap.rs | 151 +++++++++++++++++- crates/types/src/dag_export.rs | 11 ++ crates/types/src/post_asap/cse.rs | 19 +++ crates/types/src/post_asap/executable_dag.rs | 29 +++- .../src/post_asap/execution_data_state.rs | 21 ++- crates/types/src/post_asap/expr.rs | 13 +- tools/dag-viewer/node-style.js | 1 + 13 files changed, 387 insertions(+), 25 deletions(-) diff --git a/crates/asap-aware-mapping/src/replacement.rs b/crates/asap-aware-mapping/src/replacement.rs index 03da4fa8..5ba17f88 100644 --- a/crates/asap-aware-mapping/src/replacement.rs +++ b/crates/asap-aware-mapping/src/replacement.rs @@ -360,8 +360,10 @@ use asap_types::post_asap::{AccuracyError, CompositionOperator, GuaranteeSource, use asap_types::pre_asap::agg_intent::{agg_is_mergeable, AggIntent}; use asap_types::pre_asap::cse::{share_common_subtrees, structural_hash, HashCache}; use asap_types::pre_asap::expr_ir::ColumnRef; -use asap_types::pre_asap::query_expr::{BinaryOpKind, QueryExpr, QueryExprError, Reduction}; -use asap_types::pre_asap::schema::Schema; +use asap_types::pre_asap::query_expr::{ + BinaryOpKind, Predicate, QueryExpr, QueryExprError, Reduction, +}; +use asap_types::pre_asap::schema::{ColumnId, Schema}; use asap_types::types::AccuracyTarget; use asap_types::workload::{QueryRecurrence, QueryWorkload, RepeatedDemand}; use std::rc::Rc; @@ -1741,8 +1743,8 @@ fn realize_binary( return Ok(None); } - lhs_node = finalize_exact_accumulator(lhs_node); - rhs_node = finalize_exact_accumulator(rhs_node); + lhs_node = finalize_exact_accumulator(lhs_node, lhs)?; + rhs_node = finalize_exact_accumulator(rhs_node, rhs)?; let guarantee = [lhs_node.guarantee.as_ref(), rhs_node.guarantee.as_ref()] .into_iter() @@ -1769,7 +1771,10 @@ fn realize_binary( /// Put an explicit read boundary between maintained exact state and a /// query-time value consumer. Approximate summaries must already carry a /// `SummaryEstimate`, so they deliberately do not pass this predicate. -fn finalize_exact_accumulator(node: Rc) -> Rc { +fn finalize_exact_accumulator( + node: Rc, + logical_output: &QueryExpr, +) -> Result, ImplementError> { let is_exact_state = matches!( node.expr, SummaryExpr::SummaryAgg { @@ -1778,11 +1783,15 @@ fn finalize_exact_accumulator(node: Rc) -> Rc { } ); if !is_exact_state { - return node; + return Ok(node); } - let schema = node.schema.clone(); + // The child edge carries accumulator state, while this explicit read + // boundary produces the logical operator's ordinary values. Preserve the + // canonical pre-ASAP output types instead of leaking ExactAggregate into + // query-time operators that follow this node. + let schema = lift(&logical_output.output_schema()?); let guarantee = node.guarantee.clone(); - Rc::new(SummaryNode { + Ok(Rc::new(SummaryNode { expr: SummaryExpr::ValueOperation { child: node, operation: ValueOperation::FinalizeExactAccumulator, @@ -1790,7 +1799,7 @@ fn finalize_exact_accumulator(node: Rc) -> Rc { }, schema, guarantee, - }) + })) } fn is_supported_exact_binary(root: &QueryExpr) -> bool { @@ -1936,11 +1945,10 @@ pub(crate) fn construct_summary_with( allocation, )?; if is_counter_weighted_topk(intent, child) { - let values = finalize_exact_accumulator(realize_child_with( + let values = finalize_exact_accumulator( + realize_child_with(child, models, Some(&AccuracyTarget::Exact))?, child, - models, - Some(&AccuracyTarget::Exact), - )?); + )?; if !values .guarantee .as_ref() @@ -3621,6 +3629,48 @@ pub struct GlobalSelection<'a> { materialized: RefCell>>, } +fn normalize_cross_input_equi_predicate( + pred: &Predicate, + left_width: usize, + total_width: usize, +) -> Option { + let QueryExpr::Compare { + left, + op: asap_types::pre_asap::CompareOpKind::Eq, + right, + } = pred.0.as_ref() + else { + return None; + }; + let (QueryExpr::Column(left_id), QueryExpr::Column(right_id)) = (left.as_ref(), right.as_ref()) + else { + return None; + }; + let is_left = |id: ColumnId| id < left_width; + let is_right = |id: ColumnId| left_width <= id && id < total_width; + let (left_id, right_id) = if is_left(*left_id) && is_right(*right_id) { + (*left_id, *right_id) + } else if is_right(*left_id) && is_left(*right_id) { + (*right_id, *left_id) + } else { + return None; + }; + Some(Predicate(Rc::new(QueryExpr::Compare { + left: Rc::new(QueryExpr::Column(left_id)), + op: asap_types::pre_asap::CompareOpKind::Eq, + right: Rc::new(QueryExpr::Column(right_id)), + }))) +} + +fn relational_join_guarantee( + left: Option<&ResultGuarantee>, + right: Option<&ResultGuarantee>, +) -> Option { + left.zip(right) + .filter(|(left, right)| left.is_exact() && right.is_exact()) + .map(|_| ResultGuarantee::exact("RelationalJoin over exact inputs")) +} + impl<'a> GlobalSelection<'a> { /// Every selected group, in discovery order. pub fn groups(&self) -> impl Iterator> { @@ -3693,6 +3743,38 @@ impl<'a> GlobalSelection<'a> { &self, target: &Rc, ) -> Result, ImplementError> { + if let QueryExpr::Join { + left, + right, + kind, + pred, + } = target.as_ref() + { + let left_width = left.output_schema()?.columns.len(); + let total_width = left_width + right.output_schema()?.columns.len(); + let normalized_pred = matches!(kind, asap_types::pre_asap::JoinKind::Inner) + .then(|| normalize_cross_input_equi_predicate(pred, left_width, total_width)) + .flatten(); + let Some(pred) = normalized_pred else { + return keep_pre_asap(target); + }; + let left = self.materialize_inner(left)?; + let right = self.materialize_inner(right)?; + let guarantee = + relational_join_guarantee(left.guarantee.as_ref(), right.guarantee.as_ref()); + let node = Rc::new(SummaryNode { + expr: SummaryExpr::RelationalJoin { + left, + right, + kind: kind.clone(), + pred, + }, + schema: lift(&target.output_schema()?), + guarantee, + }); + validate_execution_data_states_at(&node, ExecutionDataState::READ_ROWS)?; + return Ok(node); + } let (child_target, operation) = match target.as_ref() { QueryExpr::Project { cols, @@ -5217,6 +5299,34 @@ mod tests { use asap_types::types::AccuracyTarget; use std::collections::HashMap; + fn equi_pred(left: ColumnId, right: ColumnId) -> Predicate { + Predicate(Rc::new(QueryExpr::Compare { + left: Rc::new(QueryExpr::Column(left)), + op: asap_types::pre_asap::CompareOpKind::Eq, + right: Rc::new(QueryExpr::Column(right)), + })) + } + + #[test] + fn relational_join_predicate_requires_and_normalizes_cross_input_columns() { + let forward = normalize_cross_input_equi_predicate(&equi_pred(1, 3), 2, 4) + .expect("left-to-right equality"); + let reverse = normalize_cross_input_equi_predicate(&equi_pred(3, 1), 2, 4) + .expect("right-to-left equality"); + assert_eq!(forward, reverse, "reverse equality must be canonicalized"); + assert!(normalize_cross_input_equi_predicate(&equi_pred(0, 1), 2, 4).is_none()); + assert!(normalize_cross_input_equi_predicate(&equi_pred(0, 4), 2, 4).is_none()); + } + + #[test] + fn relational_join_is_exact_only_when_both_inputs_are_exact() { + let exact = ResultGuarantee::exact("test exact input"); + assert!(relational_join_guarantee(Some(&exact), Some(&exact)) + .is_some_and(|guarantee| guarantee.is_exact())); + assert!(relational_join_guarantee(Some(&exact), None).is_none()); + assert!(relational_join_guarantee(None, Some(&exact)).is_none()); + } + fn eps(e: f64) -> AccuracyTarget { AccuracyTarget::Epsilon(e) } diff --git a/crates/asap-aware-mapping/src/summary_maintenance_cost/estimator.rs b/crates/asap-aware-mapping/src/summary_maintenance_cost/estimator.rs index b2fba4ba..eb714a1d 100644 --- a/crates/asap-aware-mapping/src/summary_maintenance_cost/estimator.rs +++ b/crates/asap-aware-mapping/src/summary_maintenance_cost/estimator.rs @@ -38,6 +38,7 @@ pub(super) fn estimate_heterogeneous_summary( } } SummaryExpr::SummarySubtract { left, right } + | SummaryExpr::RelationalJoin { left, right, .. } | SummaryExpr::BinaryOp { lhs: left, rhs: right, @@ -270,7 +271,12 @@ pub(super) fn estimate_heterogeneous_summary( return Ok(()); } match &node.expr { - SummaryExpr::BinaryOp { lhs, rhs, .. } => { + SummaryExpr::BinaryOp { lhs, rhs, .. } + | SummaryExpr::RelationalJoin { + left: lhs, + right: rhs, + .. + } => { let operation = summary_operation_evidence(node, evidence)?.resource(); *cpu_ops += evaluation_count as f64 * validated_operator_executions("exact_binary", operation)? as f64 @@ -443,6 +449,7 @@ pub(super) fn estimate_heterogeneous_summary( .for_each(|child| collect_aggs(child, seen, out)); } SummaryExpr::SummarySubtract { left, right } + | SummaryExpr::RelationalJoin { left, right, .. } | SummaryExpr::BinaryOp { lhs: left, rhs: right, @@ -649,6 +656,7 @@ fn validate_summary_edges_and_physical_ids( children.iter().map(|child| child.as_ref()).collect() } SummaryExpr::SummarySubtract { left, right } + | SummaryExpr::RelationalJoin { left, right, .. } | SummaryExpr::BinaryOp { lhs: left, rhs: right, @@ -825,6 +833,7 @@ pub(super) fn estimate_transient_liveness( children.iter().map(|child| child.as_ref()).collect() } SummaryExpr::SummarySubtract { left, right } + | SummaryExpr::RelationalJoin { left, right, .. } | SummaryExpr::BinaryOp { lhs: left, rhs: right, @@ -881,6 +890,7 @@ pub(super) fn estimate_transient_liveness( .ok_or(AnalyticalCostError::MissingOrStale("summary_join")), SummaryExpr::SummaryMerge { .. } | SummaryExpr::BinaryOp { .. } + | SummaryExpr::RelationalJoin { .. } | SummaryExpr::CandidateTopK { .. } | SummaryExpr::ValueOperation { .. } | SummaryExpr::SummarySubtract { .. } @@ -954,6 +964,7 @@ pub(super) fn evidence_nodes(root: &SummaryNode) -> (Vec<&SummaryNode>, Vec<&Sum } } SummaryExpr::SummarySubtract { left, right } + | SummaryExpr::RelationalJoin { left, right, .. } | SummaryExpr::BinaryOp { lhs: left, rhs: right, @@ -1315,6 +1326,10 @@ fn count_operations(root: &SummaryNode) -> Result { + visit(left, seen, counts)?; + visit(right, seen, counts)?; + } SummaryExpr::CandidateTopK { candidates, values, .. } => { diff --git a/crates/asap-aware-mapping/src/summary_maintenance_cost/model.rs b/crates/asap-aware-mapping/src/summary_maintenance_cost/model.rs index 72f7546d..408f0c3f 100644 --- a/crates/asap-aware-mapping/src/summary_maintenance_cost/model.rs +++ b/crates/asap-aware-mapping/src/summary_maintenance_cost/model.rs @@ -3198,6 +3198,7 @@ mod tests { } } SummaryExpr::SummarySubtract { left, right } + | SummaryExpr::RelationalJoin { left, right, .. } | SummaryExpr::BinaryOp { lhs: left, rhs: right, @@ -3412,6 +3413,7 @@ mod tests { } } SummaryExpr::SummarySubtract { left, right } + | SummaryExpr::RelationalJoin { left, right, .. } | SummaryExpr::BinaryOp { lhs: left, rhs: right, @@ -3460,6 +3462,7 @@ mod tests { } } SummaryExpr::SummarySubtract { left, right } + | SummaryExpr::RelationalJoin { left, right, .. } | SummaryExpr::BinaryOp { lhs: left, rhs: right, diff --git a/crates/asap-aware-mapping/src/summary_maintenance_cost/window.rs b/crates/asap-aware-mapping/src/summary_maintenance_cost/window.rs index 6aa6c642..d4ec85d6 100644 --- a/crates/asap-aware-mapping/src/summary_maintenance_cost/window.rs +++ b/crates/asap-aware-mapping/src/summary_maintenance_cost/window.rs @@ -51,6 +51,7 @@ pub(super) fn summary_aggregation_identities(root: &SummaryNode) -> HashSet<*con } } SummaryExpr::SummarySubtract { left, right } + | SummaryExpr::RelationalJoin { left, right, .. } | SummaryExpr::BinaryOp { lhs: left, rhs: right, diff --git a/crates/asap-aware-mapping/src/summary_maintenance_dag_export.rs b/crates/asap-aware-mapping/src/summary_maintenance_dag_export.rs index 55f1ddf8..bbb0e2c8 100644 --- a/crates/asap-aware-mapping/src/summary_maintenance_dag_export.rs +++ b/crates/asap-aware-mapping/src/summary_maintenance_dag_export.rs @@ -146,6 +146,11 @@ fn summary_children(expr: &SummaryExpr) -> Vec<&Rc> { SummaryExpr::SummaryAgg { child, .. } => vec![child], SummaryExpr::ValueOperation { child, .. } => vec![child], SummaryExpr::SummaryJoin { outer, inner, .. } + | SummaryExpr::RelationalJoin { + left: outer, + right: inner, + .. + } | SummaryExpr::SummarySubtract { left: outer, right: inner, diff --git a/crates/asap-aware-mapping/src/summary_maintenance_lifecycle.rs b/crates/asap-aware-mapping/src/summary_maintenance_lifecycle.rs index 464ed13d..922b730a 100644 --- a/crates/asap-aware-mapping/src/summary_maintenance_lifecycle.rs +++ b/crates/asap-aware-mapping/src/summary_maintenance_lifecycle.rs @@ -970,6 +970,11 @@ fn collect_summary_aggs( } SummaryExpr::ValueOperation { child, .. } => collect_summary_aggs(child, seen, output), SummaryExpr::SummaryJoin { outer, inner, .. } + | SummaryExpr::RelationalJoin { + left: outer, + right: inner, + .. + } | SummaryExpr::BinaryOp { lhs: outer, rhs: inner, diff --git a/crates/integration-tests/tests/sql_to_post_asap.rs b/crates/integration-tests/tests/sql_to_post_asap.rs index 8e7a517c..f625bfbd 100644 --- a/crates/integration-tests/tests/sql_to_post_asap.rs +++ b/crates/integration-tests/tests/sql_to_post_asap.rs @@ -27,9 +27,9 @@ use asap_aware_mapping::{ }; use asap_frontend_sql::{lower_sql, lower_sql_dialect, SqlCatalog}; use asap_types::post_asap::{ - compile_executable_dag, ExactKind, ExactParams, ExecutableOperatorPayload, GroupingStrategy, - SketchAlgorithm, SketchKind, SketchParams, SketchQuery, SummaryExpr, SummaryFamilyType, - SummaryNode, SummarySchema, SummaryUpdate, ValueOperation, + compile_executable_dag, EdgeRole, ExactKind, ExactParams, ExecutableOperatorPayload, + GroupingStrategy, SketchAlgorithm, SketchKind, SketchParams, SketchQuery, SummaryExpr, + SummaryFamilyType, SummaryNode, SummarySchema, SummaryUpdate, ValueOperation, }; use asap_types::pre_asap::expr_ir::ColumnRef; use asap_types::pre_asap::query_expr::{QueryExpr, Reduction}; @@ -203,6 +203,151 @@ async fn sql_full_query_retains_project_and_binds_inner_aggregate() { ); } +/// A relational join remains a read-time node while both derived-table +/// aggregates are independently selected as physical summaries. +#[tokio::test] +async fn sql_join_recursively_binds_both_temporal_aggregate_children() { + let pre_asap = Rc::new( + lower_sql_dialect( + "SELECT a.service, a.v / b.v AS ratio FROM \ + (SELECT service, asap_rate(latency, ts, 300000) AS v FROM metrics WHERE service='errors' GROUP BY service) a \ + INNER JOIN \ + (SELECT service, asap_rate(latency, ts, 300000) AS v FROM metrics WHERE service='requests' GROUP BY service) b \ + ON b.service=a.service", + &catalog(), + SqlDialect::ClickhouseSQL, + AccuracyTarget::Exact, + ) + .await + .expect("two-subquery rate ratio must lower"), + ); + let space = search_workload(vec![("ratio", Rc::clone(&pre_asap))]); + let selection = space.global_selection(&DefaultCostModel); + let root = selection + .materialize(&space.roots[0].1) + .expect("materialization failed") + .expect("root must be discovered"); + let SummaryExpr::ValueOperation { + child: join, + operation: ValueOperation::Project { cols, .. }, + .. + } = &root.expr + else { + panic!( + "expected Project above relational join, got {:?}", + root.expr + ); + }; + assert!(matches!( + &cols[1].expr, + QueryExpr::Arithmetic { + op: asap_types::pre_asap::ArithmeticOpKind::Div, + .. + } + )); + let SummaryExpr::RelationalJoin { + left, + right, + kind, + pred, + } = &join.expr + else { + panic!("expected read-time relational join, got {:?}", join.expr); + }; + assert_eq!(kind, &asap_types::pre_asap::JoinKind::Inner); + assert!(matches!( + pred.0.as_ref(), + QueryExpr::Compare { + left, + op: asap_types::pre_asap::CompareOpKind::Eq, + right, + } if matches!(left.as_ref(), QueryExpr::Column(0)) + && matches!(right.as_ref(), QueryExpr::Column(2)) + )); + assert_eq!( + join.schema + .fields + .iter() + .map(|field| field.name.as_str()) + .collect::>(), + vec!["service", "v", "service", "v"] + ); + for child in [left, right] { + let SummaryExpr::ValueOperation { + child: aggregate, + operation: ValueOperation::Project { .. }, + .. + } = &child.expr + else { + panic!("derived table Project was not retained: {:?}", child.expr); + }; + assert!(matches!( + aggregate.expr, + SummaryExpr::SummaryAgg { + family: SummaryFamilyType::ExactAggregate(ExactKind::Rate, ExactParams::Rate), + .. + } + )); + } + assert!(join + .guarantee + .as_ref() + .is_some_and(|value| value.is_exact())); + let executable = compile_executable_dag(&root).expect("join DAG must compile"); + let join_id = executable + .nodes + .iter() + .find(|node| { + matches!( + node.payload, + ExecutableOperatorPayload::RelationalJoin { .. } + ) + }) + .expect("relational join node") + .id; + let roles = executable + .edges + .iter() + .filter(|edge| edge.consumer == join_id) + .map(|edge| edge.role) + .collect::>(); + assert_eq!(roles, vec![EdgeRole::Left, EdgeRole::Right]); +} + +#[tokio::test] +async fn unsupported_sql_join_shapes_remain_fail_closed() { + for sql in [ + "SELECT a.service FROM (SELECT service, asap_rate(latency, ts, 300000) v FROM metrics GROUP BY service) a LEFT JOIN (SELECT service, asap_rate(latency, ts, 300000) v FROM metrics GROUP BY service) b ON a.service=b.service", + "SELECT a.service FROM (SELECT service, asap_rate(latency, ts, 300000) v FROM metrics GROUP BY service) a INNER JOIN (SELECT service, asap_rate(latency, ts, 300000) v FROM metrics GROUP BY service) b ON a.v>b.v", + "SELECT a.service FROM (SELECT service, asap_rate(latency, ts, 300000) v FROM metrics GROUP BY service) a INNER JOIN (SELECT service, asap_rate(latency, ts, 300000) v FROM metrics GROUP BY service) b ON a.service=a.service", + ] { + let pre_asap = Rc::new( + lower_sql_dialect( + sql, + &catalog(), + SqlDialect::ClickhouseSQL, + AccuracyTarget::Exact, + ) + .await + .unwrap_or_else(|error| panic!("join must lower before fail-closed mapping: {error}")), + ); + let space = search_workload(vec![("unsupported-join", Rc::clone(&pre_asap))]); + let selection = space.global_selection(&DefaultCostModel); + let root = selection + .materialize(&space.roots[0].1) + .expect("materialization failed") + .expect("root must be discovered"); + let SummaryExpr::ValueOperation { child, .. } = &root.expr else { + panic!("SQL projection must remain explicit: {:?}", root.expr); + }; + assert!( + matches!(child.expr, SummaryExpr::KeepPreAsap(_)), + "unsupported join was partially accelerated: {:?}", + child.expr + ); + } +} + /// Relational parents emitted around a derived-table aggregate remain /// explicit read-time nodes while the aggregate is summary-bound. #[tokio::test] diff --git a/crates/types/src/dag_export.rs b/crates/types/src/dag_export.rs index 9999338a..05a6f3c0 100644 --- a/crates/types/src/dag_export.rs +++ b/crates/types/src/dag_export.rs @@ -477,6 +477,7 @@ define_summary_kind_tags! { SummaryExpr::BinaryOp { .. } => "SummaryBinaryOp", SummaryExpr::CandidateTopK { .. } => "CandidateTopK", SummaryExpr::ValueOperation { .. } => "ValueOperation", + SummaryExpr::RelationalJoin { .. } => "RelationalJoin", SummaryExpr::SummaryAgg { .. } => "SummaryAgg", SummaryExpr::SummaryJoin { .. } => "SummaryJoin", SummaryExpr::SummarySubtract { .. } => "SummarySubtract", @@ -519,6 +520,15 @@ fn summary_shape(expr: &SummaryExpr) -> (&'static str, String, serde_json::Value "timing": timing.as_str(), }), ), + SummaryExpr::RelationalJoin { + kind: join_kind, + pred, + .. + } => ( + kind, + format!("RelationalJoin({join_kind:?})"), + serde_json::json!({ "join_kind": join_kind, "predicate": pred }), + ), SummaryExpr::SummaryAgg { family, input, @@ -575,6 +585,7 @@ fn summary_children(expr: &SummaryExpr) -> Vec<&Rc> { candidates, values, .. } => vec![candidates, values], SummaryExpr::ValueOperation { child, .. } => vec![child], + SummaryExpr::RelationalJoin { left, right, .. } => vec![left, right], SummaryExpr::SummaryAgg { child, .. } => vec![child], SummaryExpr::SummaryJoin { outer, inner, .. } => vec![outer, inner], SummaryExpr::SummarySubtract { left, right } => vec![left, right], diff --git a/crates/types/src/post_asap/cse.rs b/crates/types/src/post_asap/cse.rs index ce8401cf..fb4de31b 100644 --- a/crates/types/src/post_asap/cse.rs +++ b/crates/types/src/post_asap/cse.rs @@ -66,6 +66,20 @@ fn same_node(left: &SummaryNode, right: &SummaryNode) -> bool { timing: bt, }, ) => Rc::ptr_eq(ac, bc) && same_value(ao, bo) && at == bt, + ( + RelationalJoin { + left: al, + right: ar, + kind: ak, + pred: ap, + }, + RelationalJoin { + left: bl, + right: br, + kind: bk, + pred: bp, + }, + ) => Rc::ptr_eq(al, bl) && Rc::ptr_eq(ar, br) && ak == bk && same_value(ap, bp), ( SummaryAgg { child: ac, @@ -135,6 +149,7 @@ fn same_node(left: &SummaryNode, right: &SummaryNode) -> bool { | BinaryOp { .. } | CandidateTopK { .. } | ValueOperation { .. } + | RelationalJoin { .. } | SummaryAgg { .. } | SummaryJoin { .. } | SummarySubtract { .. } @@ -180,6 +195,10 @@ pub fn share_common_summary_subtrees( *values = visit(values, seen, pool); } SummaryExpr::ValueOperation { child, .. } => *child = visit(child, seen, pool), + SummaryExpr::RelationalJoin { left, right, .. } => { + *left = visit(left, seen, pool); + *right = visit(right, seen, pool); + } SummaryExpr::SummaryJoin { outer, inner, .. } => { *outer = visit(outer, seen, pool); *inner = visit(inner, seen, pool); diff --git a/crates/types/src/post_asap/executable_dag.rs b/crates/types/src/post_asap/executable_dag.rs index 79a10579..3f357b75 100644 --- a/crates/types/src/post_asap/executable_dag.rs +++ b/crates/types/src/post_asap/executable_dag.rs @@ -4,14 +4,14 @@ use std::collections::HashMap; use std::rc::Rc; use super::{ - assigned_child_data_state, validate_execution_data_states, ExecutionDataState, - ExecutionDataStateError, ResultGuarantee, SummaryExpr, SummaryNode, SummarySchema, + validate_execution_data_states, ExecutionDataState, ExecutionDataStateError, ResultGuarantee, + SummaryExpr, SummaryNode, SummarySchema, }; use super::{ BinaryOperator, CandidateCompleteness, ExecutionTiming, GroupingStrategy, SketchQuery, SummaryFamilyType, SummaryUpdate, ValueOperation, }; -use crate::pre_asap::{ColumnRef, GroupKeys, QueryExpr, Reduction}; +use crate::pre_asap::{ColumnRef, GroupKeys, JoinKind, Predicate, QueryExpr, Reduction}; use thiserror::Error; pub const POST_ASAP_DAG_WIRE_VERSION: u32 = 1; @@ -22,6 +22,7 @@ pub enum ExecutableOperator { Binary, CandidateTopK, Value, + RelationalJoin, SummaryAgg, SummaryJoin, SummarySubtract, @@ -82,6 +83,10 @@ pub enum ExecutableOperatorPayload { operation: ValueOperation, timing: ExecutionTiming, }, + RelationalJoin { + join_kind: JoinKind, + pred: Predicate, + }, SummaryAgg { family: SummaryFamilyType, input: SummaryUpdate, @@ -109,6 +114,7 @@ impl ExecutableOperatorPayload { Self::Binary { .. } => ExecutableOperator::Binary, Self::CandidateTopK { .. } => ExecutableOperator::CandidateTopK, Self::Value { .. } => ExecutableOperator::Value, + Self::RelationalJoin { .. } => ExecutableOperator::RelationalJoin, Self::SummaryAgg { .. } => ExecutableOperator::SummaryAgg, Self::SummaryJoin { .. } => ExecutableOperator::SummaryJoin, Self::SummarySubtract => ExecutableOperator::SummarySubtract, @@ -409,6 +415,9 @@ pub fn compile_executable_dag_with_node_ids( SummaryExpr::ValueOperation { child, .. } | SummaryExpr::SummaryAgg { child, .. } => { vec![(child, EdgeRole::Input)] } + SummaryExpr::RelationalJoin { left, right, .. } => { + vec![(left, EdgeRole::Left), (right, EdgeRole::Right)] + } SummaryExpr::SummaryJoin { outer, inner, .. } => { vec![(outer, EdgeRole::Left), (inner, EdgeRole::Right)] } @@ -454,6 +463,12 @@ pub fn compile_executable_dag_with_node_ids( operation: operation.clone(), timing: *timing, }, + SummaryExpr::RelationalJoin { kind, pred, .. } => { + ExecutableOperatorPayload::RelationalJoin { + join_kind: kind.clone(), + pred: pred.clone(), + } + } SummaryExpr::SummaryAgg { family, input, @@ -544,7 +559,13 @@ pub fn compile_executable_dag_with_node_ids( consumer: id, role, intermediate_schema: child.schema.clone(), - data_state: assigned_child_data_state(&node.expr, child), + // The whole-graph validator owns contextual state assignment, + // especially for shared KeepPreAsap leaves. Export that + // authoritative result instead of independently deriving the + // edge state a second time. + data_state: assignment + .data_state_of(child) + .expect("validated child has data state"), grouping, window: if maintenance_dependency { WindowEdgeCompatibility::RequiresAlignedPanePhaseOrExactBoundaryResidual diff --git a/crates/types/src/post_asap/execution_data_state.rs b/crates/types/src/post_asap/execution_data_state.rs index 76478f40..cff0c039 100644 --- a/crates/types/src/post_asap/execution_data_state.rs +++ b/crates/types/src/post_asap/execution_data_state.rs @@ -223,9 +223,9 @@ impl ExecutionDataStateAssignment { pub fn produced_data_state(expr: &SummaryExpr) -> Option { Some(match expr { SummaryExpr::KeepPreAsap(_) => return None, - SummaryExpr::BinaryOp { .. } | SummaryExpr::CandidateTopK { .. } => { - ExecutionDataState::READ_ROWS - } + SummaryExpr::BinaryOp { .. } + | SummaryExpr::CandidateTopK { .. } + | SummaryExpr::RelationalJoin { .. } => ExecutionDataState::READ_ROWS, SummaryExpr::SummaryAgg { .. } | SummaryExpr::SummaryJoin { .. } | SummaryExpr::SummarySubtract { .. } @@ -342,6 +342,20 @@ fn visit( } Ok(()) } + SummaryExpr::RelationalJoin { left, right, .. } => { + for input in [left, right] { + let state = + produced_data_state(&input.expr).unwrap_or(ExecutionDataState::READ_ROWS); + if state != ExecutionDataState::READ_ROWS { + return Err(ExecutionDataStateError::IllegalChildDataState { + edge: "RelationalJoin input", + child: state, + }); + } + visit(input, state, assignment)?; + } + Ok(()) + } SummaryExpr::SummaryAgg { child, .. } => { let child_domain = child_domain( child, @@ -442,6 +456,7 @@ pub fn assigned_child_data_state(parent: &SummaryExpr, child: &SummaryNode) -> E SummaryExpr::KeepPreAsap(_) | SummaryExpr::BinaryOp { .. } | SummaryExpr::CandidateTopK { .. } + | SummaryExpr::RelationalJoin { .. } | SummaryExpr::SummaryAgg { .. } | SummaryExpr::SummaryJoin { .. } | SummaryExpr::SummarySubtract { .. } diff --git a/crates/types/src/post_asap/expr.rs b/crates/types/src/post_asap/expr.rs index 87944a24..2e82150d 100644 --- a/crates/types/src/post_asap/expr.rs +++ b/crates/types/src/post_asap/expr.rs @@ -6,7 +6,8 @@ use super::sketch::{GroupingStrategy, SketchQuery, SummaryUpdate}; use crate::pre_asap::agg_intent::AggIntent; use crate::pre_asap::query_expr::Predicate; use crate::pre_asap::{ - BinaryOpKind, ColumnRef, GroupKeys, ProjectItem, QueryExpr, Reduction, SortKey, VectorMatch, + BinaryOpKind, ColumnRef, GroupKeys, JoinKind, ProjectItem, QueryExpr, Reduction, SortKey, + VectorMatch, }; #[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)] @@ -143,6 +144,16 @@ pub enum SummaryExpr { timing: super::execution_data_state::ExecutionTiming, }, + /// Read-time relational join over two row-producing children. This is + /// distinct from [`SummaryJoin`](Self::SummaryJoin), which combines + /// summary states for join estimation during maintenance. + RelationalJoin { + left: Rc, + right: Rc, + kind: JoinKind, + pred: Predicate, + }, + /// Summary aggregation. Post-ASAP binding chose `family` — which /// summary family (exact accumulator, sketch, sample, wavelet, or /// statistical model) and its `(kind, params)` — from the catalog for diff --git a/tools/dag-viewer/node-style.js b/tools/dag-viewer/node-style.js index 429f614b..a13c51d8 100644 --- a/tools/dag-viewer/node-style.js +++ b/tools/dag-viewer/node-style.js @@ -22,6 +22,7 @@ const KIND_CATEGORY_JSON = `{ "TimeShift": "window", "SQLWindowFunc": "window", "Join": "join", + "RelationalJoin": "join", "Dedup": "set", "SetOp": "set", "Concat": "combine", From 7ad7877214df3c74e59242ce4e5c36e7f2cada27 Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 07:10:55 -0600 Subject: [PATCH 2/3] feat(sql): lower temporal and histogram planning bridges --- crates/frontend-sql/src/sql/mod.rs | 121 ++++++++++++++++++++++ crates/frontend-sql/tests/sql_lowering.rs | 51 +++++++++ 2 files changed, 172 insertions(+) diff --git a/crates/frontend-sql/src/sql/mod.rs b/crates/frontend-sql/src/sql/mod.rs index f2153bed..e7608d02 100644 --- a/crates/frontend-sql/src/sql/mod.rs +++ b/crates/frontend-sql/src/sql/mod.rs @@ -229,6 +229,17 @@ impl<'a> SqlLowerer<'a> { builtin.arity, )); } + // Planning-only relation markers. They let a workload author state + // the PromQL temporal/classic-histogram semantics of an equivalent SQL + // rewrite without teaching the canonical IR a second, SQL-specific + // spelling of either operation. `lower_projection` consumes these + // calls; they can never survive as executable scalar functions. + for (name, arity) in [ + ("asap_promql_subquery", Arity::Exact(2)), + ("asap_histogram_quantile", Arity::Exact(1)), + ] { + ctx.register_udf(clickhouse_scalar_builtin_stub_udf(name, arity)); + } // Register a stub `WindowUDF` for every catalog-listed ClickHouse-only // *window* builtin — same reason as the two loops above, but with no // rewrite step to follow: `lower_window_func_kind` already maps each @@ -643,6 +654,25 @@ impl<'a> SqlLowerer<'a> { &self, proj: &logical_expr::Projection, ) -> Result { + if let Some(bridge) = planning_bridge(proj)? { + let child = Rc::new(self.lower_plan(&proj.input)?); + return Ok(match bridge { + PlanningBridge::PromqlSubquery { range, resolution } => { + Unresolved::PromqlSubquery { + range, + resolution: Some(resolution), + child, + } + } + PlanningBridge::HistogramQuantile { q } => Unresolved::Aggregate { + reduction: Reduction::Reduce(GroupKeys::none()), + measures: vec![AggIntent::HistogramQuantile { q }], + output_names: vec!["value".into()], + having: None, + child, + }, + }); + } // SELECT * — no column constraint; pass through without a Project. if proj.expr.iter().any(|e| matches!(e, Expr::Wildcard { .. })) { return self.lower_plan(&proj.input); @@ -1083,6 +1113,94 @@ impl<'a> SqlLowerer<'a> { } } +/// A deliberately explicit marker accepted only in a projection of planning +/// SQL. The marker describes a relation operator, so it is removed rather than +/// lowered to the ordinary scalar `FunctionCall` variant. +enum PlanningBridge { + PromqlSubquery { + range: Duration, + resolution: Duration, + }, + HistogramQuantile { + q: f64, + }, +} + +fn planning_bridge( + projection: &logical_expr::Projection, +) -> Result, LoweringError> { + let mut found = None; + for expr in &projection.expr { + let Expr::ScalarFunction(call) = unalias(expr) else { + continue; + }; + let name = call.func.name().to_ascii_lowercase(); + let bridge = match name.as_str() { + "asap_promql_subquery" => { + let [range, resolution] = call.args.as_slice() else { + return Err(LoweringError::InvalidExpression( + "asap_promql_subquery requires (range_ms, resolution_ms)".into(), + )); + }; + let range = positive_millis_literal(range, "range_ms")?; + let resolution = positive_millis_literal(resolution, "resolution_ms")?; + PlanningBridge::PromqlSubquery { range, resolution } + } + "asap_histogram_quantile" => { + let [q] = call.args.as_slice() else { + return Err(LoweringError::InvalidExpression( + "asap_histogram_quantile requires one literal quantile".into(), + )); + }; + let q = float_literal(q).ok_or_else(|| { + LoweringError::InvalidExpression( + "asap_histogram_quantile quantile must be a numeric literal".into(), + ) + })?; + if !q.is_finite() || !(0.0..=1.0).contains(&q) { + return Err(LoweringError::InvalidExpression(format!( + "asap_histogram_quantile quantile must be finite and in [0,1], got {q}" + ))); + } + PlanningBridge::HistogramQuantile { q } + } + _ => continue, + }; + if found.is_some() { + return Err(LoweringError::InvalidExpression( + "a planning projection may contain only one asap_* relation marker".into(), + )); + } + found = Some(bridge); + } + Ok(found) +} + +fn positive_millis_literal(expr: &Expr, argument: &str) -> Result { + let millis = match unalias(expr) { + Expr::Literal(DfScalarValue::Int64(Some(value))) if *value > 0 => *value as u64, + Expr::Literal(DfScalarValue::UInt64(Some(value))) if *value > 0 => *value, + Expr::Literal(DfScalarValue::Int32(Some(value))) if *value > 0 => *value as u64, + other => { + return Err(LoweringError::InvalidExpression(format!( + "{argument} must be a positive integer millisecond literal, got {other}" + ))) + } + }; + Ok(Duration::from_millis(millis)) +} + +fn float_literal(expr: &Expr) -> Option { + match unalias(expr) { + Expr::Literal(DfScalarValue::Float64(Some(value))) => Some(*value), + Expr::Literal(DfScalarValue::Float32(Some(value))) => Some(*value as f64), + Expr::Literal(DfScalarValue::Int64(Some(value))) => Some(*value as f64), + Expr::Literal(DfScalarValue::UInt64(Some(value))) => Some(*value as f64), + Expr::Literal(DfScalarValue::Int32(Some(value))) => Some(*value as f64), + _ => None, + } +} + // ── ClickHouse-builtin compatibility, taught to DataFusion itself ────────────── // // Generalized over `asap_sql_function_catalog::CLICKHOUSE_BUILTINS` (issue @@ -1191,6 +1309,9 @@ fn clickhouse_scalar_builtin_return_type(name: &str) -> ArrowDataType { } // 1-based match position, 0 if not found. "positioncaseinsensitive" => ArrowDataType::UInt64, + // Relation markers are removed by `lower_projection`; Float64 merely + // lets DataFusion type the temporary SELECT list. + "asap_promql_subquery" | "asap_histogram_quantile" => ArrowDataType::Float64, other => unreachable!( "{other}: every CLICKHOUSE_SCALAR_BUILTINS entry must have a return type listed here" ), diff --git a/crates/frontend-sql/tests/sql_lowering.rs b/crates/frontend-sql/tests/sql_lowering.rs index b91a4a4f..1580170d 100644 --- a/crates/frontend-sql/tests/sql_lowering.rs +++ b/crates/frontend-sql/tests/sql_lowering.rs @@ -49,6 +49,57 @@ async fn lower(sql: &str) -> QueryExpr { .unwrap_or_else(|e| panic!("lower failed for {sql:?}: {e}")) } +#[tokio::test] +async fn planning_subquery_bridge_reuses_canonical_promql_subquery() { + let query = lower( + "SELECT max(value) FROM (\ + SELECT asap_promql_subquery(21600000, 60000) AS value FROM (\ + SELECT sum(bytes) AS value FROM metrics))", + ) + .await; + let QueryExpr::Project { child, .. } = query else { + panic!("expected outer SQL projection"); + }; + let QueryExpr::Aggregate { child, .. } = child.as_ref() else { + panic!("expected outer max aggregate, got {child:?}"); + }; + let QueryExpr::PromqlSubquery { + range, + resolution, + child, + } = child.as_ref() + else { + panic!("expected canonical subquery bridge, got {child:?}"); + }; + assert_eq!(*range, std::time::Duration::from_secs(6 * 60 * 60)); + assert_eq!(*resolution, Some(std::time::Duration::from_secs(60))); + assert!(matches!(child.as_ref(), QueryExpr::Project { .. })); +} + +#[tokio::test] +async fn planning_histogram_bridge_reuses_classic_bucket_intent() { + let query = lower( + "SELECT asap_histogram_quantile(0.95) AS value FROM (\ + SELECT service AS le, sum(bytes) AS value FROM metrics GROUP BY service)", + ) + .await; + let QueryExpr::Aggregate { + reduction, + measures, + child, + .. + } = query + else { + panic!("expected canonical histogram aggregate"); + }; + assert!(reduction.expect_reduce().keys().is_empty()); + assert!(matches!( + measures.as_slice(), + [AggIntent::HistogramQuantile { q }] if (*q - 0.95).abs() < 1e-12 + )); + assert!(matches!(child.as_ref(), QueryExpr::Project { .. })); +} + /// Find the first `Aggregate` node along the single-child spine. fn find_aggregate(qe: &QueryExpr) -> Option<(&GroupKeys, &Vec)> { match qe { From b3dc404e22391e2aca2be0e16878a401b32da3fe Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 07:12:28 -0600 Subject: [PATCH 3/3] fix(sql): validate planning bridge projections --- crates/frontend-sql/src/sql/mod.rs | 61 ++++++++++++++++++++++- crates/frontend-sql/tests/sql_lowering.rs | 32 ++++++++++++ 2 files changed, 91 insertions(+), 2 deletions(-) diff --git a/crates/frontend-sql/src/sql/mod.rs b/crates/frontend-sql/src/sql/mod.rs index e7608d02..5226840b 100644 --- a/crates/frontend-sql/src/sql/mod.rs +++ b/crates/frontend-sql/src/sql/mod.rs @@ -655,9 +655,10 @@ impl<'a> SqlLowerer<'a> { proj: &logical_expr::Projection, ) -> Result { if let Some(bridge) = planning_bridge(proj)? { - let child = Rc::new(self.lower_plan(&proj.input)?); + let input = self.lower_plan(&proj.input)?; return Ok(match bridge { PlanningBridge::PromqlSubquery { range, resolution } => { + let child = Rc::new(temporal_bridge_projection(proj, input)?); Unresolved::PromqlSubquery { range, resolution: Some(resolution), @@ -669,7 +670,7 @@ impl<'a> SqlLowerer<'a> { measures: vec![AggIntent::HistogramQuantile { q }], output_names: vec!["value".into()], having: None, - child, + child: Rc::new(input), }, }); } @@ -1173,9 +1174,65 @@ fn planning_bridge( } found = Some(bridge); } + if matches!(found, Some(PlanningBridge::HistogramQuantile { .. })) && projection.expr.len() != 1 + { + return Err(LoweringError::InvalidExpression( + "asap_histogram_quantile must be the projection's only expression".into(), + )); + } Ok(found) } +/// Rebuild the SQL projection around the relation sampled by the temporal +/// marker. The marker's alias names the existing child column that occupies +/// its output slot (`... asap_promql_subquery(...) AS value ...`). This makes +/// the bridge schema-preserving without silently retaining columns that SQL +/// projected away. +fn temporal_bridge_projection( + projection: &logical_expr::Projection, + child: Unresolved, +) -> Result { + let cols = projection + .expr + .iter() + .map(|expr| { + if let Expr::ScalarFunction(call) = unalias(expr) { + if call + .func + .name() + .eq_ignore_ascii_case("asap_promql_subquery") + { + let Expr::Alias(alias) = expr else { + return Err(LoweringError::InvalidExpression( + "asap_promql_subquery must have an alias naming its child value column" + .into(), + )); + }; + return Ok(ProjectItem { + expr: Unresolved::Column(ColumnRef::Named(alias.name.clone())), + alias: Some(alias.name.clone()), + }); + } + } + match expr { + Expr::Alias(alias) => Ok(ProjectItem { + expr: df_expr_to_unresolved(&alias.expr)?, + alias: Some(alias.name.clone()), + }), + other => Ok(ProjectItem { + expr: df_expr_to_unresolved(other)?, + alias: None, + }), + } + }) + .collect::, LoweringError>>()?; + Ok(Unresolved::Project { + cols, + qualifier: None, + child: Rc::new(child), + }) +} + fn positive_millis_literal(expr: &Expr, argument: &str) -> Result { let millis = match unalias(expr) { Expr::Literal(DfScalarValue::Int64(Some(value))) if *value > 0 => *value as u64, diff --git a/crates/frontend-sql/tests/sql_lowering.rs b/crates/frontend-sql/tests/sql_lowering.rs index 1580170d..6099bfd5 100644 --- a/crates/frontend-sql/tests/sql_lowering.rs +++ b/crates/frontend-sql/tests/sql_lowering.rs @@ -100,6 +100,38 @@ async fn planning_histogram_bridge_reuses_classic_bucket_intent() { assert!(matches!(child.as_ref(), QueryExpr::Project { .. })); } +#[tokio::test] +async fn planning_relation_bridges_reject_ambiguous_shapes() { + let missing_alias = lower_sql( + "SELECT asap_promql_subquery(300000, 60000) FROM metrics", + &catalog(), + AccuracyTarget::Exact, + ) + .await + .unwrap_err(); + assert!(missing_alias.to_string().contains("must have an alias")); + + let histogram_with_extra_column = lower_sql( + "SELECT service, asap_histogram_quantile(0.95) AS value FROM metrics", + &catalog(), + AccuracyTarget::Exact, + ) + .await + .unwrap_err(); + assert!(histogram_with_extra_column + .to_string() + .contains("only expression")); + + let invalid_q = lower_sql( + "SELECT asap_histogram_quantile(1.5) AS value FROM metrics", + &catalog(), + AccuracyTarget::Exact, + ) + .await + .unwrap_err(); + assert!(invalid_q.to_string().contains("finite and in [0,1]")); +} + /// Find the first `Aggregate` node along the single-child spine. fn find_aggregate(qe: &QueryExpr) -> Option<(&GroupKeys, &Vec)> { match qe {