External Merge Sort: Runs, Fan-in, and the Sort-Merge Join

Sorting is the one relational operator that cannot be implemented in a single pass over its input without knowing the input’s size in advance. Every other operator has a streaming or bounded-memory form: a filter streams, a hash join builds one side, an aggregate keeps a table of groups. A sort must see all of its input before it can emit its first row, and its input may be larger than the memory available to the query. The external merge sort is the standard answer, and its structure determines how a database behaves when a query exceeds its memory budget.

The algorithm is old and stable: build sorted runs that fit in memory, then merge them. What differs between implementations is how runs are built, how many are merged at once, and what happens when the join above the sort can consume sorted input directly.

Runs, memory, and the cost formula

Suppose the available sort memory holds M tuples and the input has N tuples. The naive approach reads M tuples, sorts them in memory, writes them out as one run, and repeats, producing ⌈N/M⌉ runs. The merge phase then reads all runs, merges them with a priority queue of size equal to the number of runs, and writes the result.

Cost in I/O is roughly 2N for the run-building pass (one read, one write) plus 2N per merge pass. If all runs fit into a single merge pass, the total is 4N tuple transfers. If not, the merge is multi-pass: each pass merges F runs into one, where F is the fan-in, so the number of passes is ⌈log_F (N/M)⌉, and the total I/O is 2N · (1 + passes).

The fan-in is limited by memory, not by the file system. Each input run in the merge needs at least one buffer page, plus one page for the output, so with M pages of memory the fan-in is roughly M - 1. If memory is 64 pages and the input is 6400 pages, the run-building pass produces 100 runs, and a single merge pass with fan-in 63 can absorb them; two passes are needed only if the run count exceeds the fan-in.

That formula is why the practical tuning knob is the number of runs, not the merge width. Doubling memory halves the run count and can eliminate an entire merge pass, which is a factor-of-two change in total I/O. Widening the fan-in beyond what memory allows means reading input buffers from disk repeatedly, which is worse than an extra pass over sequential data.

Replacement selection and average run length

Naive chunking produces runs of exactly M tuples. A heap-based variant called replacement selection produces runs averaging twice that length, which halves the run count for the same memory and can remove a merge pass.

The mechanism: fill a heap of size M with input tuples. Emit the smallest tuple to the current run. Read the next input tuple; if it is larger than the last one emitted, insert it into the heap and continue the current run. If it is smaller, it cannot belong to the current run without breaking sorted order, so it is held for the next run while the heap continues to supply the current one. When the heap contains only tuples held for the next run, the current run ends and a new one starts.

The average run length is about 2M because each input tuple has roughly even odds of being larger or smaller than the last emitted one, and the heap’s contents drain over a run of that length. The variance is large, and the worst case is an input in reverse sorted order, where almost every tuple is smaller than the last emitted and the average run length falls back toward M.

import heapq
from itertools import islice

def replacement_runs(stream, capacity):
    source = iter(stream)
    current = list(islice(source, capacity))   # heap of the run being built
    heapq.heapify(current)
    deferred = []                              # tuples too small for this run
    run = []
    for item in source:
        value = heapq.heappop(current)
        run.append(value)
        if item >= value:
            heapq.heappush(current, item)      # extends the current run
        else:
            heapq.heappush(deferred, item)     # belongs to the next run
        if not current:                        # every entry is deferred
            yield run
            run, current, deferred = [], deferred, []
    while current or deferred:                 # drain the last runs
        while current:
            run.append(heapq.heappop(current))
        if run:
            yield run
        run, current, deferred = [], deferred, []

The two-heap form makes the transition trivial: when the current heap drains, the deferred heap becomes the current one and a fresh run starts. Total entries in both heaps never exceed capacity, because each iteration removes one and adds one, so the memory bound holds at every step. The check item >= value compares against the value just emitted, which is the only ordering constraint the run needs; a tuple smaller than that cannot be appended without breaking sortedness and must wait.

Replacement selection has a cost that is easy to overlook. It requires reading the input strictly sequentially and holding tuples that belong to a future run, which uses memory that could otherwise hold more of the current run. The standard formulation uses the same total memory and accepts a slightly smaller heap, and the net gain in run length is still close to a factor of two on random input.

The merge phase and read-ahead

The merge is a k-way merge over sorted runs using a heap of the current head of each run. Each pop writes one tuple to the output and refills that run’s slot from its buffer. The inner loop is dominated by comparisons and buffer management, not by I/O, provided each run has at least one page of buffer and the pages are read with read-ahead.

Read-ahead is the part that separates a merge that runs at sequential I/O speed from one that runs at random I/O speed. Each input run is a separate sequential stream; the operating system’s read-ahead heuristic often fails with many concurrent streams because it cannot tell which pages will be needed next. Databases therefore issue explicit prefetch requests for the next page of each run while the current one is being consumed, which keeps the device queue full.

The output side has a matching concern. Writing the merged result with buffered writes coalesces small tuples into full pages; writing each tuple with a separate system call would dominate the cost. Implementations buffer a full page, and, when the merged output is the final result of the query rather than an intermediate, may bypass the buffer pool to avoid evicting useful pages.

Sort-merge join without a second sort

A sort-merge join requires both inputs sorted on the join key. When both inputs arrive sorted, the join is a single pass: advance the two cursors, compare keys, and on equality collect the matching groups from both sides before advancing past them.

The group collection is the part that makes the join’s cost depend on data rather than on input size. If the key is unique on both sides, each match is a single pair. If one side has many duplicates of a key, the join must produce the cross product for that key, and the number of output rows can exceed the input sizes by a large factor. The standard implementation buffers the duplicates of one side for the current key so that the other side can be replayed against them, which means memory proportional to the duplicate count and a spill if it grows too large.

The join’s advantage over hash join is that neither input needs a hash table. For inputs that are already sorted and too large to build a hash table from, this is decisive. The disadvantage is the requirement: if either input is unsorted, the sort costs O(N log N) and may spill, and the merge join then has no advantage over a hash join built from the smaller input.

The interesting case is when the sort can be avoided entirely. An index scan on the join key delivers rows in key order, so one side is sorted for free. A merge join over an index scan and a sorted intermediate result is the classic plan for a large table joined to a small sorted one. The planner has to know the index’s ordering, and has to avoid inserting an operator that destroys it; a filter does not destroy order, but a parallel scan that merges worker outputs may.

When the sort is not the bottleneck

A sort’s I/O cost is bounded by the number of passes, and for inputs under the memory budget the whole sort is in memory and costs only comparisons. The comparisons are not free: a tuple comparison may involve multiple columns, collation rules for text, and a tie-break on a hidden column to make the sort stable. A sort on a wide text column can be dominated by comparison cost rather than by I/O, which is why some systems sort on a hash or on a prefix of the value and verify the full comparison only for ties.

A second subtlety is that a sort’s output order can be partially satisfied by the plan. If the query needs ORDER BY a, b and the input is already sorted on a, a sort that only orders within each a group is cheaper than a full sort, and a merge of pre-sorted groups exploits that. Systems that track “interesting orders” through the plan can choose such a path; systems that treat sort as an opaque operator cannot.

Memory allocation is the failure mode that matters most in practice. A sort that is given less memory than it needs spills, and a spill multiplies the cost by the number of passes. The allocator typically grants memory per operator, so a plan with two concurrent sorts gets half the budget each, and a plan with a sort under a hash join competes with the hash table. The interaction between operators for memory is a global property of the plan, not a local property of any one operator, which is one reason plan choice and memory grant are decided together. Hash-based operators face the analogous problem in hash joins and hash aggregation, and the memory accounting that decides whether a sort fits is shared with the grouping operators described in grouping, duplicate elimination, and grouping sets.