feat(sql): plan joins over summary children

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<SummaryNode>) -> Rc<SummaryNode> {
+fn finalize_exact_accumulator(
+    node: Rc<SummaryNode>,
+    logical_output: &QueryExpr,
+) -> Result<Rc<SummaryNode>, ImplementError> {
     let is_exact_state = matches!(
         node.expr,
         SummaryExpr::SummaryAgg {
@@ -1778,19 +1783,23 @@ fn finalize_exact_accumulator(node: Rc<SummaryNode>) -> Rc<SummaryNode> {
         }
     );
     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,
             timing: ExecutionTiming::ReadTime,
         },
         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<HashMap<*const QueryExpr, Rc<SummaryNode>>>,
 }
 
+fn normalize_cross_input_equi_predicate(
+    pred: &Predicate,
+    left_width: usize,
+    total_width: usize,
+) -> Option<Predicate> {
+    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<ResultGuarantee> {
+    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<Item = &SelectedGroup<'a>> {
@@ -3693,6 +3743,38 @@ impl<'a> GlobalSelection<'a> {
         &self,
         target: &Rc<QueryExpr>,
     ) -> Result<Rc<SummaryNode>, 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)
     }

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<SummaryOperationCounts, Analyt
                 visit(lhs, seen, counts)?;
                 visit(rhs, seen, counts)?;
             }
+            SummaryExpr::RelationalJoin { left, right, .. } => {
+                visit(left, seen, counts)?;
+                visit(right, seen, counts)?;
+            }
             SummaryExpr::CandidateTopK {
                 candidates, values, ..
             } => {

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,

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,

crates/asap-aware-mapping/src/summary_maintenance_dag_export.rs

@@ -146,6 +146,11 @@ fn summary_children(expr: &SummaryExpr) -> Vec<&Rc<SummaryNode>> {
         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,

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,