Module spillable_join 

Per-join selection of a spillable algorithm under a memory cap.

The failure this fixes

TPC-H q18 on a 4500 MiB executor:

Resources exhausted: Failed to allocate additional 1015.5 KB for
HashJoinInput[0] with 732.4 MB already allocated for this reservation -
205.0 KB remain available

Not a leak, not a mis-sized budget: the pool refused correctly. DataFusion’s hash join holds its entire build side in memory with no spill path, so when the pool is exhausted the operator has nowhere to put the overflow and the query fails. Sort-merge join spills.

Why per-join, and why this exists as a rule instead of a config bit

The first attempt (8f72a340, reverted) set datafusion.optimizer.prefer_hash_join = false for the whole session whenever a cgroup limit existed. Measured on the cluster, q2 — ten stages of joins whose build sides all fit comfortably — went from 189 s to past a 2400 s timeout. Sorting both sides of every join to rescue the one join that overflows is a catastrophic trade.

So the decision is made where the information is: at each hash join, from that join’s estimated build size against the per-task share of the query pool. Three deliberately conservative gates, each a direct lesson from the q2 regression:

1. No cap, no change. An embedded engine on 23 GB keeps hash joins.

2. Unknown statistics keep hash join. A missing estimate is not evidence of a big build side, and guessing “big” re-creates the blanket regression. The cost of guessing “small” wrongly is the status quo — q18 fails as it does today — while the cost of guessing “big” wrongly is a q2-shaped timeout on healthy queries.

3. The join mode must be convertible. Partitioned inputs are already hashed on the join keys — the distribution sort-merge needs — so the conversion adds per-partition sorts, not exchanges. CollectLeft converts too when the plan has a single partition, where sorting alone satisfies sort-merge. Anything else keeps hash join.

   This bullet used to read “CollectLeft build sides are small by construction”. They are not: CollectLeft is picked from an estimate and buffers the whole build side. Worse, a task engine plans with target_partitions = cores / slots, which is 1 on a 3-core, 3-slot executor — so every join was CollectLeft and the rule converted nothing at all while five SF100 queries died on it.

The sorts are inserted explicitly (with partitioning preserved) rather than left to EnforceSorting, because appended optimizer rules run after the enforcement passes — a requirement declared here would never be satisfied.

Which spillable algorithm

Sort-merge is not the only way to make a join spill, and it is the worse one: it sorts both inputs in full even when nearly all the data would have fitted, which is what cost q2 6.3x. crate::grace_hash_join partitions both sides by key and joins bucket by bucket instead — no sorting, and the buckets that fit never reach the disk.

So when grace is set the rule tries that first and keeps sort-merge as the fallback for shapes it refuses. It is off by default: sort-merge is what the SF100 sweeps have actually been measured against, and a newer operator earns the default by beating it on the cluster.

Structs

SpillableJoinSelection

Convert hash joins whose estimated build side cannot fit the per-task memory share into sort-merge joins, which can spill.

Constants

SPILL_JOIN_BUILD_BYTES_ENV

Environment override for the build-size threshold, in bytes.