Joins, external sort, and aggregation

Where this sits in the engine

Hold the running query in mind. Most of the course follows SELECT name FROM emp WHERE salary > 75000 ORDER BY name, which has no join. So widen it for one lesson. Imagine the real reporting query is SELECT e.name, d.dname FROM emp e JOIN dept d ON e.dept_id = d.id WHERE e.salary > 75000 ORDER BY e.name. Now there are two tables to combine and a final sort, and the executor has to decide how.

The layer above is the optimizer (week 11). It already chose the physical operators: it picked a join algorithm and a scan type, costed them, and handed down a physical plan tree. This lesson is the menu the optimizer was choosing from, plus the exact cost formulas it used to choose. The layer below is the access method and buffer pool: every operator here reads and writes pages through the buffer pool (week 4), which is why buffer count B shows up in every cost formula on this page. An index nested loop join leans on the B+tree from week 6.

And the operators themselves are the iterators from week 9. Each join, sort, and aggregate is an iterator with open / next / close. The new wrinkle this week is that some of them cannot stream: a sort or a hash build has to swallow its entire input before it can hand out the first tuple. Those are the pipeline breakers, and they are exactly where memory runs out and the operator spills to disk.

The problem: the product is quadratic

A join of R and S is the Cartesian product R × S with the matching rows kept. There are |R| × |S| pairs to consider. For two tables of a million rows that is 10¹² comparisons, which is not a number you want in a per-query budget. Every join algorithm is a scheme to look at far fewer than all the pairs.

Nested loop and why buffers matter

The naive nested loop is two for-loops: for each tuple of the outer relation R, scan all of the inner relation S. With M pages in R holding m tuples and N pages in S, the cost is

# naive: reread all of S once per outer TUPLE
cost = M + (m * N)

The m * N term is the killer. On the standard CMU worked example (M=1000, m=100000, N=500, B=100 buffers, 0.1 ms per I/O) naive nested loop is about 1.4 hours [CMU 15-445 L12]. The block variant fixes the obvious waste: reread S once per outer block, not per tuple. Give the outer scan B − 2 frames (one frame holds an inner page, one holds output):

# block: reread all of S once per outer BLOCK
cost = M + (ceil(M / (B - 2)) * N)

On the same example this drops to about 6.5 seconds. This is the first place buffer count earns its keep: more frames mean a bigger outer block, fewer rereads of S, lower cost. The index variant replaces the inner scan with an index probe at constant cost C per outer tuple:

# index: probe an index on S's join key per outer tuple
cost = M + (m * C)

Index nested loop wins when S has an index on the join column and either R is small or the join is selective. PostgreSQL flags this case directly: nested loop "can be a good strategy" specifically "if the right relation can be scanned with an index scan" [PG planner].

Sort-merge join

Sort both inputs on the join key, then walk two cursors forward in lockstep and emit matches, exactly like the merge step of merge sort. Because most keys are mostly unique, the merge is roughly M + N: one pass over each side. The total is the cost of the two sorts plus the merge.

The worst case is pathological: if every tuple shares one join value, the merge degrades to M * N because each outer page must rescan the whole inner side for that single value. That is the same quadratic blow-up as nested loop, hiding inside a merge.

Hash join, in memory and on disk

In-memory hash join has two phases. Build: scan the smaller (build) input and insert each tuple into a hash table keyed on the join attribute. Probe: scan the larger (probe) input, hash each tuple, look up matches, and re-check the actual join values because buckets collide. When the build side fits in memory the cost is about M + N, the cheapest of the lot (about 0.45 s on the CMU example).

When the build side does not fit in memory, in-memory hashing breaks. Grace (partitioned) hash join fixes it. Phase 1, partition: hash both inputs with one hash function h1 into the same set of partitions written to disk, so that R partition i can only match S partition i. Phase 2, probe: for each partition pair, build a hash table on the smaller side and probe with the other. If a single partition still does not fit, recursively repartition it with a different hash function h2.

# grace: read once + write once to partition, then read once to join
partition_cost = 2 * (M + N)     # read both, write both
probe_cost     = (M + N)         # read both partitioned files once
total          = 3 * (M + N)

That is linear in the input size, which is why grace hash at 3*(M+N) crushes naive nested loop at M + m*N: 0.45 seconds versus 1.4 hours on the same tables. PostgreSQL implements the hybrid variant, which keeps one partition resident in memory and probes it immediately instead of spilling it [nodeHashjoin.c].

Simulator: the join algorithm cost playground

Set the table sizes, the buffer frames, and the join selectivity, then read the exact I/O cost of all five algorithms at once. Experiment: drag B up and watch block nested loop and the sort passes fall; make the join selective and watch index nested loop take the lead; push the build side over the memory limit and watch in-memory hash flip to grace and gain the partition pass. The bars are log-scaled wall time so a quadratic loser and a linear winner fit on one chart.

Join cost playground exact I/O formulas from CMU 15-445 L12

R pages M

R tuples m

S pages N

buffers B

index probe C

ms per I/O

build side over memory (force grace)

External merge sort: sorting more than fits in memory

Sort-merge join and sort-based aggregation both need to sort, and the input is often larger than memory. You cannot sort in place, so you sort in two parts. Run generation (pass 0): read the input B pages at a time, sort each chunk in memory, write it out as a sorted run. That produces ceil(N / B) sorted runs. Multi-way merge: merge runs B − 1 at a time (one input buffer per run, one output buffer), repeatedly, until one run remains. Each pass cuts the run count by a factor of B − 1.

passes  = 1 + ceil(log_{B-1}(ceil(N / B)))   # the "1+" is run generation
total_IO = 2 * N * passes                    # every pass reads and writes all N pages

External merge sort pass counter watch the formula emerge

data pages N

buffer pages B

Aggregation two ways

A GROUP BY (or DISTINCT) has to bring equal keys together. There are two ways, and they are duals of the two join strategies.

Sort-based versus hash-based aggregation.

Property	Sort-based	Hash-based
Method	sort on the group key, then break into a new group whenever the key changes	one hash-table entry per distinct group, updated in place
State held	one running group at a time	one entry per distinct group
Wins when	input already sorted on the key, or output must be sorted	few distinct groups relative to input size
Output order	sorted on the group key	no particular order
Spills	via the external sort	overflow groups to per-partition files (grace-style)

Pipeline breakers: where memory bites

Last week the iterator model let tuples stream through filter and projection one at a time. Some operators cannot stream. A sort cannot know the first sorted tuple until it has seen the last input tuple. A hash join's build side must be fully built before any probe can match. Aggregation and DISTINCT cannot emit the first group until the input ends, because the last tuple might belong to that group. These are the pipeline breakers, and they are exactly where a whole input has to live in memory, so they set the query's memory budget and are where spilling to disk begins.

The cost cheat sheet

I/O cost of each join algorithm. M,N are pages; m is tuples in R; B is buffer frames; C is the per-probe index cost.

Algorithm	I/O cost	Wins when
Naive nested loop	M + m·N	tiny inner, or no other option
Block nested loop	M + ceil(M/(B−2))·N	more buffers available
Index nested loop	M + m·C	index on inner key, selective join
Sort-merge	sort(M) + sort(N) + (M+N)	inputs pre-sorted, or output sorted
Grace hash	3·(M+N)	equi-join, build side over memory

Deeper: why one general operator handles join, aggregation, and DISTINCT

The Volcano paper folds join, semi-join, all three outer joins, anti-join, intersection, union, difference, aggregation, and duplicate elimination into one physical operator it calls one-to-one match (Section III.B.2). The unifying step is the same for all of them: separate the matching and non-matching subsets of two inputs, then emit the subsets you want, possibly transformed. A join emits the matching pairs; a semi-join emits matching R rows once; aggregation treats one input as empty and groups the other. Both the hash and the sort-merge implementations live inside that one module, which is why the same overflow machinery (the packing and spilling thresholds) serves a hash join and a hash aggregation. Graefe even added a "flush" phase to the hash version specifically so aggregation could emit its groups after the build, and the build / probe / flush phases map cleanly onto open (build) and next (probe then flush).

Active recall

Paper viva

This week shares its assigned paper with week 9: Graefe's Volcano. The examiner will probe how the abstract iterator model from last week meets the concrete join and overflow machinery of this one.

Model answer

In the hash one-to-one match, open performs the build: it consumes the entire first (build) input and inserts it into the in-memory hash table. The probe and flush phases both run inside next. Successive next calls probe the second input against the hash table and emit matches; when the second input is exhausted, next switches automatically to the flush phase, which is Graefe's addition so the same operator can serve aggregation (emit the accumulated groups). Because the entire build input is consumed in open before any output, the build side is a pipeline breaker. This is from Volcano Section III.B.2. The point to make is that even a stateful, memory-hungry operator hides behind the same three-procedure interface, so the operators above it never learn it is a hash join rather than a scan.

Model answer

Volcano keeps intermediate streams on virtual devices whose pages exist only in the buffer pool and vanish when unpinned (Section III.A, III.B). The cost argument is that routing data through real temporary files carries a substantial penalty and is used in neither real systems nor Volcano. Each record handed between operators is pinned in the buffer and owned by exactly one operator at a time; an operator may hold it, unfix it on a predicate failure, or pass it on. To keep buffer-manager call counts low the interface was redesigned to need two buffer calls per cluster on the producer side and one per cluster on the consumer side, independent of how many records a cluster holds. The connection to this week: this is exactly why a pipelined chain (scan to filter to projection to probe) never touches disk, while a pipeline breaker that overflows memory is the one place data does spill, through the same buffer and file machinery.

Model answer

The hash table points directly into buffer-resident records, with no copying. When the item count reaches the packing threshold, items are packed densely into overflow files but not yet written to disk. When it reaches the spilling threshold, the first partition file is unfixed and written to disk and the count is reduced; the cycle repeats and partitioning recurses with adjusted thresholds (Section III.B.2). Setting both thresholds to zero gives Grace-style overflow avoidance: nothing is kept resident, everything is partitioned to disk up front, which is the textbook grace hash join. This is Volcano's "mechanism, not policy" principle in miniature. The operator provides the spilling mechanism, and an optimizer sets the thresholds from estimated input-size distributions, choosing anywhere on the spectrum from fully in-memory hybrid hash to pure grace.

Check yourself