andygrove commented on PR #2657: URL: https://github.com/apache/datafusion-ballista/pull/2657#issuecomment-6432418474
## Why the slower SF1000 queries regress This follows up on the SF1000 numbers above, using the scheduler event logs from the blocking and pipelined runs. **Root cause: with no AQE, every join is a sort-merge join.** Pipelined jobs are always planned with the static planner (`task_manager.rs`, "A pipelined job plans with the static planner whatever the AQE setting"). With `datafusion.optimizer.prefer_hash_join=false` (the default) and no runtime statistics, every join in all 22 pipelined plans is a `SortMergeJoinExec`, and none is a hash join or a broadcast. In blocking mode, AQE (on by default) replans joins from runtime sizes and produces broadcast (`CollectLeft`) and partitioned hash joins wherever an input is small. `ballista.optimizer.broadcast_join_threshold_rows/bytes` only takes effect through AQE. Q8 is the clearest case: - **Blocking:** the 1.33M filtered `part` rows are broadcast into the `lineitem` scan stage, a `HashJoinExec` with `mode=CollectLeft`. That stage takes 23 s, and `lineitem` is never shuffled. - **Pipelined:** all 6B `lineitem` rows are hash-repartitioned on `l_partkey` and then sorted for a `SortMergeJoinExec` against the 1.33M parts. That one stage takes 346 s and 1624 CPU-s. These large sorts are also what pushed the spills past the 100 GB default limit before #2659. Q9, Q17, Q2 and Q11 have the same shape. Q17 sort-merge-joins about 12B rows, where blocking mode has no sort-merge join at all. **Experiment: pipelined with `datafusion.optimizer.prefer_hash_join=true`** | Query | Blocking (AQE) | Pipelined | Pipelined + `prefer_hash_join` | |---|---:|---:|---:| | 8 | 45.0 s | 361.2 s | 137.9 s | | 17 | 225.9 s | 374.8 s | 155.5 s | | 9 | 199.6 s | 598.5 s | fails, out of memory | - **Q9 fails.** The 1.5B-row `orders` table becomes the build side of a partitioned hash join against `lineitem`. A pipelined job keeps every build resident at once, hash-join builds can't spill, and pipelined tasks aren't retried. The error was `ResourcesExhausted` with several `HashJoinInput` reservations of 3.2 GB each in a 44.7 GB pool. AQE keeps a sort-merge join for this join in blocking mode. - **Q8 is still 3x blocking with hash joins.** The `part`⋈`lineitem` join stage uses only 75 CPU-s over 138 s. It is waiting on the 6B-row `lineitem` repartition over the 2.5 GbE network. Blocking avoids that by broadcasting the filtered parts into the scan. **Implication:** a global hash-join or sort-merge-join switch won't fix this. Pipelined mode needs size-aware join selection at static planning time: estimate build-side sizes from Parquet and row-count statistics, and apply the existing broadcast thresholds, roughly what AQE decides at runtime. Broadcasting small filtered build sides into the large scan looks like the biggest win (Q8, Q9, Q17), with partitioned hash joins for medium inputs and sort-merge joins kept for very large ones. Setup: same cluster as above, at 83edc876f. -- 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]
