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",