Repository navigation
perf: faster staged execution (semi-join placement, shuffle spill accounting, scheduler latency) - #2792
Closed
andygrove wants to merge 18 commits into
Closed
perf: faster staged execution (semi-join placement, shuffle spill accounting, scheduler latency)#2792andygrove wants to merge 18 commits into
andygrove wants to merge 18 commits into
Conversation
DataFusion decorrelates IN/EXISTS/NOT EXISTS subqueries into semi and anti joins placed above the FROM clause's join tree, so the subquery filters the joined result. Add a logical rule, like Spark's PushDownLeftSemiAntiJoin, that moves them below passthrough projections, filters, inner joins and the preserved side of outer joins, onto the input that holds their keys. TPC-H q18 then filters orders down to the qualifying orders before joining lineitem. Controlled by ballista.optimizer.semi_join_pushdown (default true).
A task holds a file per shuffle partition it reads or writes plus its spill files, so stages with a few hundred partitions need thousands of descriptors. With the common 1024 soft limit, TPC-H at 128+ partitions fails with "Too many open files". The JVM raises its soft limit at startup, so Spark executors are unaffected; do the same.
For an inner join against a small dimension filtered by =, IN or LIKE on literals (a nation, region or part type), also semi-join the dimension with the table the join key comes from, below the joins in between, so the filter removes rows before they are joined and shuffled. This is the effect of Spark's runtime filters. In TPC-H q7 supplier and customer are reduced to the two filtered nations before joining lineitem. Logical plans carry no statistics, so the scheduler runs the rule before physical planning: candidates are found by shape, planned on their own to read their row estimate, and used only under ballista.optimizer.broadcast_join_threshold_rows. Controlled by ballista.optimizer.semi_join_reduction (default true).
Under pull scheduling the scheduler binds tasks, marking them running, before tonic encodes the PollWorkResult. When the response exceeds the gRPC encoding limit (many tasks with large plans, such as broadcast readers listing every upstream location at 128 partitions) the executor gets an error, the tasks are never run, and the job hangs. Send only the tasks that fit and report the rest as killed, which re-queues their partitions without counting a failure. A single task larger than the limit fails its job with an error naming the setting to raise.
Aggregates emit their output as one large array sliced into batch-size pieces. The sort shuffle writer summed get_array_memory_size per buffered batch, which counts the whole shared buffer for every slice, so after the first slice the 256 MiB per-task budget looked exceeded and the writer spilled on almost every batch: TPC-H q18's partial aggregate stage spilled 1,145 times per input partition at SF100. Use DataFusion's RecordBatchMemoryCounter, which counts each buffer once across batches, in the sort shuffle buffer and in BufferExec's dam.
Pushing a semi join below a Filter is sound but makes it probe the unfiltered input: TPC-H q4's EXISTS probed all orders instead of the 5.7M in its date range (3.3 s to 9.6 s at SF100). Stop at filters. A dimension semi join that lands on a tiny table, such as nation reduced by region in q2/q5/q8, only adds a join. Estimate the input it lands on and remove it below 100K rows.
A join whose other input is filtered can remove most rows itself, so a semi or anti join pushed below it probes rows that join would have dropped. TPC-H q21's EXISTS and NOT EXISTS moved below the orders (o_orderstatus = 'F') and nation joins onto all 379M l1 rows at SF100, doubling its run time (32 s to 65 s). Keep subquery semi joins above such joins; dimension reductions still sink through them, since there the semi join is the selective side.
DataFusion caps spill files at 100 GB per runtime. At TPC-H SF1000 on two executors, q5 and q18 spill more than that and fail, although the executor's work dir is a 1 TB volume. When --max-temp-directory-size is not set, use the space free on the work dir's filesystem at startup, never less than DataFusion's default.
The scheduler serves gRPC through axum, which, unlike tonic's own server, leaves Nagle's algorithm on for accepted connections. A poll response is several small HTTP/2 frames, so the later ones wait for the executor's delayed ACK: about 40 ms per poll on Linux. Under pull scheduling that is added to every task launch, so a stage of N waves pays N x 40 ms; TPC-H q2 at 128 partitions spent about 2 s of its 5.6 s waiting. Set TCP_NODELAY on every accepted connection.
github-actions
Bot
added
the
documentation
This was referenced Nov 7, 2026
added 6 commits
November 7, 2026 10:34
The build side of a broadcast hash join is a broadcast ShuffleReaderExec that reads every partition of the upstream stage. Each task runs its own plan, so every task on an executor fetched and decoded the same build side: at 128 partitions a 16-task broadcast stage made 16,384 fetch requests for one 318K-row table. Add an executor-wide BroadcastReadCache. The first task to run a broadcast read fetches it; tasks that ask meanwhile wait for that fetch, and later ones reuse the decoded batches. Entries are keyed by job, stage and the exact locations read, expire after 30 s unused, are not kept above 1 GiB, and are dropped with the job's data. A failed fetch is not cached, so each task reports its own fetch error as before. Controlled by ballista.shuffle.cache_broadcast_reads (default true).
An undecided AQE join (DynamicJoinSelectionExec) reported the partitioning of the join it was planned as. A CollectLeft join passes its right input's partitioning through, so a join planned with a one-partition table such as nation on the right reported one partition, although the join it resolves into may build nation and output the other side's partitions. Meanwhile DataFusion's CombinePartialFinal- Aggregate saw an aggregate over one partition and merged it into a Single aggregate, which later replans cannot split. Once the join resolved, the aggregate's input was coalesced and exchanged below it, so every task of the consuming stage read and aggregated the whole input: TPC-H q11's scalar sum read all 32M rows in each of 16 tasks at SF1000 with 256 partitions. Report at least as many partitions as the widest input until the join is decided.
with_new_children is deprecated in DataFusion 55, and clippy rejects the call under -D warnings.
The existing multi-input test never spilled, so nothing checked that each output partition's range stitches every input's spill and in-memory bytes together in order. Run it again with a tiny buffer budget so every input spills several times and keeps an in-memory remainder.
The spill manager kept one IPC writer open per spilled output partition until the task finished, each holding a file descriptor and its own compression context outside the memory pool. With hundreds of output partitions that is hundreds of open files per input partition and, with zstd, hundreds of megabytes of compressor state. Write each spill event as one complete stream appended to the partition's file and close it; readers already cross concatenated streams.
zstd writes shuffle data at roughly half the size of lz4 (about 8 bytes per row against 16 for TPC-H lineitem's shuffled columns) for little extra CPU. Shuffles that cross the network get correspondingly faster, and smaller files also cut local disk and page-cache traffic. lz4 stays available through ballista.shuffle.compression.codec.
This was referenced Nov 7, 2026
# Conflicts: # ballista/core/src/config.rs # ballista/core/src/execution_plans/sort_shuffle/spill.rs # docs/source/user-guide/configs.md
# Conflicts: # ballista/core/src/config.rs # ballista/core/src/execution_plans/broadcast_cache.rs # ballista/core/src/execution_plans/shuffle_reader.rs # ballista/core/src/optimizer.rs # ballista/core/src/optimizer/semi_join_pushdown.rs # ballista/core/src/optimizer/semi_join_reduction.rs # ballista/executor/src/config.rs # ballista/executor/src/executor_process.rs # docs/source/user-guide/configs.md
Member
Author
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
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),mainand 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),mainis 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:IN/EXISTSsemi joins above the wholeFROMjoin tree, and applies selective dimension filters (such asnation) 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 withPushDownLeftSemiAntiJoinand runtime filters.What changes are included in this PR?
This PR collects the benchmark results. Every change on this branch has now merged into
mainthrough its own PR: #2793, #2794, #2795 (poll response fit), #2796 (shared buffers in shuffle buffering), #2797, #2798 (Nagle), #2799, #2805 (zstd as the default shuffle codec at level 1, with #2802's spill writers closed after each spill), #2806 (undecided join partitioning) and #2807 (broadcast read cache). The branch is merged withmainas of 3f300da, and its only remaining diff againstmainis theBufferExechalf of (5): #2796 changed the sort shuffle buffer only, soBufferExec's dam mode still charges every zero-copy slice with the whole shared buffer. That will go in a follow-up PR. This PR stays a draft to collect benchmark results.perf: push semi and anti joins below inner joins: logical ruleSemiJoinPushdown(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.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/LIKEon 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 underbroadcast_join_threshold_rows, dropping any that land on tables under 100K rows.fix: raise the executor's open file limit to the hard limit at startup.fix: reschedule tasks that do not fit in a poll response: tasks that would push aPollWorkResultpast 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. Merged in fix: reschedule tasks that do not fit in a poll response #2795.fix: count buffers shared by sliced batches once in shuffle buffering: uses DataFusion'sRecordBatchMemoryCounterin the sort shuffle buffer andBufferExec. Merged in fix: count buffers shared by sliced batches once in shuffle buffering #2796 for the sort shuffle buffer; theBufferExecpart is still on this branch.fix: let executor spills use the work dir's free space by default, when--max-temp-directory-sizeis unset.perf: disable Nagle's algorithm on scheduler connections. Merged in perf: disable Nagle's algorithm on scheduler connections #2798.Results on one machine, TPC-H SF100, 2 executors x 8 cores, 16 partitions, one iteration, both engines on the same query text:
mainThe 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 onmain, mostly from the Nagle fix. All 22 queries pass--verifyagainst DataFusion at SF10, at 16 and at 128 partitions.Two-node cluster, TPC-H SF1000 on node-local disk, 2 executors x 16 cores, 32 partitions, one iteration,
benchmarks/cluster/run.pyfor every engine:mainNotable queries (main / branch / Comet, seconds): q7 342 / 68 / 132, q21 713 / 279 / 243, q11 48 / 4.7 / 44, q19 36 / 18 / 18. q3 is slower in this run (128 / 165 / 83) with the same plan, spill counts and CPU time, so the difference is in how fast the 120 GB of spill files were written and read. At 32 partitions at this scale, Ballista still spills heavily in sorts and shuffles (q3, q5, q10). With 256 partitions the branch completes all 22 queries in 1744 s, 1.31x faster than Comet: spill-heavy queries improve (q3 165 to 103 s, q5 439 to 266 s, q10 134 to 109 s, q18 231 to 185 s, q21 279 to 223 s), while short and broadcast-heavy ones get slower (q11 4.7 to 14.7 s, q8 33 to 48 s).
Two later commits address that.
e3be5d583shares each broadcast read between the tasks on an executor (q2's broadcast stage at 128 partitions: 3.0 s to 0.7 s of task time, 16,384 to 2,048 fetches), which is worth about 1% at SF1000.c0913e62f(merged in #2806) fixes undecided AQE joins reporting one output partition, which let DataFusion merge q11's scalarsuminto aSingleaggregate that every task then ran over all 32M input rows (also onmain); q11 goes from 14.7 s to 5.4 s. With both, the branch completes SF1000 in 1702 s, 1.35x faster than Comet.The remaining large stages at SF1000 were bound by shuffle bytes: the lineitem join in q5 pulls half of lineitem's shuffle across the network, and the map side spends much of its time writing and copying spill files. lz4 IPC stores lineitem's shuffled columns at about 16 bytes per row (Comet's shuffle: 14.8), zstd at about 8.2. Making zstd the default (
b83da19ea, merged in #2805) brings SF1000 to 1318 s, 1.74x faster than Comet: q5 274 to 170 s, q3 113 to 76 s, q9 156 to 117 s, q18 183 to 145 s, q21 199 to 161 s. It also helps on one machine with no network in between (SF100, 138.8 s to 132.3 s). Before that,c072080dastops the sort shuffle spill manager from keeping a writer open per spilled output partition until the task ends; with zstd each open writer holds its own compression context outside the memory pool, which at 256 partitions came to thousands of open files and several GB per executor.TPC-DS SF100 on the single machine (2 x 8 cores, 16 partitions, 98 queries, q72 skipped):
main510.4 s, this branch 362.6 s (1.41x). Every query returns the same row count on both. The largest gains are q50 10.9 s to 1.1 s, q25 20.9 s to 2.7 s, q29 13.3 s to 1.8 s, q17 12.6 s to 2.3 s and q4 41.8 s to 20.2 s. The only queries more than 15% slower are q34 (+0.18 s) and q60 (+0.37 s).Still open: coarse tasks leave vcores idle behind their slowest partition and during the task's single-threaded shuffle file write (parallel writes to the one file do not help, since buffered writes to a file serialize in the kernel); q2, q4, q10, q13, q16, q19 and q22 remain 3-60% slower than Comet at SF1000. Storing string offsets as lengths before zstd, as Comet does, would shrink string-heavy shuffles by a further 20-55% (#2804).
Are there any user-facing changes?
Yes:
ballista.optimizer.semi_join_pushdownandballista.optimizer.semi_join_reduction. Plans for queries with subqueries or filtered dimension joins change.--max-temp-directory-sizeis not set, the executor's spill limit is the work dir's free space at startup instead of 100 GB.