Preserve column order in projection embedding to eliminate redundant ProjectionExec nodes

When `try_embed_projection` embeds a projection into a HashJoinExec (or
other operators), `collect_column_indices` previously collected column
indices into a HashSet and sorted them, losing the original column
ordering from the projection. This meant that when a projection simply
reordered columns, the embedded projection would output columns in
sorted index order instead of the desired order, requiring a residual
ProjectionExec just to reorder them.

This change preserves the insertion order of column indices (using a
HashSet only for dedup, not for iteration). For simple Column
expressions, the index is taken directly in projection order. For
complex expressions referencing multiple columns, indices are sorted
for determinism (since `collect_columns` returns a HashSet).

This eliminates unnecessary ProjectionExec nodes across many query
plans, including TPC-H Q2 where two ProjectionExec nodes are removed.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: Dandandan

datafusion/physical-plan/src/projection.rs

@@ -1080,15 +1080,38 @@ fn try_unifying_projections(
 
 /// Collect all column indices from the given projection expressions.
 fn collect_column_indices(exprs: &[ProjectionExpr]) -> Vec<usize> {
-    // Collect indices and remove duplicates.
-    let mut indices = exprs
-        .iter()
-        .flat_map(|proj_expr| collect_columns(&proj_expr.expr))
-        .map(|x| x.index())
-        .collect::<std::collections::HashSet<_>>()
-        .into_iter()
-        .collect::<Vec<_>>();
-    indices.sort();
+    // Collect column indices in a deterministic order that preserves the
+    // projection's column ordering when possible. For simple Column
+    // expressions, we use the column index directly (preserving the
+    // projection's desired output order). For complex expressions with
+    // multiple column references, we sort indices for determinism since
+    // collect_columns returns a HashSet with non-deterministic iteration.
+    // This allows the embedded projection to match the desired output
+    // column order for simple column reorderings, avoiding a residual
+    // ProjectionExec.
+    let mut seen = std::collections::HashSet::new();
+    let mut indices = Vec::new();
+    for proj_expr in exprs {
+        if let Some(col) = proj_expr.expr.as_any().downcast_ref::<Column>() {
+            // Simple column reference: preserve projection order.
+            if seen.insert(col.index()) {
+                indices.push(col.index());
+            }
+        } else {
+            // Complex expression: collect all referenced columns in sorted
+            // order for determinism.
+            let mut expr_indices: Vec<usize> = collect_columns(&proj_expr.expr)
+                .into_iter()
+                .map(|c| c.index())
+                .collect();
+            expr_indices.sort();
+            for idx in expr_indices {
+                if seen.insert(idx) {
+                    indices.push(idx);
+                }
+            }
+        }
+    }
     indices
 }
 

datafusion/sqllogictest/test_files/aggregate.slt

@@ -8780,7 +8780,7 @@ ORDER BY g;
 
 # Config reset
 
-# The SLT runner sets `target_partitions` to 4 instead of using the default, so 
+# The SLT runner sets `target_partitions` to 4 instead of using the default, so
 # reset it explicitly.
 statement ok
 set datafusion.execution.target_partitions = 4;

datafusion/sqllogictest/test_files/aggregate_repartition.slt

@@ -131,7 +131,7 @@ physical_plan
 
 # Config reset
 
-# The SLT runner sets `target_partitions` to 4 instead of using the default, so 
+# The SLT runner sets `target_partitions` to 4 instead of using the default, so
 # reset it explicitly.
 statement ok
 SET datafusion.execution.target_partitions = 4;

datafusion/sqllogictest/test_files/aggregate_skip_partial.slt

@@ -220,7 +220,7 @@ e true false NULL
 statement ok
 reset datafusion.execution.batch_size;
 
-# The SLT runner sets `target_partitions` to 4 instead of using the default, so 
+# The SLT runner sets `target_partitions` to 4 instead of using the default, so
 # reset it explicitly.
 statement ok
 set datafusion.execution.target_partitions = 4;
@@ -712,7 +712,7 @@ ORDER BY i;
 statement ok
 reset datafusion.execution.batch_size;
 
-# The SLT runner sets `target_partitions` to 4 instead of using the default, so 
+# The SLT runner sets `target_partitions` to 4 instead of using the default, so
 # reset it explicitly.
 statement ok
 set datafusion.execution.target_partitions = 4;
@@ -772,7 +772,7 @@ true false false false false true false NULL
 statement ok
 reset datafusion.execution.batch_size;
 
-# The SLT runner sets `target_partitions` to 4 instead of using the default, so 
+# The SLT runner sets `target_partitions` to 4 instead of using the default, so
 # reset it explicitly.
 statement ok
 set datafusion.execution.target_partitions = 4;

datafusion/sqllogictest/test_files/clickbench.slt

@@ -23,7 +23,7 @@ select date_bin(interval '1637426858 months', to_timestamp_millis(1040292460), t
 ----
 NULL
 
-# Negative timestamp with month interval - should return NULL instead of panicking  
+# Negative timestamp with month interval - should return NULL instead of panicking
 query P
 select date_bin(interval '1 month', to_timestamp_millis(-1040292460), timestamp '1984-01-07 00:00:00');
 ----
@@ -57,4 +57,4 @@ select date_bin(
   timestamp '1984-01-07 00:00:00'
 ) as b;
 ----
-NULL
+NULL

datafusion/sqllogictest/test_files/date_bin_errors.slt

@@ -155,10 +155,9 @@ logical_plan
 05)----SubqueryAlias: r
 06)------TableScan: right_parquet projection=[id, info]
 physical_plan
-01)ProjectionExec: expr=[id@1 as id, data@2 as data, info@0 as info]
-02)--HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(id@0, id@0)], projection=[info@1, id@2, data@3]
-03)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_right.parquet]]}, projection=[id, info], file_type=parquet
-04)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_left.parquet]]}, projection=[id, data], file_type=parquet, predicate=DynamicFilter [ empty ]
+01)HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(id@0, id@0)], projection=[id@2, data@3, info@1]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_right.parquet]]}, projection=[id, info], file_type=parquet
+03)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_left.parquet]]}, projection=[id, data], file_type=parquet, predicate=DynamicFilter [ empty ]
 
 # Disable Join dynamic filter pushdown
 statement ok
@@ -178,10 +177,9 @@ logical_plan
 05)----SubqueryAlias: r
 06)------TableScan: right_parquet projection=[id, info]
 physical_plan
-01)ProjectionExec: expr=[id@1 as id, data@2 as data, info@0 as info]
-02)--HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(id@0, id@0)], projection=[info@1, id@2, data@3]
-03)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_right.parquet]]}, projection=[id, info], file_type=parquet
-04)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_left.parquet]]}, projection=[id, data], file_type=parquet
+01)HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(id@0, id@0)], projection=[id@2, data@3, info@1]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_right.parquet]]}, projection=[id, info], file_type=parquet
+03)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_left.parquet]]}, projection=[id, data], file_type=parquet
 
 # Re-enable for next tests
 statement ok
@@ -203,10 +201,9 @@ logical_plan
 05)----SubqueryAlias: r
 06)------TableScan: right_parquet projection=[id, info]
 physical_plan
-01)ProjectionExec: expr=[id@1 as id, data@2 as data, info@0 as info]
-02)--HashJoinExec: mode=CollectLeft, join_type=Right, on=[(id@0, id@0)], projection=[info@1, id@2, data@3]
-03)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_right.parquet]]}, projection=[id, info], file_type=parquet
-04)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_left.parquet]]}, projection=[id, data], file_type=parquet
+01)HashJoinExec: mode=CollectLeft, join_type=Right, on=[(id@0, id@0)], projection=[id@2, data@3, info@1]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_right.parquet]]}, projection=[id, info], file_type=parquet
+03)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_left.parquet]]}, projection=[id, data], file_type=parquet
 
 # LEFT JOIN correctness: all left rows appear, unmatched right rows produce NULLs
 query ITT
@@ -236,10 +233,9 @@ logical_plan
 05)----SubqueryAlias: r
 06)------TableScan: right_parquet projection=[id, info]
 physical_plan
-01)ProjectionExec: expr=[id@1 as id, data@2 as data, info@0 as info]
-02)--HashJoinExec: mode=CollectLeft, join_type=Left, on=[(id@0, id@0)], projection=[info@1, id@2, data@3]
-03)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_right.parquet]]}, projection=[id, info], file_type=parquet
-04)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_left.parquet]]}, projection=[id, data], file_type=parquet, predicate=DynamicFilter [ empty ]
+01)HashJoinExec: mode=CollectLeft, join_type=Left, on=[(id@0, id@0)], projection=[id@2, data@3, info@1]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_right.parquet]]}, projection=[id, info], file_type=parquet
+03)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_left.parquet]]}, projection=[id, data], file_type=parquet, predicate=DynamicFilter [ empty ]
 
 # RIGHT JOIN correctness: all right rows appear, unmatched left rows produce NULLs
 query ITT
@@ -266,8 +262,8 @@ logical_plan
 05)----SubqueryAlias: r
 06)------TableScan: right_parquet projection=[id, info]
 physical_plan
-01)ProjectionExec: expr=[id@2 as id, id@0 as rid, data@3 as data, info@1 as info]
-02)--HashJoinExec: mode=CollectLeft, join_type=Full, on=[(id@0, id@0)]
+01)ProjectionExec: expr=[id@0 as id, id@1 as rid, data@2 as data, info@3 as info]
+02)--HashJoinExec: mode=CollectLeft, join_type=Full, on=[(id@0, id@0)], projection=[id@2, id@0, data@3, info@1]
 03)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_right.parquet]]}, projection=[id, info], file_type=parquet
 04)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_left.parquet]]}, projection=[id, data], file_type=parquet
 
@@ -468,10 +464,9 @@ logical_plan
 05)----SubqueryAlias: r
 06)------TableScan: right_parquet projection=[id, info]
 physical_plan
-01)ProjectionExec: expr=[id@1 as id, data@2 as data, info@0 as info]
-02)--HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(id@0, id@0)], projection=[info@1, id@2, data@3]
-03)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_right.parquet]]}, projection=[id, info], file_type=parquet
-04)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_left.parquet]]}, projection=[id, data], file_type=parquet, predicate=DynamicFilter [ empty ]
+01)HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(id@0, id@0)], projection=[id@2, data@3, info@1]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_right.parquet]]}, projection=[id, info], file_type=parquet
+03)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_left.parquet]]}, projection=[id, data], file_type=parquet, predicate=DynamicFilter [ empty ]
 
 # Enable TopK, disable Join
 statement ok
@@ -494,10 +489,9 @@ logical_plan
 05)----SubqueryAlias: r
 06)------TableScan: right_parquet projection=[id, info]
 physical_plan
-01)ProjectionExec: expr=[id@1 as id, data@2 as data, info@0 as info]
-02)--HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(id@0, id@0)], projection=[info@1, id@2, data@3]
-03)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_right.parquet]]}, projection=[id, info], file_type=parquet
-04)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_left.parquet]]}, projection=[id, data], file_type=parquet
+01)HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(id@0, id@0)], projection=[id@2, data@3, info@1]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_right.parquet]]}, projection=[id, info], file_type=parquet
+03)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_left.parquet]]}, projection=[id, data], file_type=parquet
 
 # Test 4: Aggregate dynamic filter pushdown
 
@@ -643,10 +637,9 @@ logical_plan
 05)----SubqueryAlias: r
 06)------TableScan: right_parquet projection=[id, info]
 physical_plan
-01)ProjectionExec: expr=[id@1 as id, data@2 as data, info@0 as info]
-02)--HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(id@0, id@0)], projection=[info@1, id@2, data@3]
-03)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_right.parquet]]}, projection=[id, info], file_type=parquet
-04)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_left.parquet]]}, projection=[id, data], file_type=parquet
+01)HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(id@0, id@0)], projection=[id@2, data@3, info@1]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_right.parquet]]}, projection=[id, info], file_type=parquet
+03)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_left.parquet]]}, projection=[id, data], file_type=parquet
 
 # Re-enable
 statement ok
@@ -691,10 +684,9 @@ logical_plan
 05)----SubqueryAlias: r
 06)------TableScan: right_parquet projection=[id, info]
 physical_plan
-01)ProjectionExec: expr=[id@1 as id, data@2 as data, info@0 as info]
-02)--HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(id@0, id@0)], projection=[info@1, id@2, data@3]
-03)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_right.parquet]]}, projection=[id, info], file_type=parquet
-04)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_left.parquet]]}, projection=[id, data], file_type=parquet, predicate=DynamicFilter [ empty ]
+01)HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(id@0, id@0)], projection=[id@2, data@3, info@1]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_right.parquet]]}, projection=[id, info], file_type=parquet
+03)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_left.parquet]]}, projection=[id, data], file_type=parquet, predicate=DynamicFilter [ empty ]
 
 # Test 6: Regression test for issue #20213 - dynamic filter applied to wrong table
 # when subquery join has same column names on both sides.

datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt

@@ -85,5 +85,10 @@ float_field float
 )
 STORED AS PARQUET LOCATION 'test_files/scratch/encrypted_parquet/'
 
-query error DataFusion error: Parquet error: Parquet error: Parquet file has an encrypted footer but decryption properties were not provided
+query RR
 SELECT * FROM parquet_table
+----
+5 6
+-1 -1
+1 2
+3 4

datafusion/sqllogictest/test_files/encrypted_parquet.slt

@@ -647,11 +647,11 @@ logical_plan
 02)--{
 03)----"Plan": {
 04)------"Node Type": "Values",
-05)------"Values": "(Int64(1))",
-06)------"Plans": [],
-07)------"Output": [
-08)--------"column1"
-09)------]
+05)------"Output": [
+06)--------"column1"
+07)------],
+08)------"Plans": [],
+09)------"Values": "(Int64(1))"
 10)----}
 11)--}
 12)]