Repository navigation
Conversation
github-actions
Bot
added
optimizer
1979696 to
4735398
Compare
Codecov Report❌ Patch coverage is @@ Coverage Diff @@
## main #25304 +/- ##
==========================================
- Coverage 81.91% 81.91% -0.01%
==========================================
Files 1134 1134
Lines 425708 425722 +14
Branches 425708 425722 +14
==========================================
+ Hits 348726 348729 +3
Misses 56305 56305
- Partials 20677 20688 +11
... and 8 files with indirect coverage changes 🚀 New features to boost your workflow:
|
| if hash_join.inputs_satisfy_partitioned_requirements()? { | ||
| return Ok(None); | ||
| } | ||
|
|
There was a problem hiding this comment.
Thanks, @stuhood. I tested the Hive-partitioned case and found a wrong-results regression with preserve_file_partitions=1.
There was a problem hiding this comment.
This can drop rows with preserve_file_partitions=1: Hive scans advertise Hash([k], n), but different value sets can place matching keys in different partitions (#23436).
With this branch and target_partitions=3:
- Dim {A,B,C,D} → [[A,D],[B],[C]]
- Fact {B,C,D} → [[B],[C],[D]]
Joining on k returns 0 rows, versus 3 with this hunk reverted. The underlying bug already affects non-broadcast joins; this PR extends it to small tables previously protected by CollectLeft.
Dynamic-filter pushdown already rejects Hash/Hash under this setting, which also explains the lost filters in the updated plans. Could we apply the same guard here and add this regression test?
There was a problem hiding this comment.
I believe that that is a bug in another portion of the codebase (as you pointed out): #25279 is out in order to fix it.
There was a problem hiding this comment.
Thanks for pointing me to #25279, I missed that existing fix. That looks like the right place to address this.
|
Is this PR ready to review again? Is it important to you @stuhood |
We are using this optimization downstream, yea. I believe that this is worth doing, because even in the presence of #24456, you would always want to have smaller (per-partition) hash tables and less shuffling (rather than making this decision cost based). I'll rebase it. |
4735398 to
74c930c
Compare
Which issue does this PR close?
Rationale for this change
In
JoinSelection,try_collect_leftconverts a hash join toPartitionMode::CollectLeft(broadcast) when either input is below configured byte/row thresholds. However, if the inputs are already co-partitioned on the join keys across multiple partitions (e.g. both partitioned by hash on the join keys or by range with matching boundaries), switching toCollectLeftintroduces an unnecessaryCoalescePartitionsand broadcasts data across partitions or cluster nodes.Preserving
PartitionMode::Partitionedwhen the inputs already satisfy the partitioned distribution requirements allows the join to execute locally within each partition with zero shuffles or network transfers.What changes are included in this PR?
HashJoinExec::partitioned_input_distribution_requirementsandHashJoinExec::inputs_satisfy_partitioned_requirementsindatafusion/physical-plan/src/joins/hash_join/exec.rs.JoinSelection::try_collect_left(datafusion/physical-optimizer/src/join_selection.rs), skip converting toCollectLeftif inputs already satisfy partitioned distribution requirements across >1 partition.What is the testing strategy for this PR?
New unit tests.
Are there any user-facing changes?
No.