[gstamatakis95]

Which issue does this PR close?

• Closes Preserve partitioning through co-partitioned Full Outer Joins #25436.

Rationale for this change

A partitioned Full Outer Join reports UnknownPartitioning even when both inputs are already partitioned the same way on the join keys. Every output row stays in the partition its key was routed to, so the report throws away a real guarantee. A later aggregate or join on the join key then gets a RepartitionExec it does not need.

The left key alone cannot describe the output, because right only rows carry NULL in the left key in every partition. The key that agrees with the partition number is coalesce(left_key, right_key).

What changes are included in this PR?

• symmetric_join_output_partitioning takes the join keys and the join's equivalence properties. For a Full join whose inputs pass the existing co-partitioning check, it reports the left partitioning restated on CASE WHEN l IS NOT NULL THEN l ELSE r END for each key pair, through Partitioning::adapt. Hash keeps its count, Range keeps its split points and sort options.
• The join records that both coalesce orders of each key pair are equal, so a parent keyed on COALESCE(r.k, l.k) matches as well.
• The optimization only applies when every join key is a column. A computed key such as coalesce(a, 0) is not NULL on an unmatched row, so the coalesced key would be wrong. SQL planning already projects computed keys into columns below the join, so this check covers plans built directly through the physical plan API.
• It also stays off when an input holds a key column equal to a computed expression, for example through a filter c = coalesce(y, 0). The join output inherits that equality, but it fails on unmatched rows, and a parent could then skip a shuffle it needs. This is the check that protects SQL plans.
• In every other case the join still reports UnknownPartitioning.
• HashJoinExec in partitioned mode, SortMergeJoinExec, and SymmetricHashJoinExec pass their keys and equivalence properties into the helper.

The CASE form is what the planner produces for a two argument COALESCE, so a parent's expression compares equal structurally.

What is the testing strategy for this PR?

• Unit tests in joins/utils.rs cover the Hash, Range, and multi key outputs, the fallbacks to UnknownPartitioning including a computed key and a key held equal to a computed expression by a filter, projections on the hash and sort merge joins that keep, reorder, or drop a key, the symmetric hash join path, and the equivalence of both coalesce orders.
• enforce_distribution.rs tests show no shuffle for a parent join on the coalesced key over Hash and Range inputs, in either key order and over a swapped join, and a shuffle for a parent join on the plain key.
• range_partitioning.slt tests 50 to 54 show plans and results for an aggregate on COALESCE, a chained Full join, the hash repartitioned variant, the negative GROUP BY l.range_key case with its single NULL group, and the reversed coalesce order.
• repartition.slt runs small nullable tables over four partitions: the filter equality case, which keeps its shuffle and returns one group, a computed join key with an unmatched row feeding an aggregate, and NULL join keys under = and IS NOT DISTINCT FROM, which need no shuffle and return a single NULL group.
• The existing dataframe test that asserted UnknownPartitioning for Full joins now asserts the coalesced Hash partitioning.

Are there any user-facing changes?

Plans with a partitioned Full join on column keys followed by an operator keyed on COALESCE of the join keys lose a RepartitionExec. Results do not change. In a chain of Full joins on the coalesced key, the outer join reports UnknownPartitioning when its input still carries the inner join's key columns, because the inner join's coalesce equivalence then shares a class with the outer key. Results stay correct, and a parent of the outer join keeps one shuffle. symmetric_join_output_partitioning is crate private, so there is no public API change.

[codecov-commenter]

Codecov Report

❌ Patch coverage is 96.79487% with 10 lines in your changes missing coverage. Please review.
✅ Project coverage is 82.84%. Comparing base (2c568b4) to head (ed37a26).
⚠️ Report is 8 commits behind head on main.

Files with missing lines	Patch %	Lines
datafusion/physical-plan/src/joins/utils.rs	97.58%	0 Missing and 7 partials ⚠️
...tafusion/physical-plan/src/joins/hash_join/exec.rs	85.71%	1 Missing ⚠️
...on/physical-plan/src/joins/sort_merge_join/exec.rs	85.71%	1 Missing ⚠️
...ion/physical-plan/src/joins/symmetric_hash_join.rs	87.50%	1 Missing ⚠️

Additional details and impacted files

@@           Coverage Diff            @@
##             main   #25554    +/-   ##
========================================
  Coverage   82.83%   82.84%            
========================================
  Files        1149     1149            
  Lines      452798   453103   +305     
  Branches   452798   453103   +305     
========================================
+ Hits       375082   375372   +290     
- Misses      54893    54899     +6     
- Partials    22823    22832     +9     

☔ 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.

[kosiew]

@gstamatakis95,

Thanks for working on this. Preserving partitioning across full outer joins is a useful optimization, and the added regression coverage is substantial.

I found one correctness issue with computed join keys that needs to be addressed before merging. I've also left an optional suggestion for sort-merge join projection coverage.

[kosiew]

Computed join keys can invalidate the reported partitioning and coalesce-order equivalence. For example, with a left key CASE WHEN a IS NOT NULL THEN a ELSE 0 END and right key c, right-only rows produce 0 for the left-first coalesce but c for the right-first one, potentially causing incorrect downstream aggregation. Please restrict this optimization to column keys or validate that computed keys preserve their values and become NULL on null-extended rows, falling back to UnknownPartitioning otherwise. Please also add a regression test covering unmatched rows and a downstream aggregate.

[gstamatakis95]

Good catch, thanks! It now only kicks in when both join keys are plain columns, otherwise it falls back to UnknownPartitioning like before. I added a test as well

[kosiew]

Could we also test sort-merge join projections? SortMergeJoinExec handles property projection separately, so it would be useful to verify that retaining or reordering both keys preserves the remapped coalesced partitioning, while dropping one key invalidates it. This is optional coverage and can be deferred.

[gstamatakis95]

Added three tests for this through with_projection