andygrove opened a new pull request, #2663: URL: https://github.com/apache/datafusion-ballista/pull/2663
# Which issue does this PR close? Closes #2661. # Rationale for this change AQE decided whether to broadcast a join's build side from DataFusion's row estimate even when that side was a join that had not run yet. Shuffles carry no distinct counts, so DataFusion's inner-join estimate falls back to treating each side's row count as its NDV, and the result is the smaller input's row count. For a key/foreign-key join whose large side is the foreign key, that undercounts by the join's fan-out, with no upper bound. In TPC-H q21 at SF1000 with `nation` joined early, `(supplier ⋈ nation) ⋈ lineitem` was estimated at 399,809 rows (under the 1M row threshold) and produced 151,688,609. AQE broadcast it, `CollectLeft` concatenated its `s_name` column past `Utf8`'s 2 GiB offset limit, and the job failed with `OffsetOverflowError`. Nothing re-checked the size after the broadcast stage measured it. A filter's estimate does not have this problem to the same degree: it is bounded by its input. And when the build side is not a plain scan, the `CollectLeft` path already gives it its own broadcast stage, so measuring it first costs no extra stage. # What changes are included in this PR? - `DynamicJoinSelectionExec::to_actual_join`: when the join would be `CollectLeft` but the build side's size is estimated through an unexecuted join, return `StageBuildSide` instead, so the build side runs as its own stage and the join is decided on its measured size. On the next pass the existing logic either broadcasts the resolved exchange as it stands (it is reused, not re-run) or falls back to a partitioned hash or sort-merge join. With `ballista.optimizer.stage_build_side=false` the join is repartitioned instead. - New `size_estimated_through_join`, which walks the build side for a hash, sort-merge, nested-loop, cross or symmetric hash join, stopping at `ExchangeExec` (a resolved exchange reports measured sizes). Null-aware anti joins are unchanged, since they must run as `CollectLeft`. **Tests** - `stages_a_build_side_estimated_through_a_join_instead_of_broadcasting`: the q21 shape, with a precondition asserting the join's estimate alone would pass the broadcast thresholds. - `repartitions_a_build_side_estimated_through_a_join_without_staging`. - `broadcasts_a_join_output_once_it_is_measured`: the same join behind a resolved exchange that measured small is broadcast as before. - `size_estimated_through_join_looks_below_other_operators`. - `test_hash_join_three_tables_collect_left` (AQE planner, end to end) now shows the inner join staged on a hash exchange, then, once it measures small, broadcast from that same resolved stage. The final plan has the same shape and stage count as before. `test_resolve_nested_dynamic_joins` is updated for the staged build side. - The static `tpch_plan_stability` goldens are unchanged. **Validation**: TPC-H q21 at SF1000 with `nation` second in the `FROM` list, 2 executors, AQE on: | | Before | This PR | |---|---|---| | Result | fails with `OffsetOverflowError` | 100 rows in 429 s | | `supplier ⋈ nation ⋈ lineitem` → `orders` join | broadcast on an estimate of 399,809 rows | staged, measured at 151,688,609 rows (5.8 GB), then sort-merge join | # Are there any user-facing changes? No configuration or API changes. Queries that join the output of another join to a large probe side now measure that output before broadcasting it; when it is small the plan is the same as before. -- 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]
