feat: spilling shuffled hash join (research draft) by andygrove · Pull Request #7057 · apache/datafusion-comet

andygrove · 2026-10-16T18:08:58Z

Which issue does this PR close?

Part of #2545. This is a research draft for discussion, not a change to merge as it stands.

Rationale for this change

Comet's native shuffled hash join is DataFusion's HashJoinExec, whose build side has to fit in the task's share
of the memory pool. When it does not, the task fails. That is why spark.comet.exec.forceShuffledHashJoin is off by
default and has a build-size guard (#6384), even though a hash join is faster than a sort-merge join when it fits.

DataFusion has an open design for spilling hash joins (apache/datafusion#24768), but no agreed approach yet. The
sort-merge fallback (apache/datafusion#25217) is waiting on review, and a refactor of the hash join stream is in
flight. Comet's case is much simpler than DataFusion's general one. Every native shuffled hash join runs in
Partitioned mode, on one partition per task, and the output order contract lives on the Spark side, not in
DataFusion's optimizer. This draft explores a Comet-only design for that case, to find out what it costs and what
breaks. Parts of it could be offered upstream later.

What changes are included in this PR?

The design is written up in docs/source/contributor-guide/spilling-hash-join-design.md. In short:

SpillingHashJoinExec (native/operators/src/spilling_hash_join/) wraps the HashJoinExec that Comet already
plans. An equi-join splits exactly by a hash of its keys, for every join type, so each bucket is joined by a stock
HashJoinExec, and the operator implements no join semantics of its own.
- While the build side fits, it runs that join unchanged.
- When the memory pool refuses, it routes both sides to 16 buckets, destages the largest buckets to disk, joins
  the buckets left in memory with the probe input, and then joins each spilled bucket from its files. A bucket
  that does not fit is split on the next bits of the hash. A bucket that splitting cannot shrink, because most of
  its rows share one key or have a null key, is joined in chunks, with a replay of its probe rows for each chunk.
- Spill files hold full batches. Routing splits every input batch 16 ways, so a file's writer holds its parts,
  reserved, until they make a batch of the session's batch size or 256 KiB.
- Each spilled bucket has a Bloom filter of its build keys. A probe row that the filter rejects matches nothing
  anywhere, so it goes to the join of the buckets left in memory instead of to disk.
- Everything the operator and its inner joins hold is reserved under one memory consumer. That includes what
  HashJoinExec allocates without reserving it: its concatenated copy of the build batches, and its chain of row
  indices. A spilled bucket's build side is read back into one batch, which its inner join keeps without a
  copy, so that the bucket needs about its own size in memory. A build side with nested, dictionary or view
  columns is read back compacted instead.
Two configs. spark.comet.exec.hashJoin.spill.enabled is on by default, so that CI runs the new join.
spark.comet.exec.hashJoin.spill.maxBuildBufferBytes forces spilling, for tests. The serde enables spilling only
for shuffled hash joins that are not null-aware, do not use the join runtime filter, and whose output order no
operator above relies on. CometExecRule walks up the plan from each shuffled hash join to find that out, and a
spilling join reports no output order.
The Spark plan keeps CometHashJoinExec, and the native operator displays as the HashJoinExec it runs, so plans
and golden files do not change. New metrics: spilled_buckets, split_buckets, chunked_buckets,
bloom_filtered_rows and budget_overflow_bytes, plus the spill metrics.
CometHashJoinMemoryBenchmark compares Spark's sort-merge join with Comet's sort-merge join, its hash join with
spilling disabled, and its hash join with spilling, across native memory budgets. A native benchmark,
cargo bench --bench spilling_hash_join, runs one task's join over in-memory batches.
User docs for the join and its metrics.

It reviews best commit by commit. The first four commits are the first version, and the rest are a second round
that made spilling cheaper and let more joins spill.

Findings so far

CometHashJoinMemoryBenchmark joins 4M build rows with 20M probe rows on local[4], with a native pool per task
from 1024 MiB down to 64 MiB. These runs were on an otherwise idle AMD 7950X3D, pinned to its eight V-cache cores,
median of three. Before is the first version, after is the second round.

Inner join, median ms / CPU s	Comet sort-merge join	Hash join, no spilling	Before	After
1024 MiB	1051 / 4.4	641 / 2.7	626 / 2.5	637 / 2.6
192 MiB	1152 / 5.1	617 / 2.4	730 / 3.2	707 / 3.2
128 MiB	1161 / 5.1	fails	864 / 3.8	825 / 3.4
64 MiB	1152 / 5.2	fails	889 / 3.9	787 / 3.2

The spilling join completes for every join shape at every budget, including where the join without spilling
fails. The shapes are inner, left, right and full outer, left anti, and an inner join with a hot key.
When it spills, it is faster than Comet's sort-merge join in both wall time and CPU time, at every budget and
for every join shape. At 64 MiB the second round is 5 to 19% faster than the first and spills 38 to 50% fewer
bytes.
The first draft's numbers, which showed it no faster, came from a machine that another benchmark was also using.
While the build side fits, it costs the same as the join without spilling.
Reserving honestly moves the spill point. At 192 MiB the join without spilling fits, but the spilling join
spills, mostly because it counts the concatenated copy of the build side that HashJoinExec makes without
reserving it.
In the native benchmark, a join under a 20 MiB pool takes 24% less time than in the first version, mostly from
writing full batches: spill files held slivers of 512 rows, or 32 once split, and every one cost an IPC message
and an lz4 frame per buffer. Under 10 MiB the first version split every bucket again, and the join now takes 84%
less time, because a spilled bucket's inner join gets one build batch and needs half the memory, so no bucket is
split.
The Bloom filters keep 1.4 to 2.3 million of the native benchmark's 5 million probe rows off disk, where half of
the probe rows match. A first version with Parquet's split block filter cost about 9 ns a lookup and lost
overall. The one here sets four bits in one 64-bit word.
On spill compression: in CometHashJoinMemoryBenchmark's inner join at 64 MiB, zstd makes the spilling join 14%
slower than lz4 and spills 39% fewer bytes, and uncompressed files make it 4% faster and spill twice as many. The
join uses spark.comet.exec.spill.compression.codec, the setting it shares with sorts and aggregates, and which
is best depends on the disks.
Two DataFusion facts came up along the way. A batch read back from a spill file can hold up to twice the memory
it held when written: DataFusion 55.2 reads spill files in chunks of 128 KiB, and arrow assembles a message that
spans chunks in a buffer that grows by doubling. fix: frame IPC messages when reading spill files so batches pin only their own bytes datafusion#24592 fixed that on main after 55.2 (it closed
Memory should not blow up after Arrow IPC write-read round trip during spilling datafusion#17340) and was not backported, so this PR reads a spilled build side back compacted. The
backport to branch-55 is proposed in [branch-55] fix: frame IPC messages when reading spill files so batches pin only their own bytes (#24592) datafusion#26435. And
HashJoinExec counts its final pass over the build rows as probe input in input_rows.

TPC-H at scale factor 100

On the same machine, as a Spark 4.1.1 standalone cluster of two executors with 8 cores each, AQE on, one run of
each query. The hash join arms force it with spark.comet.exec.forceShuffledHashJoin and maxBuildSize=-1. Total
time of the 22 queries:

Off-heap pool per executor, shuffle partitions	Comet sort-merge join	Spilling hash join	Hash join, no spilling
16 GiB, 200	255.5 s	232.8 s	232.5 s
2 GiB, 32	268.0 s, 12 tasks retried	242.2 s, 1 task retried	q21 fails
1 GiB, 16	272.1 s, 11 tasks retried	q21 fails	q9, q21 and q22 fail

Every query that completed returned the same rows as with the sort-merge join, by row count and a hash of its
sorted rows. That includes every query that spilled.
With 16 GiB nothing spills, and the spilling join costs nothing over the join without spilling.
With 2 GiB, q21 spills 384 buckets, 3.84 GB, and takes 33.5 s, against 36.7 s with the sort-merge join. Without
spilling it fails. Its NOT EXISTS anti join probes the output of the sort-merge join of EXISTS, so Spark
reports an order for it, and until CometExecRule looked at the operators above, it could not spill.
With 1 GiB, q9, q10, q17 and q22 spill. Every query but q21 completes, the 21 of them in 208.0 s, against
235.6 s with the sort-merge join.
With 1 GiB, q21 fails with spilling too. The anti join takes 124 MB of the task's 130 MB share for its build
side, and then a sort below its probe side, for the sort-merge join of EXISTS, cannot get the 10 MiB it needs
to merge. The join takes whatever the pool grants, and nothing makes it give memory back.
Capping the join's build buffer at 64 MiB with maxBuildBufferBytes lets q21 complete in 33.9 s. Making
Comet's fair pool divide the task's share instead of the executor's pool, with
spark.comet.exec.memoryPool.fraction=0.125, lets it complete in 34.9 s, but then consumers that cannot spill
hit their smaller limits: q16's broadcast hash join fails in both arms, and so does a sort's merge in q2 with
the sort-merge join.
A prototype pool that limits only the consumers that can spill, as DataFusion's FairSpillPool does, over
the task's share, fixes both. Every TPC-H query completes in both arms at 1, 2 and 16 GiB with no task
retried, in up to 3% more total time, and the spilling join completes all 22 at 1 GiB in 248.1 s. It is not
part of this PR, because it changes when every Comet operator spills.
The sort-merge join arm's retried tasks were the same kind of failure: a sort refused memory for its merge while
another sort of its task held most of the task's share. Those retries succeed, because a sort's need is
bounded and a retry finds a larger share. A retried spilling join takes the larger share as well.

TPC-DS at scale factor 100

On the same cluster, with 2 GiB and 32 shuffle partitions:

TPC-DS, 2 GiB, 32 partitions	Comet sort-merge join	Spilling hash join	Hash join, no spilling
All 99 queries	974.8 s, 15 tasks retried	2174.7 s, 3 tasks retried	q72 runs past 40 min
The 98 queries other than q72	596.8 s	564.3 s	564.5 s

Only q72 spills, and every query returned the same rows in every arm, except q39, whose floating point results
differ in their last digits from run to run: every row agrees within a relative 4e-16, and two runs of the
sort-merge join differ from each other the same way.
q72 is slow with either hash join, not because of spilling. Its join of catalog_sales with inventory, on the
item alone with a residual filter, produces 16.8 billion rows, and the tasks of that stage take 179 s on average
with the sort-merge join, 678 s with the spilling join and over 1,000 s without spilling. Forcing the hash join by
default would need a guard on fan-out, not only on build size.

Open questions

Is the honest reservation of the concatenated copy worth the earlier spill point, or should the build side be
concatenated as it arrives, so the copy is smaller? A spilled bucket now avoids the copy by handing its inner
join one build batch, but the join that fits in memory still makes it.
Spilling is on by default so that CI runs it. Now that it beats the sort-merge join when it spills, can
forceShuffledHashJoin be on by default, after measuring TPC-H and TPC-DS at realistic memory sizes?
How does this fit with fix: choose the forced shuffled hash join at planning time so Spark keeps its sorts #6785, which plans the forced shuffled hash join at planning time, and with
forceShuffledHashJoin.maxBuildSize once joins can spill?
How should a spilling join share the task's memory with the operators that run after its build, such as a sort
below its probe side? Comet's fair_unified pool divides the executor's whole pool among a task's consumers,
so its limit does not bind while Spark gives each task an even share. Limiting only the consumers that can
spill, to a share of what the task's share leaves after the ones that cannot, as DataFusion's FairSpillPool
does, helps every operator, the sort-merge join included, as the prototype above shows, but changes when all
of them spill. Or the join could keep resident only part of what the pool grants when its probe side has
operators that need memory.
A join whose output order an operator above relies on cannot spill. With forceShuffledHashJoin, there are none
among the 611 native shuffled hash joins in the TPC-DS plans at scale factor 1. Before CometExecRule looked at
the operators above, there were 10, in six queries, for which Spark reported an order that nothing used. Should a join whose order is used keep
the sort-merge join?
Is 16 buckets per level the right fan-out?

The remaining work, in priority order, and how to pick it up are in the design doc's Remaining work and
Working on it sections.

How are these changes tested?

Rust tests compare the operator with a plain HashJoinExec, row for row, for all 10 DataFusion join types, with
and without a residual filter. They cover each path: in memory, bucketed, split again, chunked for a hot key and
for null keys, real pool refusals, every bucket spilled, and empty sides, each with uncompressed and lz4 spill
files. They also check that no inner join reserves beyond its estimate, that the operator's row counts are right,
that a failed acquire fails the join instead of spilling, and that dropping the join deletes its spill files.
Four deliberate breakages of the operator were each caught.
Further differential tests join keys of decimal, date, timestamp, string view, binary and dictionary type, carry
list, struct, string view, dictionary, wide decimal, fixed-size binary and boolean payloads, and write rows wide
enough that spilled batches span DataFusion's read chunks. The payload test is what showed that a dictionary's
values have to be copied when a build side is read back. A unit test checks the one-batch build against
concat_batches.
A planner test checks that the join is wrapped, that it keeps the projection swap_inputs adds, and that a join
with the runtime filter is left alone.
CometSpillingHashJoinSuite checks that under a native pool too small for the build side the join fails without
spilling, and completes and matches Spark with it. It also checks that every Spark join type matches Spark while
spilling, and which joins are eligible.
CometSpillingHashJoinGreedyPoolSuite checks that under the greedy_unified pool, with a small off-heap pool,
the join spills when Spark refuses it memory, and matches Spark.
CometJoinSpillingSuite runs all of CometJoinSuite with every native shuffled hash join spilling all of its
build side.
These suites are now listed in the PR workflows. CometSpillingHashJoinSuite was not, which failed preflight, so
no PR build ran for the first version.
Locally, CometJoinSuite, CometJoinSpillingSuite and both spilling suites pass on Spark 4.1, and every Spark
profile, 3.4, 3.5 with strict warnings, 4.0, 4.1 and 4.2, compiles with its tests.
TPC-H and TPC-DS at scale factor 1 match the golden answers of CometTPCHQuerySuite and CometTPCDSQuerySuite,
all 22 and all 134 queries, with the suites' shuffled hash join configuration also setting
spark.comet.exec.hashJoin.spill.maxBuildBufferBytes, to 1 MiB and 256 KiB, so that every larger build side
spills.
TPC-H at scale factor 100, under three sizes of the off-heap pool, and TPC-DS at scale factor 100, under one,
return the same rows with the spilling join as with the sort-merge join for every query that completed, but for
floating point noise in TPC-DS q39, see the findings above.
The Spark SQL suites pass on Spark 4.1, in CI with the run-spark-4.1-tests label. They run Comet on-heap,
where the native pool is unbounded and nothing spills, so they show that the wrapper is transparent. Not yet
run: the Spark SQL suites on the other Spark versions, and the Iceberg suites.

SpillingHashJoinExec wraps the HashJoinExec that Comet plans for a shuffled hash join. While the build side fits in memory it runs that join unchanged. When the memory pool refuses, it routes both sides to 16 buckets by a hash of the join keys, destages the largest buckets to disk, joins the resident buckets with the probe input, and then joins each spilled bucket from its files. A bucket that does not fit is split on the next bits of the hash, and one that splitting cannot shrink is joined in chunks. Each bucket is joined by a stock HashJoinExec, so the operator implements no join semantics of its own. Everything the operator and its inner joins hold is reserved under one memory consumer, including what HashJoinExec allocates without reserving.

spark.comet.exec.hashJoin.spill.enabled, on by default, runs a native shuffled hash join with SpillingHashJoinExec. The serde enables it only for shuffled hash joins that are not null-aware, do not use the join runtime filter, and whose output order Spark does not rely on, because rows of spilled buckets are emitted after the others. spark.comet.exec.hashJoin.spill.maxBuildBufferBytes makes a join spill once its build side holds that many bytes, for tests. CometHashJoinExec declares the join's spill metrics, so its spills count in the task's spill totals.

…y budgets

It runs one task's join of CometHashJoinMemoryBenchmark over in-memory batches, so that the cost of spilling is not diluted by scans and shuffles: a build side of unique keys with a string payload, and a probe side five times as long, half of whose rows find a match. The cases are the plain join and the spilling join with no memory limit, the spilling join under pools of 64, 40, 20 and 10 MiB with lz4 spill files, and at 20 MiB with zstd and with uncompressed files. Each case prints what it spilled before it is timed.

Routing splits every input batch into one part per bucket, a sixteenth of a batch, and a split bucket's parts are split again. Each part was written to its file as a batch of its own, so spill files held slivers of about 512 rows, or 32 once split. Every one cost an IPC message and an lz4 frame per buffer, and on reading back a decode and a probe batch of the inner join. A spill file's writer now holds its parts, reserved, until they make a batch of the session's batch size or 256 KiB, and writes them as one. When the pool refuses the memory, it writes what it holds at once. A batch read back from a spill file can hold up to twice the memory it held when written: DataFusion reads spill files in chunks of 128 KiB, and arrow assembles a message that spans chunks in a buffer that grows by doubling. With full batches nearly every message spans chunks. The build side of a spilled bucket is now read back with every batch copied into buffers of its own size, dictionary values and view data included, and is reserved at what it held when written whatever the codec. A compressed file's build side was reserved at its in-memory size plus the file's size before. In the native benchmark a join under a 20 MiB pool takes 19% less time, and one under 10 MiB, which splits every bucket again, 67% less. split_again's limit is lowered, because one of its buckets now fits the old one.

The differential tests joined only an integer or string key with integer and string payloads. key_types joins keys of decimal, date, timestamp, string view, binary and dictionary type through every path, with a hot key so that a bucket is joined in chunks. payload_types carries list, struct, large string, string view, dictionary, wide decimal, fixed-size binary and boolean payloads through routing, the writers' concatenated parts and the copy of a build side read back. It is what showed that a dictionary's values have to be copied too. wide_rows writes batches larger than the chunks DataFusion reads spill files in.

A probe row of a spilled bucket was written to the bucket's probe file even when no build row of the bucket had its key. Such a row matches nothing anywhere, so the join of the resident buckets, where no build row has its key either, emits it exactly as the bucket's own join would, for every join type. Each spilled bucket now gets a Bloom filter of its build keys when it is destaged, and a probe row the filter rejects goes to the resident join instead of to disk. The filter is checked for every probe row of a spilled bucket, so a lookup has to cost far less than writing and reading the row back. A key sets four bits in one 64-bit word, from bits of the routing hash that routing at the first level does not use, and a lookup is one load and one comparison. A first version with Parquet's split block filter cost about 9 ns a lookup and was slower overall. The filter grows as a scalable Bloom filter does, since a bucket's final size is not known when it is destaged, its memory is reserved, and a filter that rejects fewer than one in eight of its first 64K lookups is dropped. The new metric bloom_filtered_rows counts the rows it kept off disk. In the native benchmark, where half of the probe rows match, the filter keeps 1.4 to 2.3 million of 5 million probe rows off disk. Pinned to one CCD, a join under a 40 MiB pool takes 7% less time, one under 10 MiB 20% less, and one with zstd spill files 12% less. With uncompressed files, where spilling a narrow row costs little, it takes 2% more.

CometSpillingHashJoinSuite was not listed in the PR workflows, so the missing-suite check failed preflight and no PR build ran. CometJoinSpillingSuite, which runs CometJoinSuite with every native shuffled hash join spilling, is declared in CometJoinSuite.scala, so the check does not require it, but suites run only when they are listed.

Under greedy_unified the join has no share of its own and takes memory from Spark's off-heap pool until Spark refuses. CometSpillingHashJoinGreedyPoolSuite runs with 32 MiB of off-heap memory, so that Spark refuses before a task's build side fits, and checks that the join spills and matches Spark. It forces the hash join on the Comet side only, so that Spark's own run keeps a sort-merge join it can spill.

The design now describes what the operator does: routed parts kept as they are for resident buckets and held by spill files' writers until they make a full batch, a spilled bucket's build side read back compacted, and the Bloom filters that keep probe rows matching nothing off disk. The status section has CometHashJoinMemoryBenchmark and the native benchmark on an idle machine, before and after, and the remaining work is updated with what the profiles showed.

HashJoinExec concatenates its build batches into one copy and holds the batches and the copy at once while it builds its table, so a spilled bucket needed twice its build side in memory to be joined, and had to be split again when it did not fit. Given a single batch, HashJoinExec keeps it as it is, because arrow's concat returns a lone array unchanged. A spill file's writer now records the value bytes of each column, and a spilled bucket whose columns are all flat, fixed-width, boolean, string or binary, is read back one batch at a time into buffers allocated once at their final size. Its inner join gets that one batch, and no concatenated copy is reserved for it. A build side with nested, dictionary or view columns is still read back compacted. In the native benchmark under a 10 MiB pool, no bucket has to be split again any more, the spilled bytes halve, and the join takes 38% less time. Larger pools, where no bucket was split, are unchanged.

The design describes how a spilled bucket with flat columns is read back into one batch that its inner join keeps without a copy. The status tables are rerun on the final code, the native benchmark's 10 MiB case now needs no second split, and the remaining work records that TPC-H matches its golden answers with spilling forced, and that arrow-rs already tried reusing the lz4 encoder.

TPC-DS at scale factor 1 matches its golden answers with spilling forced, as TPC-H does. With forceShuffledHashJoin, 10 of the 611 native shuffled hash joins in the TPC-DS plans cannot spill because Spark relies on their output order. And in CometHashJoinMemoryBenchmark at 64 MiB, zstd makes the spilling join 14% slower than lz4 for 39% fewer bytes spilled, and uncompressed files 4% faster for twice the bytes.

apache/datafusion#24592 fixed the inflated read-back of spill files on DataFusion's main branch after 55.2, closing apache/datafusion#17340, and was not backported to branch-55. The design now cites it rather than listing the finding as one to report.

A shuffled hash join reports its streamed side's order, for example when it probes the sorted output of a sort-merge join below it, and the serde kept every such join off the spilling join, because rows of spilled buckets come out after the others. But the order matters only if an operator above uses it. In TPC-H q21 at scale factor 100 with forceShuffledHashJoin, the NOT EXISTS anti join probes the output of the EXISTS sort-merge join and feeds a shuffle, which discards any order. It could not spill and failed when its build side did not fit. CometExecRule now tags each shuffled hash join whose order no operator above relies on: none requires a child ordering the order could satisfy, and the order does not reach the root, whose output order a query relies on once RemoveRedundantSorts has dropped a sort. An operator that reports no order or sorts its input itself ends the order. A tagged join can spill, and a spilling join reports no output order.

…es pin only their own bytes (#24592) (#26435) ## Which issue does this PR close? - Fixes #17340 on the `branch-55` release line. - Backports the merged fix in #24592. ## Rationale for this change In 55.x, a batch read back from a spill file can hold far more memory than was recorded for it when it was written: up to twice as much when its IPC message spans two of the 128 KiB chunks that `SpillReaderStream` reads, and the whole 128 KiB chunk when a small batch fits inside one. Operators that read spills back budget memory from the size recorded at spill time, among them `SortExec`'s multi-level merge and the spilling aggregate, so under a memory limit they hold memory the pool does not know about. The `Record batch memory usage (...) exceeds the expected limit` warnings reported on #17340 come from this. #24592 fixed it on `main`, but it is not in 55.1.0 or 55.2.0, and it first ships in 56.0.0 (#24461). This backport is for a 55.3.0 patch release, if one is made. DataFusion Comet runs on 55.2.0 and hits this. In its spilling shuffled hash join (apache/datafusion-comet#7057), a spilled build side read back in batches of 256 KiB held up to twice what it held when written, so the join now copies every batch it reads back into buffers of their own size. Comet's sorts and aggregates spill through the same reader. ## Past backports of spill fixes Fixes to spilling have been backported to earlier release lines: | Release line | Backport | Fix | | ------------ | -------- | -------------------------------------------------------------------------------- | | `branch-54` | #23654 | Lost wakeup in `SpillPoolReader` with multiple concurrent readers (#23522) | | `branch-53` | #21451 | Use the spill writer's schema instead of the first batch's for spill files (#21293) | | `branch-53` | #20918 | Duplicate group keys after hash aggregation spill (#20724, #20858) | | `branch-53` | #20792 | Repartition dropping data when spilling (#20672) | | `branch-52` | #21403 | Use the spill writer's schema instead of the first batch's for spill files (#21293) | | `branch-52` | #20917 | Duplicate group keys after hash aggregation spill (#20724, #20858) | | `branch-52` | #20777 | Repartition dropping data when spilling (#20672) | | `branch-52` | #20684 | Arrow spill underrun when reading spill files back (#20159) | #20684 is the closest: like this one, it fixed how spill files are read back. ## What changes are included in this PR? - A cherry-pick with `git cherry-pick -x` of the commit that #24592 merged (cf083e0). It applies to `branch-55` without conflicts, and this PR changes nothing else. - `SpillReaderStream` reassembles each IPC message into allocations sized from the message's own headers before decoding it, so a decoded batch pins only its own bytes. The change is confined to `datafusion/physical-plan/src/spill/mod.rs`, with the tests from #24592. ## Are these changes tested? Yes, with the tests that #24592 added, on `branch-55`: - `cargo fmt --all -- --check` - `cargo clippy -p datafusion-physical-plan --all-targets --all-features -- -D warnings` - `cargo test -p datafusion-physical-plan --lib spill`: 93 passed, including the five tests #24592 added (`test_read_back_does_not_inflate_batch_memory`, `test_stringview_spill_read_back_memory_accounting`, `test_message_framer_across_chunk_boundaries`, `test_message_framer_reports_truncation_like_the_decoder` and `test_truncated_spill_file_errors`) - `cargo test -p datafusion-physical-plan --lib sorts`: 81 passed - `cargo test -p datafusion-physical-plan --lib aggregates`: 179 passed - `cargo test -p datafusion --test core_integration memory_limit`: 31 passed ## Are there any user-facing changes? No API changes. Queries that spill hold less memory when they read their spills back, and the warning from #17340 no longer fires for this cause. Co-authored-by: Jay Zhan <jayzhan211@gmail.com>

Describe the eligibility rule that walks up from each shuffled hash join, add the TPC-H results at three off-heap sizes, and record that q21 fails at 1 GiB because the join takes the memory a sort below it needs. Lead the remaining work with that, with what the fair pool and a build buffer cap showed.

…ototype Add TPC-DS at scale factor 100 with 2 GiB, where every query completes with spilling and q72 shows that a forced hash join can lose badly to the sort-merge join on a high fan-out join, spilling or not. Record that a pool limiting only the consumers that can spill, over the task's share, removes the q21 failure and every retried task in TPC-H for up to 3% more time.

andygrove added 4 commits October 16, 2026 12:07

test: add a benchmark of hash and sort-merge joins under native memor…

0677330

…y budgets

docs: add the spilling hash join design and document the join

6486ee6

github-actions Bot added the enhancement New feature or request label Oct 16, 2026

andygrove added 11 commits October 16, 2026 12:16

docs: record the spilling hash join's status and remaining work

268cb8c

andygrove added the run-spark-4.1-tests Run the Spark 4.1 SQL tests on this pull request instead of waiting for the merge queue label Oct 16, 2026

andygrove mentioned this pull request Oct 16, 2026

[branch-55] fix: frame IPC messages when reading spill files so batches pin only their own bytes (#24592) apache/datafusion#26435

Merged

andygrove added 2 commits October 16, 2026 18:05

Navigation Menu

Repository navigation

feat: spilling shuffled hash join (research draft) - #7057

feat: spilling shuffled hash join (research draft)#7057
andygrove wants to merge 19 commits into
apache:mainfrom
andygrove:spilling-hash-join-research

andygrove commented Oct 16, 2026 •

edited

Loading

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

1 participant

Conversation

andygrove commented Oct 16, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Which issue does this PR close?

Rationale for this change

What changes are included in this PR?

Findings so far

TPC-H at scale factor 100

TPC-DS at scale factor 100

Open questions

How are these changes tested?

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

1 participant

andygrove commented Oct 16, 2026 •

edited

Loading