Hash Joins and Hash Aggregation: Build, Probe, and Spill

A hash join turns an equality predicate into a lookup. One input is consumed entirely and organised into a hash table keyed by the join columns; the other is streamed past it, each row probing for matches. The asymmetry is the whole design: the build side must fit in memory or be partitioned, while the probe side needs only a current row.

The same structure serves aggregation. A hash aggregate builds a table keyed by the grouping columns and accumulates state per group; the input is streamed and each row either creates a group or updates an existing one. Duplicate elimination is the degenerate case with no aggregate functions, where the table’s keys are the output.

Both operators have the same failure mode: the table grows past the memory granted to the operator, and the operator must either spill or abandon the algorithm. What the implementation does at that moment determines whether a query degrades by a factor of two or by a factor of a hundred.

Build-side mechanics and the probe loop

The build phase reads every row of the designated input, computes a hash over the join key values, and inserts the row into a bucket. Implementations use a chained hash table, an open-addressed table, or a two-level structure where a directory of partitions points to per-partition tables. The choice matters for cache behaviour: open addressing keeps probes within a contiguous array, while chaining follows a pointer per collision.

The probe phase reads the other input, hashes the key, walks the bucket, and for each candidate compares the full key. The comparison is necessary because the hash is not unique; a hash collision between different keys produces a false candidate, and only the key comparison rules it out. Skipping the comparison on the grounds that collisions are rare produces wrong results, not merely a performance problem.

The probe loop is the hot path, and its cost is dominated by memory access. Each probe touches the hash directory, then a bucket, then the stored key. A bucket that fits in cache makes the walk cheap; a bucket that spills to a different page makes it a cache miss per candidate. Implementations therefore size buckets to a cache line and store the hash value alongside the key, so that a mismatched hash rejects a candidate without touching the key bytes.

def build(rows, key_of, n_buckets):
    table = [[] for _ in range(n_buckets)]
    for r in rows:
        k = key_of(r)
        table[hash(k) % n_buckets].append((k, r))
    return table

def probe(table, rows, key_of, emit):
    n = len(table)
    for r in rows:
        k = key_of(r)
        for stored_key, stored_row in table[hash(k) % n]:
            if stored_key == k:        # full comparison, not just the hash
                emit(stored_row, r)

Two properties of the output are worth noting. First, a hash join emits rows in probe-side order within each bucket, which is no order at all from the query’s perspective; the operator cannot feed a merge join or satisfy an ORDER BY. Second, the join is not stable with respect to the build side’s duplicate keys: multiple matches come out in bucket order, not in build-side input order.

Which side is built

The build side should be the smaller input, because its size determines the memory requirement. Estimating which input is smaller is the planner’s job, and getting it wrong is expensive in a way that is invisible in the plan text: the join still works, it just spills.

There is a second criterion beyond size. If the build side’s rows are wider than the probe side’s, hashing them costs more memory per row, so a plan may build the smaller-in-rows but wider input. And if one side is already sorted and the other is not, a merge join may be cheaper than a hash join even when the hash join would fit, because the sort’s absence removes a whole operator.

Semi-joins and anti-joins reuse the structure with a change in the probe step. A semi-join stops at the first match and emits the probe row once; an anti-join emits the probe row only when no match is found. Both can store a bit per build row instead of the row itself when the join’s output needs no build-side columns, which shrinks the table substantially. Anti-join with NULL requires care: the standard’s NOT IN semantics treat a NULL on the build side as making every non-matching probe row unknown, so the operator must either detect the presence of a NULL key and switch to a null-aware mode or refuse the plan.

Grace partitioning and the spill path

When the build input exceeds the memory grant, the standard response is to partition both inputs by hash into a number of buckets chosen so that each partition fits in memory. This is the grace hash join. The build input is read once and written to partition files; the probe input is read and written to matching partition files; then each partition pair is joined independently with the in-memory algorithm.

The partition count is ⌈build_size / memory⌉ rounded up to a power of two, and the partitioning must use the same hash function as the in-memory table so that matching keys land in the same partition. This is the invariant that makes the algorithm correct: a key can only match a key in its own partition.

Skew is the failure mode. If one key value dominates the build input, its partition is as large as the whole input and cannot be processed in memory, no matter how many partitions exist. Implementations handle this by detecting oversized partitions after the first pass and recursively partitioning only those, or by isolating the most frequent key into a special partition processed with a nested loop over a smaller table. A recursive repartition of a single hot key never terminates, so the fallback must be a different algorithm for that partition.

The cost of a spilled hash join is the cost of writing and reading the partitioned inputs plus the cost of the in-memory join per partition. That is roughly three passes over each input instead of one, so the operator’s cost grows by a factor of two to four. The plan does not change, the estimates do not change, and the query simply becomes slower, which is why memory grants are a first-order tuning concern rather than a detail.

Hash aggregation and the group-state table

Hash aggregation is the same build-and-probe structure with the probe replaced by an update. For each input row, look up the group key; if absent, create a group with the aggregate’s initial state; then apply the aggregate’s transition function to the state with the row’s value. At the end, a finalization function turns each state into an output value.

The aggregate state is per-function and can be larger than the input value. An average keeps a sum and a count; a distinct count keeps a set of seen values, which can grow without bound; a percentile or a median keeps a collection that must be sorted at finalization. These are the aggregates that spill even when the number of groups is small, because the state per group is proportional to the number of distinct inputs in it. Implementations either impose a memory limit on such states or rewrite the aggregate into a sort-based form.

Grouping keys are compared the same way join keys are: hash first, then full comparison. A subtlety is that NULL groups together in GROUP BY, so the key comparison must treat NULL as equal to NULL, which is the opposite of the = operator’s semantics. Implementations normalise the key representation so that a single comparison function serves both.

The output order of a hash aggregate is arbitrary, which means an ORDER BY on the grouping columns needs a sort above it. A sort-based aggregate can produce grouped output in key order and avoid that sort, so the choice between hash and sort aggregation is partly a question of whether the query needs ordered output. The details of the sort-based form, including grouping sets and rollup, are in grouping, duplicate elimination, and grouping sets.

Skew, spilling, and the estimate

Every hash operator’s behaviour is decided by an estimate made before execution: the expected number of build rows, and therefore whether the table fits. That estimate comes from statistics on the base tables and from the selectivity of the predicates applied to them. A join whose estimated build size is a hundred rows but whose actual size is ten million will spill catastrophically, and the plan will show a hash join with no hint that anything went wrong.

The asymmetry is structural. An underestimate causes a spill and a large slowdown; an overestimate causes a plan that avoids the hash join entirely in favour of a nested loop, which is slower for the actual data but not catastrophic. Systems bias their estimates in the direction that fails gracefully, and the bias is visible in the cost formulas as a preference for the algorithm that does not depend on a memory grant.

A second structural point is that the memory grant is allocated before execution begins and is not renegotiated. An operator that spills cannot ask for more memory later, because the memory it would take is committed to other operators running concurrently. This is why a plan with several hash operators under one another is fragile: each gets a share, and any one of them spilling changes the whole plan’s timing. The problem compounds when operators run concurrently across workers, as described in parallel execution and the exchange operator.