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 <Phoenix500526@163.com>

Author: Phoenix500526

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<Option<Arc<dyn ExecutionPlan>>> {
-        // 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)
+        }
     }
 }
 

datafusion/physical-plan/src/projection.rs

@@ -650,60 +650,35 @@ pub fn try_pushdown_through_join(
     schema: &SchemaRef,
     filter: Option<&JoinFilter>,
 ) -> Result<Option<JoinData>> {
-    // 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::<Vec<_>>();
 
-    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)],