Preserve partitioning through co-partitioned Full Outer Joins
Is your feature request related to a problem or challenge?
When executing a partitioned join between two co-partitioned inputs, symmetric_join_output_partitioning in datafusion/physical-plan/src/joins/utils.rs computes the output partitioning:
https://github.com/apache/datafusion/blob/c4f72bad34249798182ac7de605eb63ab134f26e/datafusion/physical-plan/src/joins/utils.rs#L1979-L1982
For JoinType::Full, it unconditionally returns Partitioning::UnknownPartitioning:
JoinType::Full => {
// We could also use left partition count as they are necessarily equal.
Partitioning::UnknownPartitioning(right_partitioning.partition_count())
}In a partitioned Full Outer Join where both inputs are co-partitioned on the join key (either via matching Partitioning::Range split points or compatible Partitioning::Hash):
- Matching rows are produced in partition
i. - Unmatched left rows are produced in partition
i. - Unmatched right rows are produced in partition
i.
All output rows remain partitioned by the join key across the existing partitions. Unconditionally dropping to UnknownPartitioning throws away this partition guarantee. If a downstream operator subsequently joins or aggregates on the join key (for example, in a multi-way join pipeline), DataFusion is forced to insert an unnecessary repartition shuffle.
Describe the solution you'd like
In symmetric_join_output_partitioning, when both inputs are co-partitioned on the join keys for a Partitioned Full Outer Join, preserve the output partitioning on the join key (projected or coalesced to the output schema) rather than degrading to UnknownPartitioning.
Describe alternatives you've considered
Downstream queries must currently re-shuffle after a full outer join even when operating on pre-partitioned datasets.
Additional context
Part of epic #25421.
Source: apache/datafusion