From dca264a65571a7989bb270e05c826e6aab41287d Mon Sep 17 00:00:00 2001 From: Jiawei Zhao Date: Thu, 25 Jun 2026 20:01:39 +0800 Subject: [PATCH 1/6] refactor: make join projection pushdown schema-aware via ColumnIndex/JoinSide try_pushdown_through_join decided whether each projected join-output column belonged to the left or right child by comparing its output index against the left child's field count, assuming a `left ++ right` output schema. That is wrong for joins whose output is not a plain concatenation -- mark joins append a synthetic `mark` column (JoinSide::None) belonging to neither child -- which is why #22902 had to bypass projection pushdown for LeftMark / RightMark. Group the projected columns by the join's output-origin metadata (ColumnIndex.side) instead of output position: Left/Right columns push to the matching child via the child-relative ColumnIndex.index, and a synthetic (JoinSide::None) column makes the pushdown decline so the projection is embedded into the join instead -- no panic, no mis-routing. This lets the LeftMark / RightMark bypass be removed from HashJoinExec and NestedLoopJoinExec. The shared helpers used by cross / symmetric / sort-merge joins (new_join_children, join_table_borders, update_join_on/filter) keep their signatures; the side-grouped path uses a new new_join_children_from_groups and reuses update_join_on/filter with a zero column-index offset. Mark joins still fall back to embedding (pushing their child columns down is a follow-up). Closes #23010 Signed-off-by: Jiawei Zhao --- .../physical-plan/src/joins/hash_join/exec.rs | 35 +++--- .../src/joins/nested_loop_join.rs | 35 +++--- datafusion/physical-plan/src/projection.rs | 106 ++++++++++++------ .../sqllogictest/test_files/subquery.slt | 90 +++++++++++++++ 4 files changed, 198 insertions(+), 68 deletions(-) diff --git a/datafusion/physical-plan/src/joins/hash_join/exec.rs b/datafusion/physical-plan/src/joins/hash_join/exec.rs index a5da391ee7635..3b6a7332b3200 100644 --- a/datafusion/physical-plan/src/joins/hash_join/exec.rs +++ b/datafusion/physical-plan/src/joins/hash_join/exec.rs @@ -885,6 +885,11 @@ impl HashJoinExec { true } + /// column indices + pub fn column_indices(&self) -> &Vec { + &self.column_indices + } + /// left (build) side which gets hashed pub fn left(&self) -> &Arc { &self.left @@ -1557,23 +1562,21 @@ impl ExecutionPlan for HashJoinExec { return Ok(None); } - // TODO: split by `col`/`JoinSide` instead so mark joins can also push down to children. let schema = self.schema(); - if !matches!(self.join_type(), JoinType::LeftMark | JoinType::RightMark) - && let Some(JoinData { - projected_left_child, - projected_right_child, - join_filter, - join_on, - }) = try_pushdown_through_join( - projection, - self.left(), - self.right(), - self.on(), - &schema, - self.filter(), - )? - { + if let Some(JoinData { + projected_left_child, + projected_right_child, + join_filter, + join_on, + }) = try_pushdown_through_join( + projection, + self.left(), + self.right(), + self.on(), + &schema, + self.filter(), + self.column_indices(), + )? { self.builder() .with_new_children(vec![ Arc::new(projected_left_child), diff --git a/datafusion/physical-plan/src/joins/nested_loop_join.rs b/datafusion/physical-plan/src/joins/nested_loop_join.rs index db552fed96724..be4a0266dd479 100644 --- a/datafusion/physical-plan/src/joins/nested_loop_join.rs +++ b/datafusion/physical-plan/src/joins/nested_loop_join.rs @@ -336,6 +336,11 @@ impl NestedLoopJoinExec { .build() } + /// column indices + pub fn column_indices(&self) -> &Vec { + &self.column_indices + } + /// left side pub fn left(&self) -> &Arc { &self.left @@ -733,23 +738,21 @@ impl ExecutionPlan for NestedLoopJoinExec { return Ok(None); } - // TODO: split by `col`/`JoinSide` instead so mark joins can also push down to children. let schema = self.schema(); - if !matches!(self.join_type(), JoinType::LeftMark | JoinType::RightMark) - && let Some(JoinData { - projected_left_child, - projected_right_child, - join_filter, - .. - }) = try_pushdown_through_join( - projection, - self.left(), - self.right(), - &[], - &schema, - self.filter(), - )? - { + if let Some(JoinData { + projected_left_child, + projected_right_child, + join_filter, + .. + }) = try_pushdown_through_join( + projection, + self.left(), + self.right(), + &[], + &schema, + self.filter(), + self.column_indices(), + )? { Ok(Some(Arc::new(NestedLoopJoinExec::try_new( Arc::new(projected_left_child), Arc::new(projected_right_child), diff --git a/datafusion/physical-plan/src/projection.rs b/datafusion/physical-plan/src/projection.rs index 16b0a5ad7e4b5..b717a19227c57 100644 --- a/datafusion/physical-plan/src/projection.rs +++ b/datafusion/physical-plan/src/projection.rs @@ -649,54 +649,64 @@ pub fn try_pushdown_through_join( join_on: JoinOnRef, schema: &SchemaRef, filter: Option<&JoinFilter>, + column_indices: &[ColumnIndex], ) -> Result> { // Convert projected expressions to columns. We can not proceed if this is not possible. let Some(projection_as_columns) = physical_to_column_exprs(projection.expr()) else { return Ok(None); }; - let (far_right_left_col_ind, far_left_right_col_ind) = - join_table_borders(join_left.schema().fields().len(), &projection_as_columns); + if projection_as_columns.len() >= schema.fields().len() { + return Ok(None); + } + let mut left_proj: Vec<(Column, String)> = Vec::new(); + let mut right_proj: Vec<(Column, String)> = Vec::new(); + let mut seen_right = false; + for (col, alias) in &projection_as_columns { + let origin = &column_indices[col.index()]; + match origin.side { + // Keep the "left block before right block" contiguity the current + // pushdown supports; a left column after a right one is "mixed". + JoinSide::Left => { + if seen_right { + return Ok(None); + } + left_proj.push((Column::new(col.name(), origin.index), alias.clone())); + } + JoinSide::Right => { + seen_right = true; + right_proj.push((Column::new(col.name(), origin.index), alias.clone())); + } + // Synthetic column (e.g. mark): belongs to neither child. + // Phase 2 declines; Phase 3 keeps it at the join output instead. + JoinSide::None => return Ok(None), + } + } - if !join_allows_pushdown( - &projection_as_columns, - schema, - far_right_left_col_ind, - far_left_right_col_ind, - ) { + // Parity: neither side fully dropped. + if left_proj.is_empty() || right_proj.is_empty() { return Ok(None); } + // `left_proj` / `right_proj` carry *child* indices (from `column_indices`), + // so the shared `update_join_*` helpers must use a 0 column-index offset for + // both sides (the offset bridges child -> join-output index, which is the + // identity here). let new_filter = if let Some(filter) = filter { - match update_join_filter( - &projection_as_columns[0..=far_right_left_col_ind as _], - &projection_as_columns[far_left_right_col_ind as _..], - filter, - join_left.schema().fields().len(), - ) { - Some(updated_filter) => Some(updated_filter), + match update_join_filter(&left_proj, &right_proj, filter, 0) { + Some(updated) => Some(updated), None => return Ok(None), } } else { None }; - let Some(new_on) = update_join_on( - &projection_as_columns[0..=far_right_left_col_ind as _], - &projection_as_columns[far_left_right_col_ind as _..], - join_on, - join_left.schema().fields().len(), - ) else { + let Some(new_on) = update_join_on(&left_proj, &right_proj, join_on, 0) else { return Ok(None); }; - let (new_left, new_right) = new_join_children( - &projection_as_columns, - far_right_left_col_ind, - far_left_right_col_ind, - join_left, - join_right, - )?; + let (new_left, new_right) = + new_join_children_from_groups(&left_proj, &right_proj, join_left, join_right)?; Ok(Some(JoinData { projected_left_child: new_left, @@ -880,6 +890,34 @@ pub fn new_join_children( Ok((new_left, new_right)) } +/// Build the projected left and right children from side-grouped projection +/// columns whose indices are already *child*-relative (e.g. derived from a +/// join's `ColumnIndex`). Unlike [`new_join_children`], this does not infer +/// child ownership from output position, so it is safe for join schemas whose +/// output is not a plain `left ++ right` (used by the schema-aware +/// `try_pushdown_through_join`). +fn new_join_children_from_groups( + left_proj: &[(Column, String)], + right_proj: &[(Column, String)], + left_child: &Arc, + right_child: &Arc, +) -> Result<(ProjectionExec, ProjectionExec)> { + let build = |cols: &[(Column, String)], child: &Arc| { + ProjectionExec::try_new( + cols.iter().map(|(col, alias)| ProjectionExpr { + expr: Arc::new(Column::new(col.name(), col.index())) as _, + alias: alias.clone(), + }), + Arc::clone(child), + ) + }; + + Ok(( + build(left_proj, left_child)?, + build(right_proj, right_child)?, + )) +} + /// Checks three conditions for pushing a projection down through a join: /// - Projection must narrow the join output schema. /// - Columns coming from left/right tables must be collected at the left/right @@ -946,14 +984,10 @@ pub fn update_join_on( .map(|(left, right)| (left, right)) .unzip(); - let new_left_columns = new_columns_for_join_on(&left_idx, proj_left_exprs, 0); - let new_right_columns = - new_columns_for_join_on(&right_idx, proj_right_exprs, left_field_size); - - match (new_left_columns, new_right_columns) { - (Some(left), Some(right)) => Some(left.into_iter().zip(right).collect()), - _ => None, - } + let new_left = new_columns_for_join_on(&left_idx, proj_left_exprs, 0)?; + let new_right = + new_columns_for_join_on(&right_idx, proj_right_exprs, left_field_size)?; + Some(new_left.into_iter().zip(new_right).collect()) } /// Tries to update the column indices of a [`JoinFilter`] as if the input of diff --git a/datafusion/sqllogictest/test_files/subquery.slt b/datafusion/sqllogictest/test_files/subquery.slt index 908ee6bb3be75..897b6640a6bd9 100644 --- a/datafusion/sqllogictest/test_files/subquery.slt +++ b/datafusion/sqllogictest/test_files/subquery.slt @@ -1370,6 +1370,96 @@ where t1_id > 40 or not exists (select 1 from t2 where t2.t2_int > t1.t1_int) 33 44 +########## +# Regression for https://github.com/apache/datafusion/issues/23010: +# a projection that selects / reorders a subset of columns over a mark join. +# Schema-aware projection pushdown (driven by ColumnIndex / JoinSide) must keep +# the synthetic `mark` column (JoinSide::None) at the join output while pushing +# the child columns down. These lock the query results (which must stay stable +# across the refactor) and the current plan shape (the physical plan is expected +# to change once child pushdown is enabled for mark joins). Cover hash LeftMark, +# negated mark, and nested-loop mark. +########## + +query TT +EXPLAIN SELECT t1_name, t1_id FROM t1 +WHERE t1_id > 40 OR t1_id IN (SELECT t2_id FROM t2 WHERE t1_int > 0) +---- +logical_plan +01)Projection: t1.t1_name, t1.t1_id +02)--Filter: t1.t1_id > Int32(40) OR __correlated_sq_1.mark +03)----Projection: t1.t1_id, t1.t1_name, __correlated_sq_1.mark +04)------LeftMark Join: t1.t1_id = __correlated_sq_1.t2_id Filter: t1.t1_int > Int32(0) +05)--------TableScan: t1 projection=[t1_id, t1_name, t1_int] +06)--------SubqueryAlias: __correlated_sq_1 +07)----------TableScan: t2 projection=[t2_id] +physical_plan +01)FilterExec: t1_id@0 > 40 OR mark@2, projection=[t1_name@1, t1_id@0] +02)--HashJoinExec: mode=CollectLeft, join_type=RightMark, on=[(t2_id@0, t1_id@0)], filter=t1_int@0 > 0, projection=[t1_id@0, t1_name@1, mark@3] +03)----DataSourceExec: partitions=1, partition_sizes=[2] +04)----RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1 +05)------DataSourceExec: partitions=1, partition_sizes=[2] + +query TI rowsort +SELECT t1_name, t1_id FROM t1 +WHERE t1_id > 40 OR t1_id IN (SELECT t2_id FROM t2 WHERE t1_int > 0) +---- +a 11 +b 22 +d 44 + +query TT +EXPLAIN SELECT t1_int, t1_name FROM t1 +WHERE t1_id < 20 OR NOT EXISTS (SELECT 1 FROM t2 WHERE t1.t1_id = t2.t2_id) +---- +logical_plan +01)Projection: t1.t1_int, t1.t1_name +02)--Filter: t1.t1_id < Int32(20) OR NOT __correlated_sq_1.mark +03)----LeftMark Join: t1.t1_id = __correlated_sq_1.t2_id +04)------TableScan: t1 projection=[t1_id, t1_name, t1_int] +05)------SubqueryAlias: __correlated_sq_1 +06)--------TableScan: t2 projection=[t2_id] +physical_plan +01)FilterExec: t1_id@0 < 20 OR NOT mark@3, projection=[t1_int@2, t1_name@1] +02)--HashJoinExec: mode=CollectLeft, join_type=RightMark, on=[(t2_id@0, t1_id@0)] +03)----DataSourceExec: partitions=1, partition_sizes=[2] +04)----RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1 +05)------DataSourceExec: partitions=1, partition_sizes=[2] + +query IT rowsort +SELECT t1_int, t1_name FROM t1 +WHERE t1_id < 20 OR NOT EXISTS (SELECT 1 FROM t2 WHERE t1.t1_id = t2.t2_id) +---- +1 a +3 c + +query TT +EXPLAIN SELECT t1_name FROM t1 +WHERE t1_id > 40 OR EXISTS (SELECT 1 FROM t2 WHERE t1.t1_int > t2.t2_int) +---- +logical_plan +01)Projection: t1.t1_name +02)--Filter: t1.t1_id > Int32(40) OR __correlated_sq_1.mark +03)----Projection: t1.t1_id, t1.t1_name, __correlated_sq_1.mark +04)------LeftMark Join: Filter: t1.t1_int > __correlated_sq_1.t2_int +05)--------TableScan: t1 projection=[t1_id, t1_name, t1_int] +06)--------SubqueryAlias: __correlated_sq_1 +07)----------TableScan: t2 projection=[t2_int] +physical_plan +01)FilterExec: t1_id@0 > 40 OR mark@2, projection=[t1_name@1] +02)--NestedLoopJoinExec: join_type=RightMark, filter=t1_int@0 > t2_int@1, projection=[t1_id@0, t1_name@1, mark@3] +03)----DataSourceExec: partitions=1, partition_sizes=[2] +04)----RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1 +05)------DataSourceExec: partitions=1, partition_sizes=[2] + +query T rowsort +SELECT t1_name FROM t1 +WHERE t1_id > 40 OR EXISTS (SELECT 1 FROM t2 WHERE t1.t1_int > t2.t2_int) +---- +b +c +d + statement ok set datafusion.explain.logical_plan_only = true; From 03fc98a5f51fd4041a321370f431504c02400a97 Mon Sep 17 00:00:00 2001 From: Jiawei Zhao Date: Thu, 25 Jun 2026 20:47:22 +0800 Subject: [PATCH 2/6] refactor: make try_pushdown_through_join pub(crate) It is a crate-internal helper, called only by HashJoinExec and NestedLoopJoinExec within datafusion-physical-plan, is not re-exported, and has no external consumers. Reducing its visibility keeps it off the public API surface so future signature changes no longer trip cargo-semver-checks. Signed-off-by: Jiawei Zhao --- datafusion/physical-plan/src/projection.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/datafusion/physical-plan/src/projection.rs b/datafusion/physical-plan/src/projection.rs index b717a19227c57..0ed5a61d1c00f 100644 --- a/datafusion/physical-plan/src/projection.rs +++ b/datafusion/physical-plan/src/projection.rs @@ -642,7 +642,7 @@ pub struct JoinData { pub join_on: JoinOn, } -pub fn try_pushdown_through_join( +pub(crate) fn try_pushdown_through_join( projection: &ProjectionExec, join_left: &Arc, join_right: &Arc, From 5bfbeba819391e11c7a45a1f05e4d04f5d76d681 Mon Sep 17 00:00:00 2001 From: Jiawei Zhao Date: Mon, 29 Jun 2026 23:35:36 +0800 Subject: [PATCH 3/6] refactor: avoid public join column API Remove public column_indices accessors from hash and nested-loop join execution nodes. Projection pushdown now passes the internal column_indices slice from within each impl, avoiding new public API surface and keeping the Vec representation private. Signed-off-by: Jiawei Zhao --- datafusion/physical-plan/src/joins/hash_join/exec.rs | 7 +------ datafusion/physical-plan/src/joins/nested_loop_join.rs | 7 +------ 2 files changed, 2 insertions(+), 12 deletions(-) diff --git a/datafusion/physical-plan/src/joins/hash_join/exec.rs b/datafusion/physical-plan/src/joins/hash_join/exec.rs index 3b6a7332b3200..c4a139c9f4aa4 100644 --- a/datafusion/physical-plan/src/joins/hash_join/exec.rs +++ b/datafusion/physical-plan/src/joins/hash_join/exec.rs @@ -885,11 +885,6 @@ impl HashJoinExec { true } - /// column indices - pub fn column_indices(&self) -> &Vec { - &self.column_indices - } - /// left (build) side which gets hashed pub fn left(&self) -> &Arc { &self.left @@ -1575,7 +1570,7 @@ impl ExecutionPlan for HashJoinExec { self.on(), &schema, self.filter(), - self.column_indices(), + self.column_indices.as_slice(), )? { self.builder() .with_new_children(vec![ diff --git a/datafusion/physical-plan/src/joins/nested_loop_join.rs b/datafusion/physical-plan/src/joins/nested_loop_join.rs index be4a0266dd479..8c620c786ee84 100644 --- a/datafusion/physical-plan/src/joins/nested_loop_join.rs +++ b/datafusion/physical-plan/src/joins/nested_loop_join.rs @@ -336,11 +336,6 @@ impl NestedLoopJoinExec { .build() } - /// column indices - pub fn column_indices(&self) -> &Vec { - &self.column_indices - } - /// left side pub fn left(&self) -> &Arc { &self.left @@ -751,7 +746,7 @@ impl ExecutionPlan for NestedLoopJoinExec { &[], &schema, self.filter(), - self.column_indices(), + self.column_indices.as_slice(), )? { Ok(Some(Arc::new(NestedLoopJoinExec::try_new( Arc::new(projected_left_child), From a62b5c6b46ba6a9ed21bb624eed6c683fab667e7 Mon Sep 17 00:00:00 2001 From: Jiawei Zhao Date: Fri, 3 Jul 2026 20:32:41 +0800 Subject: [PATCH 4/6] refactor(physical-plan): preserve join pushdown API Restore the public try_pushdown_through_join signature for semver compatibility. Move the ColumnIndex-aware implementation behind a crate-private helper for join execution internals. Signed-off-by: Jiawei Zhao --- .../physical-plan/src/joins/hash_join/exec.rs | 4 +- .../src/joins/nested_loop_join.rs | 4 +- datafusion/physical-plan/src/projection.rs | 66 ++++++++++++++++++- 3 files changed, 69 insertions(+), 5 deletions(-) diff --git a/datafusion/physical-plan/src/joins/hash_join/exec.rs b/datafusion/physical-plan/src/joins/hash_join/exec.rs index c4a139c9f4aa4..51f1d6f94b8dc 100644 --- a/datafusion/physical-plan/src/joins/hash_join/exec.rs +++ b/datafusion/physical-plan/src/joins/hash_join/exec.rs @@ -49,7 +49,7 @@ use crate::joins::{JoinOn, JoinOnRef, PartitionMode, SharedBitmapBuilder}; use crate::metrics::{Count, MetricBuilder, MetricCategory}; use crate::projection::{ EmbeddedProjection, JoinData, ProjectionExec, try_embed_projection, - try_pushdown_through_join, + try_pushdown_through_join_with_column_indices, }; use crate::repartition::REPARTITION_RANDOM_STATE; use crate::statistics::StatisticsArgs; @@ -1563,7 +1563,7 @@ impl ExecutionPlan for HashJoinExec { projected_right_child, join_filter, join_on, - }) = try_pushdown_through_join( + }) = try_pushdown_through_join_with_column_indices( projection, self.left(), self.right(), diff --git a/datafusion/physical-plan/src/joins/nested_loop_join.rs b/datafusion/physical-plan/src/joins/nested_loop_join.rs index 8c620c786ee84..7bcba43f7917b 100644 --- a/datafusion/physical-plan/src/joins/nested_loop_join.rs +++ b/datafusion/physical-plan/src/joins/nested_loop_join.rs @@ -40,7 +40,7 @@ use crate::metrics::{ }; use crate::projection::{ EmbeddedProjection, JoinData, ProjectionExec, try_embed_projection, - try_pushdown_through_join, + try_pushdown_through_join_with_column_indices, }; use crate::statistics::StatisticsArgs; use crate::{ @@ -739,7 +739,7 @@ impl ExecutionPlan for NestedLoopJoinExec { projected_right_child, join_filter, .. - }) = try_pushdown_through_join( + }) = try_pushdown_through_join_with_column_indices( projection, self.left(), self.right(), diff --git a/datafusion/physical-plan/src/projection.rs b/datafusion/physical-plan/src/projection.rs index 0ed5a61d1c00f..6170cb702c504 100644 --- a/datafusion/physical-plan/src/projection.rs +++ b/datafusion/physical-plan/src/projection.rs @@ -642,7 +642,71 @@ pub struct JoinData { pub join_on: JoinOn, } -pub(crate) fn try_pushdown_through_join( +pub fn try_pushdown_through_join( + projection: &ProjectionExec, + join_left: &Arc, + join_right: &Arc, + join_on: JoinOnRef, + schema: &SchemaRef, + filter: Option<&JoinFilter>, +) -> Result> { + // Convert projected expressions to columns. We can not proceed if this is not possible. + let Some(projection_as_columns) = physical_to_column_exprs(projection.expr()) else { + return Ok(None); + }; + + let (far_right_left_col_ind, far_left_right_col_ind) = + join_table_borders(join_left.schema().fields().len(), &projection_as_columns); + + if !join_allows_pushdown( + &projection_as_columns, + schema, + far_right_left_col_ind, + far_left_right_col_ind, + ) { + return Ok(None); + } + + let new_filter = if let Some(filter) = filter { + match update_join_filter( + &projection_as_columns[0..=far_right_left_col_ind as _], + &projection_as_columns[far_left_right_col_ind as _..], + filter, + join_left.schema().fields().len(), + ) { + Some(updated_filter) => Some(updated_filter), + None => return Ok(None), + } + } else { + None + }; + + let Some(new_on) = update_join_on( + &projection_as_columns[0..=far_right_left_col_ind as _], + &projection_as_columns[far_left_right_col_ind as _..], + join_on, + join_left.schema().fields().len(), + ) else { + return Ok(None); + }; + + let (new_left, new_right) = new_join_children( + &projection_as_columns, + far_right_left_col_ind, + far_left_right_col_ind, + join_left, + join_right, + )?; + + Ok(Some(JoinData { + projected_left_child: new_left, + projected_right_child: new_right, + join_filter: new_filter, + join_on: new_on, + })) +} + +pub(crate) fn try_pushdown_through_join_with_column_indices( projection: &ProjectionExec, join_left: &Arc, join_right: &Arc, From e9db1442a6f1b7eac7d977383d006b29f56d2cfd Mon Sep 17 00:00:00 2001 From: Jiawei Zhao Date: Sat, 11 Jul 2026 11:16:58 +0800 Subject: [PATCH 5/6] refactor(join): consolidate projection pushdown The ColumnIndex-aware path was added beside the public positional helper to preserve API compatibility, leaving duplicate implementations that could diverge as join output schemas evolve. Keep the public API as a compatibility wrapper over the schema-aware path, and use existing output-origin metadata for symmetric hash joins. This preserves semver compatibility while keeping ownership decisions in one implementation. Refs #23010 Signed-off-by: Jiawei Zhao --- .../src/joins/symmetric_hash_join.rs | 92 ++++++------------- datafusion/physical-plan/src/projection.rs | 79 ++++++---------- 2 files changed, 56 insertions(+), 115 deletions(-) diff --git a/datafusion/physical-plan/src/joins/symmetric_hash_join.rs b/datafusion/physical-plan/src/joins/symmetric_hash_join.rs index a56ad1712aa8e..41abd7b79e107 100644 --- a/datafusion/physical-plan/src/joins/symmetric_hash_join.rs +++ b/datafusion/physical-plan/src/joins/symmetric_hash_join.rs @@ -47,8 +47,7 @@ use crate::joins::utils::{ matchable_join_keys, symmetric_join_output_partitioning, update_hash, }; use crate::projection::{ - ProjectionExec, join_allows_pushdown, join_table_borders, new_join_children, - physical_to_column_exprs, update_join_filter, update_join_on, + JoinData, ProjectionExec, try_pushdown_through_join_with_column_indices, }; use crate::stream::EmptyRecordBatchStream; use crate::{ @@ -589,69 +588,36 @@ impl ExecutionPlan for SymmetricHashJoinExec { &self, projection: &ProjectionExec, ) -> Result>> { - // Convert projected PhysicalExpr's to columns. If not possible, we cannot proceed. - let Some(projection_as_columns) = physical_to_column_exprs(projection.expr()) - else { - return Ok(None); - }; - - let (far_right_left_col_ind, far_left_right_col_ind) = join_table_borders( - self.left().schema().fields().len(), - &projection_as_columns, - ); - - if !join_allows_pushdown( - &projection_as_columns, - &self.schema(), - far_right_left_col_ind, - far_left_right_col_ind, - ) { - return Ok(None); - } - - let Some(new_on) = update_join_on( - &projection_as_columns[0..=far_right_left_col_ind as _], - &projection_as_columns[far_left_right_col_ind as _..], - self.on(), - self.left().schema().fields().len(), - ) else { - return Ok(None); - }; - - let new_filter = if let Some(filter) = self.filter() { - match update_join_filter( - &projection_as_columns[0..=far_right_left_col_ind as _], - &projection_as_columns[far_left_right_col_ind as _..], - filter, - self.left().schema().fields().len(), - ) { - Some(updated_filter) => Some(updated_filter), - None => return Ok(None), - } - } else { - None - }; - - let (new_left, new_right) = new_join_children( - &projection_as_columns, - far_right_left_col_ind, - far_left_right_col_ind, + let schema = self.schema(); + if let Some(JoinData { + projected_left_child, + projected_right_child, + join_filter, + join_on, + }) = try_pushdown_through_join_with_column_indices( + projection, self.left(), self.right(), - )?; - - SymmetricHashJoinExec::try_new( - Arc::new(new_left), - Arc::new(new_right), - new_on, - new_filter, - self.join_type(), - self.null_equality(), - self.right().output_ordering().cloned(), - self.left().output_ordering().cloned(), - self.partition_mode(), - ) - .map(|e| Some(Arc::new(e) as _)) + self.on(), + &schema, + self.filter(), + self.column_indices.as_slice(), + )? { + SymmetricHashJoinExec::try_new( + Arc::new(projected_left_child), + Arc::new(projected_right_child), + join_on, + join_filter, + self.join_type(), + self.null_equality(), + self.right().output_ordering().cloned(), + self.left().output_ordering().cloned(), + self.partition_mode(), + ) + .map(|e| Some(Arc::new(e) as _)) + } else { + Ok(None) + } } } diff --git a/datafusion/physical-plan/src/projection.rs b/datafusion/physical-plan/src/projection.rs index 6170cb702c504..6ada615b9eae3 100644 --- a/datafusion/physical-plan/src/projection.rs +++ b/datafusion/physical-plan/src/projection.rs @@ -650,60 +650,35 @@ pub fn try_pushdown_through_join( schema: &SchemaRef, filter: Option<&JoinFilter>, ) -> Result> { - // Convert projected expressions to columns. We can not proceed if this is not possible. - let Some(projection_as_columns) = physical_to_column_exprs(projection.expr()) else { - return Ok(None); - }; - - let (far_right_left_col_ind, far_left_right_col_ind) = - join_table_borders(join_left.schema().fields().len(), &projection_as_columns); - - if !join_allows_pushdown( - &projection_as_columns, - schema, - far_right_left_col_ind, - far_left_right_col_ind, - ) { - return Ok(None); - } - - let new_filter = if let Some(filter) = filter { - match update_join_filter( - &projection_as_columns[0..=far_right_left_col_ind as _], - &projection_as_columns[far_left_right_col_ind as _..], - filter, - join_left.schema().fields().len(), - ) { - Some(updated_filter) => Some(updated_filter), - None => return Ok(None), - } - } else { - None - }; - - let Some(new_on) = update_join_on( - &projection_as_columns[0..=far_right_left_col_ind as _], - &projection_as_columns[far_left_right_col_ind as _..], - join_on, - join_left.schema().fields().len(), - ) else { - return Ok(None); - }; + let left_field_count = join_left.schema().fields().len(); + let column_indices = schema + .fields() + .iter() + .enumerate() + .map(|(index, _)| { + if index < left_field_count { + ColumnIndex { + index, + side: JoinSide::Left, + } + } else { + ColumnIndex { + index: index - left_field_count, + side: JoinSide::Right, + } + } + }) + .collect::>(); - let (new_left, new_right) = new_join_children( - &projection_as_columns, - far_right_left_col_ind, - far_left_right_col_ind, + try_pushdown_through_join_with_column_indices( + projection, join_left, join_right, - )?; - - Ok(Some(JoinData { - projected_left_child: new_left, - projected_right_child: new_right, - join_filter: new_filter, - join_on: new_on, - })) + join_on, + schema, + filter, + &column_indices, + ) } pub(crate) fn try_pushdown_through_join_with_column_indices( @@ -959,7 +934,7 @@ pub fn new_join_children( /// join's `ColumnIndex`). Unlike [`new_join_children`], this does not infer /// child ownership from output position, so it is safe for join schemas whose /// output is not a plain `left ++ right` (used by the schema-aware -/// `try_pushdown_through_join`). +/// `try_pushdown_through_join_with_column_indices`). fn new_join_children_from_groups( left_proj: &[(Column, String)], right_proj: &[(Column, String)], From 1ff782eccf84f5e2114b055eefa9b14114bbd2b8 Mon Sep 17 00:00:00 2001 From: Jiawei Zhao Date: Sun, 12 Jul 2026 22:26:12 +0800 Subject: [PATCH 6/6] refactor(join): deprecate positional pushdown The public helper infers column ownership from output position, so it cannot represent synthetic join columns. It also encourages callers to rely on the left-then-right schema assumption. Keep it as a compatibility wrapper and expose the ColumnIndex-aware helper as its replacement. Document and validate the public mapping so malformed metadata returns planning errors instead of panicking. Refs #23010 Signed-off-by: Jiawei Zhao --- datafusion/physical-plan/src/projection.rs | 165 ++++++++++++++++++++- 1 file changed, 162 insertions(+), 3 deletions(-) diff --git a/datafusion/physical-plan/src/projection.rs b/datafusion/physical-plan/src/projection.rs index 6ada615b9eae3..d96ab3c944f58 100644 --- a/datafusion/physical-plan/src/projection.rs +++ b/datafusion/physical-plan/src/projection.rs @@ -46,7 +46,7 @@ use datafusion_common::config::ConfigOptions; use datafusion_common::tree_node::{ Transformed, TransformedResult, TreeNode, TreeNodeRecursion, }; -use datafusion_common::{DataFusionError, JoinSide, Result, internal_err}; +use datafusion_common::{DataFusionError, JoinSide, Result, internal_err, plan_err}; use datafusion_execution::TaskContext; use datafusion_expr::ExpressionPlacement; use datafusion_physical_expr::equivalence::ProjectionMapping; @@ -642,6 +642,10 @@ pub struct JoinData { pub join_on: JoinOn, } +#[deprecated( + since = "55.0.0", + note = "Use try_pushdown_through_join_with_column_indices instead" +)] pub fn try_pushdown_through_join( projection: &ProjectionExec, join_left: &Arc, @@ -681,7 +685,25 @@ pub fn try_pushdown_through_join( ) } -pub(crate) fn try_pushdown_through_join_with_column_indices( +/// Attempts to move a projection below a join by mapping each join output +/// column to the child column that produced it. +/// +/// `schema` is the complete output schema of the join, not either child's +/// schema. `column_indices` must contain one entry for each field in `schema`. +/// Each [`JoinSide::Left`] or [`JoinSide::Right`] entry identifies the source +/// child and uses an index relative to that child's schema. +/// +/// [`JoinSide::None`] identifies a column produced by the join itself, such as +/// a mark column. If `projection` references such a column, this function +/// returns `Ok(None)` because neither child can produce it. +/// +/// Returns `Ok(None)` when the projection cannot be pushed down safely. +/// +/// # Errors +/// +/// Returns an error if `column_indices` does not match `schema` or contains an +/// index outside the corresponding child schema. +pub fn try_pushdown_through_join_with_column_indices( projection: &ProjectionExec, join_left: &Arc, join_right: &Arc, @@ -690,6 +712,29 @@ pub(crate) fn try_pushdown_through_join_with_column_indices( filter: Option<&JoinFilter>, column_indices: &[ColumnIndex], ) -> Result> { + if column_indices.len() != schema.fields().len() { + return plan_err!( + "Column index mapping has {} entries but join schema has {} fields", + column_indices.len(), + schema.fields().len() + ); + } + // Validate each output-to-child mapping before using it to rewrite the + // projection. Synthetic outputs have no child index to validate. + for (output_index, column_index) in column_indices.iter().enumerate() { + let (side, child_field_count) = match column_index.side { + JoinSide::Left => ("left", join_left.schema().fields().len()), + JoinSide::Right => ("right", join_right.schema().fields().len()), + JoinSide::None => continue, + }; + if column_index.index >= child_field_count { + return plan_err!( + "Join output column {output_index} maps to {side} child column {}, but the child has {child_field_count} fields", + column_index.index + ); + } + } + // Convert projected expressions to columns. We can not proceed if this is not possible. let Some(projection_as_columns) = physical_to_column_exprs(projection.expr()) else { return Ok(None); @@ -702,7 +747,13 @@ pub(crate) fn try_pushdown_through_join_with_column_indices( let mut right_proj: Vec<(Column, String)> = Vec::new(); let mut seen_right = false; for (col, alias) in &projection_as_columns { - let origin = &column_indices[col.index()]; + let Some(origin) = column_indices.get(col.index()) else { + return plan_err!( + "Projection column {} is outside the {}-entry column index mapping", + col.index(), + column_indices.len() + ); + }; match origin.side { // Keep the "left block before right block" contiguity the current // pushdown supports; a left column after a right one is "mixed". @@ -1257,6 +1308,7 @@ mod tests { use super::*; use crate::common::collect; + use crate::empty::EmptyExec; use crate::filter_pushdown::PushedDown; use crate::statistics::StatisticsArgs; @@ -1292,6 +1344,113 @@ mod tests { Ok(()) } + #[test] + fn test_try_pushdown_through_join_validates_column_indices() -> Result<()> { + let child_schema = + Arc::new(Schema::new(vec![Field::new("i", DataType::Int32, false)])); + let left: Arc = + Arc::new(EmptyExec::new(Arc::clone(&child_schema))); + let right: Arc = Arc::new(EmptyExec::new(child_schema)); + let join_schema = Arc::new(Schema::new(vec![ + Field::new("left_i", DataType::Int32, false), + Field::new("right_i", DataType::Int32, false), + ])); + let join: Arc = + Arc::new(EmptyExec::new(Arc::clone(&join_schema))); + let projection = ProjectionExec::try_new( + vec![ProjectionExpr { + expr: Arc::new(Column::new("left_i", 0)), + alias: "left_i".to_string(), + }], + join, + )?; + + let Err(error) = try_pushdown_through_join_with_column_indices( + &projection, + &left, + &right, + &[], + &join_schema, + None, + &[], + ) else { + panic!("expected a mismatched mapping length to return an error"); + }; + assert!( + error.to_string().contains( + "Column index mapping has 0 entries but join schema has 2 fields" + ) + ); + + let invalid_child_index = [ + ColumnIndex { + index: 1, + side: JoinSide::Left, + }, + ColumnIndex { + index: 0, + side: JoinSide::Right, + }, + ]; + let Err(error) = try_pushdown_through_join_with_column_indices( + &projection, + &left, + &right, + &[], + &join_schema, + None, + &invalid_child_index, + ) else { + panic!("expected an invalid child index to return an error"); + }; + assert!(error.to_string().contains( + "Join output column 0 maps to left child column 1, but the child has 1 fields" + )); + + let wider_join_schema = Arc::new(Schema::new(vec![ + Field::new("left_i", DataType::Int32, false), + Field::new("right_i", DataType::Int32, false), + Field::new("extra", DataType::Int32, false), + ])); + let wider_join: Arc = + Arc::new(EmptyExec::new(wider_join_schema)); + let out_of_mapping_projection = ProjectionExec::try_new( + vec![ProjectionExpr { + expr: Arc::new(Column::new("extra", 2)), + alias: "extra".to_string(), + }], + wider_join, + )?; + let valid_child_indices = [ + ColumnIndex { + index: 0, + side: JoinSide::Left, + }, + ColumnIndex { + index: 0, + side: JoinSide::Right, + }, + ]; + let Err(error) = try_pushdown_through_join_with_column_indices( + &out_of_mapping_projection, + &left, + &right, + &[], + &join_schema, + None, + &valid_child_indices, + ) else { + panic!("expected an out-of-mapping projection to return an error"); + }; + assert!( + error.to_string().contains( + "Projection column 2 is outside the 2-entry column index mapping" + ) + ); + + Ok(()) + } + #[test] fn test_join_table_borders() -> Result<()> { let projections = vec![