andygrove opened a new pull request, #2792: URL: https://github.com/apache/datafusion-ballista/pull/2792
# Which issue does this PR close? No single issue. This is a working branch for making staged Ballista faster than Spark + Comet on TPC-H, to be split into separate PRs once it settles. Related: #2662 (q21 at SF1000), #2520 (performance review tracking), #1375 (dynamic filters). # Rationale for this change On a single machine (TPC-H SF100, 2 executors x 8 cores, both engines running the queries in `benchmarks/queries`), `main` and Spark 4.1 + Comet 1.2.0 take about the same time (197 s and 204 s). On a two-node cluster at SF1000 (2 x 16 cores, 32 partitions), `main` is 1.14x slower than Comet over the queries both finish, and q5 and q18 fail. Comparing plans, stage timelines and metrics against Spark's event logs found these causes: - **Join placement.** DataFusion leaves `IN`/`EXISTS` semi joins above the whole `FROM` join tree, and applies selective dimension filters (such as `nation`) only at the join that names them. q18 joins 600M rows before its subquery keeps 6,400 orders; q7 joins and shuffles 182M lineitem rows before the nation filter drops 92% of them. Spark handles these with `PushDownLeftSemiAntiJoin` and runtime filters. - **Spilling.** The sort shuffle writer charged every zero-copy slice of an aggregate's output with the whole shared buffer, so q18's partial aggregate stage spilled 1,145 times per input partition. At SF1000 the spills of q5 and q18 exceed DataFusion's 100 GB default spill cap and fail. - **Per-task latency.** The scheduler serves gRPC through axum, which leaves Nagle's algorithm on, so each pull-mode poll waited about 40 ms for a delayed ACK. That is added to every task wave and dominates at high partition counts. - **High partition counts break.** At 128 partitions the executor hits a 1024 open-file soft limit, and a poll response over the gRPC limit leaves its bound tasks running forever, which hangs the job. # What changes are included in this PR? Each commit is meant to become its own PR: 1. `perf: push semi and anti joins below inner joins`: logical rule `SemiJoinPushdown` (`ballista.optimizer.semi_join_pushdown`). It moves semi/anti joins below passthrough projections and inner joins (and the preserved side of outer joins), but not below filters or joins whose other input is filtered, where the other operator is the cheaper filter. 2. `perf: semi-join small filtered dimensions below the joins in between`: `SemiJoinReduction` (`ballista.optimizer.semi_join_reduction`). For a join against a dimension filtered by `=`/`IN`/`LIKE` on literals, it adds a semi join against a copy of the dimension on the table the key comes from, below the joins in between. Logical plans have no statistics, so the scheduler runs it before physical planning and keeps only dimensions whose estimated rows are under `broadcast_join_threshold_rows`, dropping any that land on tables under 100K rows. 3. `fix: raise the executor's open file limit to the hard limit at startup`. 4. `fix: reschedule tasks that do not fit in a poll response`: tasks that would push a `PollWorkResult` past the encoding limit are reported back as killed (re-queued, no failure counted). A single oversized task fails its job with an error naming the setting. 5. `fix: count buffers shared by sliced batches once in shuffle buffering`: uses DataFusion's `RecordBatchMemoryCounter` in the sort shuffle buffer and `BufferExec`. 6. Two follow-ups to (1) and (2) that keep semi joins above filters and filtered joins (TPC-H q4 and q21 regressed without them) and skip reductions onto tiny tables. 7. `fix: let executor spills use the work dir's free space by default`, when `--max-temp-directory-size` is unset. 8. `perf: disable Nagle's algorithm on scheduler connections`. Results on one machine, TPC-H SF100, 2 executors x 8 cores, 16 partitions, one iteration, both engines on the same query text: | | Total (s) | |---|---:| | Spark 4.1 + Comet 1.2.0 | 203.5 | | Ballista `main` | 197.5 | | This branch | 140.9 | The biggest changes are q7 20.1 s to 5.6 s, q18 42.2 s to 16.0 s and q21 31.9 s to 15.6 s. No query is more than noise slower than on `main`. At 128 partitions the branch takes 148 s against 204 s on `main`, mostly from the Nagle fix. All 22 queries pass `--verify` against DataFusion at SF10, at 16 and at 128 partitions. SF1000 cluster results and a TPC-DS comparison are in progress and will be added here. Still open: the broadcast build side is fetched and rebuilt by every task, coarse tasks leave vcores idle behind their slowest partition, and q13/q20 remain slower than Comet. # Are there any user-facing changes? Yes: - Two new config keys, both enabled by default: `ballista.optimizer.semi_join_pushdown` and `ballista.optimizer.semi_join_reduction`. Plans for queries with subqueries or filtered dimension joins change. - When `--max-temp-directory-size` is not set, the executor's spill limit is the work dir's free space at startup instead of 100 GB. - The executor raises its open-file soft limit at startup. -- 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]
