feat: add projection support to SortMergeJoinExec (#24517)

## Which issue does this PR close?

- Closes #24518

## Rationale for this change

`SortMergeJoinExec` emits every column of both inputs.
We can add projection support to only produce the output columns (saving
some compute / copy).

```sql
select * from t1 right join t2 on t1.c3 = t2.c3   -- c3 needs a cast
```

```text
ProjectionExec: expr=[c1@0 as c1, ..., c4@8 as c4]
  SortMergeJoinExec: join_type=Right, on=[(CAST(t1.c3 AS Decimal128(10, 2))@4, c3@2)]
```

The cast column is only there to be joined on, and every operator above
the join
carries it until the projection removes it.

## What changes are included in this PR?

`SortMergeJoinExec` takes an optional projection, set with
`with_projection`, the same
shape as `HashJoinExec`'s.

`try_swapping_with_projection` embeds the projection into the join when
it cannot push
it into the children, so the query above becomes:

```text
SortMergeJoinExec: join_type=Right, on=[(CAST(t1.c3 AS Decimal128(10, 2))@4, c3@2)], projection=[c1@0, c2@1, c3@2, c4@3, c1@5, c2@6, c3@7, c4@8]
```

The change doesn't bring a big speedup but helps aligning with other
join types and helping simplify join optimization in other areas (e.g.
join enumeration).

## Are these changes tested?

Yes:

- a projection pushdown test for a projection that interleaves the two
sides, which
  the existing pushdown cannot handle
- a serialization round trip
- existing sqllogictests, whose plans lose a `ProjectionExec` in four
places


## Are there any user-facing changes?
Explain / proto changes.

---------

Co-authored-by: Claude Opus 5 <noreply@anthropic.com>

Author: Dandandan

datafusion/core/tests/physical_optimizer/projection_pushdown.rs

@@ -49,8 +49,8 @@ use datafusion_physical_plan::coop::CooperativeExec;
 use datafusion_physical_plan::filter::{FilterExec, FilterExecBuilder};
 use datafusion_physical_plan::joins::utils::{ColumnIndex, JoinFilter};
 use datafusion_physical_plan::joins::{
-    HashJoinExec, NestedLoopJoinExec, PartitionMode, StreamJoinPartitionMode,
-    SymmetricHashJoinExec,
+    HashJoinExec, NestedLoopJoinExec, PartitionMode, SortMergeJoinExec,
+    StreamJoinPartitionMode, SymmetricHashJoinExec,
 };
 use datafusion_physical_plan::projection::{ProjectionExec, ProjectionExpr, update_expr};
 use datafusion_physical_plan::repartition::RepartitionExec;
@@ -1875,3 +1875,46 @@ fn test_filter_with_embedded_projection_after_renaming_projection() -> Result<()
 
     Ok(())
 }
+
+#[test]
+fn test_sort_merge_join_interleaved_projection_embeds() -> Result<()> {
+    // SELECT t1.c, t2.c, t1.a FROM t1 JOIN t2 ON t1.b = t2.c
+    // Taking a column from each side in turn leaves the sides interleaved, which
+    // cannot be pushed into the children, so the join applies it itself.
+    let join = Arc::new(SortMergeJoinExec::try_new(
+        create_simple_csv_exec(),
+        create_simple_csv_exec(),
+        vec![(Arc::new(Column::new("b", 1)), Arc::new(Column::new("c", 2)))],
+        None,
+        JoinType::Inner,
+        vec![SortOptions::default()],
+        NullEquality::NullEqualsNothing,
+    )?);
+    let projection: Arc<dyn ExecutionPlan> = Arc::new(ProjectionExec::try_new(
+        vec![
+            ProjectionExpr::new(Arc::new(Column::new("c", 2)), "c_from_left"),
+            ProjectionExpr::new(Arc::new(Column::new("c", 7)), "c_from_right"),
+            ProjectionExpr::new(Arc::new(Column::new("a", 0)), "a_from_left"),
+        ],
+        join,
+    )?);
+
+    let after_optimize =
+        ProjectionPushdown::new().optimize(projection, &ConfigOptions::new())?;
+    let actual = displayable(after_optimize.as_ref())
+        .indent(true)
+        .to_string()
+        .trim()
+        .to_string();
+    assert_snapshot!(
+        actual,
+        @r"
+    ProjectionExec: expr=[c@0 as c_from_left, c@1 as c_from_right, a@2 as a_from_left]
+      SortMergeJoinExec: join_type=Inner, on=[(b@1, c@2)], projection=[c@2, c@7, a@0]
+        DataSourceExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], file_type=csv, has_header=false
+        DataSourceExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], file_type=csv, has_header=false
+    "
+    );
+
+    Ok(())
+}