Module grace_hash_join 

A hash join that spills.

The gap this closes

DataFusion 54’s HashJoinExec holds its entire build side in memory and has no spill path. joins/hash_join/exec.rs carries a comment that reads as if it did:

// Decide if we spill or not
let batch_size = get_record_batch_memory_size(&batch);
state.reservation.try_grow(batch_size)?;

There is no branch under that comment — the ? propagates, and it is the SF100 failure verbatim:

Resources exhausted: Failed to allocate additional 877.0 B for
HashJoinInput - 0.0 B remain available for the total memory pool

crate::spillable_join works around this by rewriting oversized hash joins into sort-merge joins, which do spill. That trade is expensive: sort-merge sorts both inputs in full even when almost all of the data would have fitted in memory, and it cost q2 a 6.3x slowdown on the cluster.

What this does instead

A classic hybrid grace hash join:

1. Buffer the build side while it fits a budget. If the whole build side fits, join in memory and never touch the disk — identical work to HashJoinExec, which is the right algorithm in that case.
2. On overflow, hash-partition both inputs into buckets spill files on the join keys.
3. Join bucket by bucket: each bucket’s build side is small enough to hold, so each bucket is an ordinary in-memory hash join.

Only the buckets that overflow pay for disk, and nothing is ever sorted.

Why the answer is the same

Rows join only to rows with equal join keys, and equal keys hash equal, so co-partitioning both sides on the same expressions with the same seed puts every row and all of its potential matches in the same bucket. The union over buckets is therefore the whole join — including the unmatched rows an outer join must emit, because a row’s bucket contains every row it could have matched, so “unmatched within the bucket” and “unmatched overall” are the same statement.

This is exactly the property the broadcast-join split bug violated: there the build side was replicated and the probe side split, so a task could call a row unmatched that another task had matched. Here both sides are partitioned by the same key, which is the safe case.

Delegation, not reimplementation

Each bucket is joined by a real HashJoinExec, built from the original join with its own builder, so join type, join filter, null equality, null-awareness and the built-in projection are DataFusion’s semantics unchanged. This operator only decides what data goes to which join; it never reimplements what a join means.

Likewise the node delegates schema(), properties() and the distribution requirements to the join it replaces, so substituting it cannot change anything a parent plan observes.

Known limits

• Skew. One key larger than the budget lands in one bucket and that bucket is still an in-memory join. Recursive re-partitioning would need a second hash seed, which BatchPartitioner does not expose. The bucket count is therefore chosen generously, and an over-budget bucket is warned about by name in join_bucket — with its size and the budget it broke — rather than being retried or passing silently.
• Disk. buckets temporary files per side stay open while partitioning.

Structs

GraceHashJoinExec

A hash join that partitions to disk when its build side does not fit.

Constants

GRACE_HASH_JOIN_BUCKETS_ENV

Override the number of hash buckets the build side is partitioned into.

GRACE_HASH_JOIN_ENV

Turn the grace hash join on. Absent or not truthy, the engine keeps the sort-merge conversion.

Functions

bucket_count

Buckets to partition into for a build side estimated at build_bytes against a per-task budget.

enabled

Whether the grace hash join is enabled for this process.