stuhood opened a new issue, #25301: URL: https://github.com/apache/datafusion/issues/25301
### Is your feature request related to a problem or challenge? When executing a hash join where both input tables are already physically co-partitioned on the join keys (for example, via matching `Partitioning::Range` split points or compatible `Partitioning::Hash`), the optimal execution mode is `PartitionMode::Partitioned` because it achieves a 0-shuffle, task-local join. However, in [`JoinSelection::try_collect_left`](https://github.com/apache/datafusion/blob/f40d978a4c72bcbbcb6a45fb4c81e95f4e3cfc73/datafusion/physical-optimizer/src/join_selection.rs#L200-L249), `JoinSelection` only consults size thresholds (`supports_collect_by_thresholds` against `hash_join_single_partition_threshold` / `hash_join_single_partition_threshold_rows`). If the build side is smaller than the threshold (default: 131,072 rows / 1 MB), `JoinSelection` unconditionally selects `PartitionMode::CollectLeft` (broadcast): https://github.com/apache/datafusion/blob/f40d978a4c72bcbbcb6a45fb4c81e95f4e3cfc73/datafusion/physical-optimizer/src/join_selection.rs#L301-L312 In distributed engines (such as DataFusion-Distributed or downstream engines like ParadeDB), `CollectLeft` inserts a `CoalescePartitions` / network broadcast across all worker tasks, serializing and transmitting the table across the network. If both sides were already co-partitioned with identical split points / partition boundaries, broadcasting is strictly worse than a local partition-aligned join: 1. A co-partitioned `Partitioned` join requires **0 network shuffles** and **0 broadcast overhead**. 2. Degrading to `CollectLeft` introduces unnecessary network broadcast, data duplication, and memory amplification. ### Describe the solution you'd like Before selecting `PartitionMode::CollectLeft` in `JoinSelection` (or within `try_collect_left`), check whether the join inputs already satisfy the join's distribution requirements for `PartitionMode::Partitioned` on the join keys. If the inputs already satisfy the partitioned join requirement (i.e. they are already co-partitioned): - Preserve `PartitionMode::Partitioned` rather than degrading to `CollectLeft`. - Only fall back to `CollectLeft` when the inputs are not co-partitioned (where broadcasting the smaller side avoids a 2-sided repartition shuffle). ### Describe alternatives you've considered Currently, downstream systems must either: 1. Globally disable broadcast joins by setting `hash_join_single_partition_threshold_rows = 0` (which harms queries joining against small, unpartitioned dimension tables where broadcast is genuinely optimal), or 2. Implement a custom downstream physical optimizer rule that inspects `CollectLeft` joins, verifies whether both children share compatible `Partitioning` on the join keys, and mutates the join mode back to `PartitionMode::Partitioned` before `EnsureRequirements` runs. Handling this directly within `JoinSelection` would benefit all users and downstream engines querying partitioned or co-partitioned tables. ### Additional context - Relevant lines in `JoinSelection`: - [`try_collect_left`](https://github.com/apache/datafusion/blob/f40d978a4c72bcbbcb6a45fb4c81e95f4e3cfc73/datafusion/physical-optimizer/src/join_selection.rs#L200-L249) - [`statistical_join_selection_subrule`](https://github.com/apache/datafusion/blob/f40d978a4c72bcbbcb6a45fb4c81e95f4e3cfc73/datafusion/physical-optimizer/src/join_selection.rs#L301-L312) - Related PRs extending range partitioning and distribution satisfaction: - PR #24600 / PR #24766 (Preserve and adapt range partitioning in `EnforceDistribution`). -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
