Reduce overhead of `Partitioned` hash join
- Dominant language
- Rust
- Stars
- 9.3k
- Forks
- 2.4k
- Avg merge
- 3d 7h
- Merged PRs (30d)
- 344
Description
# Is your feature request related to a problem or challenge?
Currently, a high part of the cost of the `Partitioned` join is repartitioning the entire build and probe side of the join - which currently copies all of the columns twice(!) during `take` and `coalesce`, this makes this hash join slower if this side is large/wide or both. We could avoid one copy in `RepartitionExec` (using arrow-rs coalesce API when fully implemented), but not two.
`CollectLeft` avoids this cost at the right side, but the building phase is single threaded, which greatly limits the parallelism in the query - the query/partitions needs to wait until the entire build side is finished, or some buffering/eager evaluation as implemented here https://github.com/apache/datafusion/pull/19761 might yield some
# Describe the solution you'd like
We should be able to do the `hash % partition` of the probe side **during** the join, avoiding the need for a `RepartitionExec`. We only pass the indices of the matching partitions to the correct partition in the left - which should greatly reduce the overhead of the repartitioning.
When this is implemented, we might want to look at `hash_join_single_partition_threshold` and `hash_join_single_partition_threshold_rows` again which could be reduced to make most joins run fully in parallel./
_Originally posted by @Dandandan in https://github.com/apache/datafusion/issues/19761#issuecomment-3743995482_
Contributor guide
Assessment
This issue has not been assessed yet.