andygrove opened a new pull request, #2124: URL: https://github.com/apache/datafusion-ballista/pull/2124
# Which issue does this PR close? Closes #2121. # Rationale for this change `DynamicJoinSelectionExec::to_actual_join` made two byte-size decisions against inputs that may not be the build side the executor actually runs. Both were raised by @phillipleblanc as non-blocking suggestions on #2084. **Broadcast eligibility was OR-ed across both sides.** `under_threshold` was `supports_collect_by_thresholds(left) || supports_collect_by_thresholds(right)`, so the join was promoted to `CollectLeft` when *either* side passed the byte threshold. Which side actually gets broadcast is decided separately, by `SelectJoinRule::supports_swap_join_order`, which falls back to row count when byte statistics are absent. So if only the right side passed the byte estimate while the wider-by-bytes left side had fewer rows, AQE would still broadcast the left, replicating more bytes to every probe task than the threshold intended. **The hash-join build-fit check measured the pre-swap input.** `max_per_partition_build_bytes(&self.left)` measured `self.left`, but `SelectJoinRule` may swap the inputs, making `self.right` the real build side. The check could therefore validate a partition the build task never consumes, letting an oversized build side through. Both defects share one root cause: the function reasons about `self.left` / `self.right` while the executor runs a possibly-swapped build side. It already computed the swap (for `build_side_join_type`) and threw the result away. # What changes are included in this PR? - Hoist the swap decision into a `swap_inputs` binding and derive `build_side` from it (`self.right` when the swap holds, `self.left` otherwise). `build_side_join_type` reuses it rather than calling `supports_swap_join_order` a second time. - Broadcast eligibility now tests `build_side` alone instead of OR-ing both sides. This is strictly a tightening: it can only reject broadcasts the OR would have allowed, never introduce one. - The build-fit check measures `build_side` instead of `self.left`. - The `AQE join decision` debug line reports which side was selected as build. Byte eligibility deliberately does *not* override the swap (i.e. "build the eligible side even though row count says otherwise"). `SelectJoinRule` re-derives the swap independently via `supports_swap_join_order`, so overriding it here would need the decision threaded through both rules, and the two would silently diverge if either drifts. The issue also asks for the budget to be applied post-coalesce. It is not, and the reason is ordering: `SelectJoinRule` runs inside `replan_stages()`, while `CoalescePartitionsRule` fires later in `actionable_stages()` on the stage the join was just placed into. At decision time there is no `CoalescePlan` to read, so a coalesce-aware check would have to re-derive the bin-pack over a leaf set that differs from the rule's own for chained joins. That is recorded as a known conservative gap in the doc comment on `max_per_partition_build_bytes`; `ballista.planner.coalesce.enabled` is `false` by default, so it is inert unless coalescing is explicitly turned on. Happy to split it out into a follow-up if reviewers would rather see it addressed. # Are there any user-facing changes? No public API or configuration change. Behavior changes in two ways, both narrowing toward what the existing thresholds already intended: fewer joins are promoted to `CollectLeft` (only when the side that actually gets broadcast fits the byte threshold), and the `hash_join_max_build_partition_bytes` fallback to sort-merge now triggers on the swapped-in build side. Worth flagging for review: one pre-existing test, `build_over_threshold_falls_back_to_sort_merge_join`, had to change its expectation. Its fixture paired a 500 MB left `ExchangeExec` with a 16 MB `StatisticsExec` on the right, so under the fix the swap correctly moves the build onto that 16 MB side, which genuinely fits the 200 MB budget — `HashJoinExec` is now the right answer, and the old `SortMergeJoinExec` assertion was pinning the exact bug being fixed. The other two build-fit tests had the same fixture problem and were passing only because the swap landed on a non-`ExchangeExec` whose size is unknown, so nothing was measured at all. All three now use a two-`ExchangeExec` helper, which is the shape the `(Repartitioned, Partitioned)` arm always sees in practice. Three tests added, two of which fail on `main` for exactly the reasons above: broadcast eligibility following the build side, the fit check measuring the post-swap build, and a no-swap mirror case pinning that the selection follows the swap decision rather than blanket-switching to the right. -- 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]
