feat: derive SortMergeJoin sort options from input ordering and allow SMJ for a single partition

[bert-beyondloops]

Which issue does this PR close?

• Closes SortMergeJoin: derive sort options from input ordering, and allow it with a single partition when prefer_hash_join=false #26260

Rationale for this change

With prefer_hash_join = false, the planner adds SortExecs under SortMergeJoinExec even when both inputs are already sorted on the join keys:

1. The sort options are hardcoded to SortOptions::default() (ASC NULLS FIRST) for every key. Inputs sorted ASC NULLS LAST, DESC, or with different options per key don't satisfy this, so EnforceSorting sorts both sides again.
2. SMJ is only planned when target_partitions > 1 && repartition_joins. With target_partitions = 1, prefer_hash_join = false is silently ignored and a HashJoinExec is planned. With a single partition no repartitioning happens, so repartition_joins has no effect there.

What changes are included in this PR?

Both changes are in physical_planner.rs.

1. The sort options are derived from the inputs' declared ordering (EquivalenceProperties::ordering_satisfy):
   • Per key, try these candidates in order: default, flipped nulls, flipped direction, both flipped.
   • Backtrack across keys until both inputs satisfy the whole key list.
   • If no combination works, fall back to SortOptions::default(), so existing plans are unchanged.
2. The SMJ condition changes from target_partitions() > 1 && repartition_joins() && ... to (target_partitions() == 1 || repartition_joins()) && .... For target_partitions > 1, behavior is unchanged.

What is the testing strategy for this PR?

New SLT's :

• New sort_merge_join_sort_options.slt covers inputs ordered ASC NULLS LAST, DESC, and mixed options per key, checking that no extra SortExec is added.
• New cases in sort_merge_join.slt for target_partitions = 1.

Are there any user-facing changes?

No

[quwin]

Thanks for your work @bert-beyondloops!
I ran the focused sort-merge join SQL logic tests against this branch. They currently fail on three EXPLAIN snapshots because the expected plans show a separate ProjectionExec, while the actual plans put the projection on SortMergeJoinExec. I left comments on those snapshots with the plans produced by the branch.

[quwin]

This expected plan seems to be stale. The focused SQL logic run fails the EXPLAIN: DataFusion puts the projection on SortMergeJoinExec, rather than emitting a separate ProjectionExec.

Suggested change

	01)ProjectionExec: expr=[k1@0 as k1, k2@1 as k2, v@2 as v, w@5 as w]
	02)--SortMergeJoinExec: join_type=Inner, on=[(k1@0, k1@0), (k2@1, k2@1)]
	03)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_merge_join_sort_options/left.parquet]]}, projection=[k1, k2, v], output_ordering=[k1@0 DESC NULLS LAST, k2@1 ASC], file_type=parquet
	04)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_merge_join_sort_options/right.parquet]]}, projection=[k1, k2, w], output_ordering=[k1@0 DESC NULLS LAST, k2@1 ASC], file_type=parquet
	01)SortMergeJoinExec: join_type=Inner, on=[(k1@0, k1@0), (k2@1, k2@1)], projection=[k1@0, k2@1, v@2, w@5]
	02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_merge_join_sort_options/left.parquet]]}, projection=[k1, k2, v], output_ordering=[k1@0 DESC NULLS LAST, k2@1 ASC], file_type=parquet
	03)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_merge_join_sort_options/right.parquet]]}, projection=[k1, k2, w], output_ordering=[k1@0 DESC NULLS LAST, k2@1 ASC], file_type=parquet

[quwin]

Thanks for resolving this!

[quwin]

This single-partition EXPLAIN snapshot seems to fail. The actual plan puts projection=[a@0, b@1, b@3] on SortMergeJoinExec, rather than emitting the expected ProjectionExec.

Suggested change

	01)ProjectionExec: expr=[a@0 as a, b@1 as b, b@3 as b]
	02)--SortMergeJoinExec: join_type=Inner, on=[(a@0, a@0)]
	03)----SortExec: expr=[a@0 ASC], preserve_partitioning=[false]
	04)------DataSourceExec: partitions=1, partition_sizes=[1]
	05)----SortExec: expr=[a@0 ASC], preserve_partitioning=[false]
	06)------DataSourceExec: partitions=1, partition_sizes=[1]
	01)SortMergeJoinExec: join_type=Inner, on=[(a@0, a@0)], projection=[a@0, b@1, b@3]
	02)--SortExec: expr=[a@0 ASC], preserve_partitioning=[false]
	03)----DataSourceExec: partitions=1, partition_sizes=[1]
	04)--SortExec: expr=[a@0 ASC], preserve_partitioning=[false]
	05)----DataSourceExec: partitions=1, partition_sizes=[1]

[quwin]

One follow-up from checking the suggested changes locally:

[quwin]

This fallback case has the same projection mismatch.

Suggested change

	01)ProjectionExec: expr=[k1@0 as k1, k2@1 as k2, v@2 as v, w@5 as w]
	02)--SortMergeJoinExec: join_type=Inner, on=[(k1@0, k1@0), (k2@1, k2@1)]
	03)----SortExec: expr=[k1@0 ASC, k2@1 ASC], preserve_partitioning=[false]
	04)------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_merge_join_sort_options/left.parquet]]}, projection=[k1, k2, v], file_type=parquet, sort_order_for_reorder=[k1@0 ASC, k2@1 ASC]
	05)----SortExec: expr=[k1@0 ASC, k2@1 ASC], preserve_partitioning=[false]
	06)------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_merge_join_sort_options/right.parquet]]}, projection=[k1, k2, w], file_type=parquet, sort_order_for_reorder=[k1@0 ASC, k2@1 ASC]
	01)SortMergeJoinExec: join_type=Inner, on=[(k1@0, k1@0), (k2@1, k2@1)], projection=[k1@0, k2@1, v@2, w@5]
	02)--SortExec: expr=[k1@0 ASC, k2@1 ASC], preserve_partitioning=[false]
	03)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_merge_join_sort_options/left.parquet]]}, projection=[k1, k2, v], file_type=parquet, sort_order_for_reorder=[k1@0 ASC, k2@1 ASC]
	04)--SortExec: expr=[k1@0 ASC, k2@1 ASC], preserve_partitioning=[false]
	05)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_merge_join_sort_options/right.parquet]]}, projection=[k1, k2, w], file_type=parquet, sort_order_for_reorder=[k1@0 ASC, k2@1 ASC]

[quwin]

Could we restore the test runner’s initial value of 4 explicitly here and in sort_merge_join.slt?
RESET restores the machine default instead. After correcting the snapshots, both files fail with target_partitions: 4 -> 8 on my machine.

Suggested change

	reset datafusion.execution.target_partitions;
	SET datafusion.execution.target_partitions = 4;

[bert-beyondloops]

I applied a patch built on top of the 55.1 release. Since then, SortMergeJoin on main has gained embedded projection support, so I updated the affected explain plans to reflect that change.

[codecov-commenter]

Codecov Report

❌ Patch coverage is 92.18750% with 5 lines in your changes missing coverage. Please review.
✅ Project coverage is 82.79%. Comparing base (2c568b4) to head (a9d122f).
⚠️ Report is 9 commits behind head on main.

Files with missing lines	Patch %	Lines
datafusion/core/src/physical_planner.rs	92.18%	1 Missing and 4 partials ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main   #26261      +/-   ##
==========================================
- Coverage   82.83%   82.79%   -0.04%     
==========================================
  Files        1149     1149              
  Lines      452798   454206    +1408     
  Branches   452798   454206    +1408     
==========================================
+ Hits       375082   376079     +997     
- Misses      54893    55267     +374     
- Partials    22823    22860      +37     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[quwin]

I reproduced a planning slowdown when the inputs are ordered on all but the final join key, with non-nullable keys and prefer_hash_join=false.
Fastest elapsed time over three (3) runs on my machine:

Join keys	main (0d79252f6)	This PR (a9d122f0e)
4	0.025 s	0.032 s
8	0.026 s	0.220 s
12	0.034 s	5.658 s
14	0.039 s	28.377 s

Could we add a physical-planning benchmark to sql_planner for this case?
It can probably be tested better with run benchmark sql_planner, but I'm not whitelisted to call it.