External Sorting

Every sort we have met so far, heapsort, mergesort, the linear-time sorts, shares a hidden assumption: the whole array fits in fast memory, and any element is as cheap to touch as any other. When the data is larger than RAM, a database table of a hundred gigabytes on a machine with eight of them, that assumption collapses. The array lives on disk, and the cost of the sort is no longer how many times we compare but how many times we move a block between disk and memory. This lesson is about sorting under that constraint: external, or out-of-core, sorting.

Why the in-memory cost model lies

A disk does not behave like slower random-access memory. Reaching an arbitrary byte costs a seek (moving the head, or, on flash, an erase-block penalty) that is four to five orders of magnitude slower than a memory reference. To amortize that latency, storage is read and written in fixed-size blocks of $B$ records at a time; once the head is positioned, the marginal cost of the rest of the block is small. The realistic accounting therefore counts block transfers, not comparisons.

This change of cost model reorders the algorithm rankings. Consider running ordinary quicksort on data that does not fit in memory, with the array paged in and out by the operating system. Its partition step sweeps the array with two pointers that jump unpredictably, and its recursion touches scattered regions. Each stray access can force a fresh block transfer, so an algorithm that is $\Theta (N\log N)$ in comparisons can degrade toward one block transfer per comparison, thrashing the disk. The $\log N$ that looked harmless is now multiplied by a constant measured in milliseconds.

The design goal flips accordingly. We want an algorithm whose disk traffic is a handful of linear sequential passes over the data, and we will measure it by counting those passes.

External merge sort

The right strategy descends directly from mergesort, whose merge step is already a sequential streaming operation: it reads two sorted inputs front to back and writes one sorted output front to back, never seeking. That is the access pattern disk handles cheaply. External merge sort is mergesort reorganized into two phases that respect the block model.

Phase 1 — run formation. Read the file one memory-load at a time. Each load of up to $M$ records is sorted in memory by any internal sort, heapsort or quicksort, then written back out as a sorted run. A file of $N$ records yields about $\lceil N/M\rceil$ initial runs, each of length $M$ (the last possibly shorter). This phase reads the whole file once and writes it once: $2\lceil N/B\rceil$ transfers, two sequential passes.

Phase 2 — merging. We now have many sorted runs and must combine them into one. Merging two runs at a time, as plain mergesort does, would take ${\log }_2(N/M)$ passes over the data. We do far better by merging $k$ runs at once in a single pass, a $k$-way merge, repeating until one run remains.

The k-way merge

The core step is selecting, repeatedly, the smallest record among $k$ sorted runs. Naively scanning all $k$ run fronts on every step costs $\Theta (k)$ per record; across $N$ records that is $\Theta (Nk)$ comparisons, throwing away the savings of a large fan-out. The fix is a structure we already own: a min-heap, the priority queue from the heap lesson, holding one candidate, the current front record, from each run. Its root is the global minimum, extracted in $O(\log k)$, and when a run supplies its next record we sift that in for another $O(\log k)$.

The heap never holds more than $k$ records, one per run, so it occupies $\Theta (k)$ memory regardless of how long the runs are. Each of the $N$ output records costs one $\text{Extract-Min}$ and at most one $\text{Insert}$, so a $k$-way merge runs in $\Theta (N\log k)$ comparisons, improving the per-record cost from $\Theta (k)$ to $\Theta (\log k)$ over the naive scan.

from __future__ import annotations

import heapq
from collections.abc import Iterable, Iterator
from typing import Generic, TypeVar

from comparable import Comparable

Record = TypeVar("Record", bound=Comparable)

def k_way_merge(runs: Iterable[Iterable[Record]]) -> Iterator[Record]:
  """
    Lazily merge several already-sorted runs into one ascending stream.\n
    Each run must itself be sorted; the output is the sorted union (with\n
    duplicates kept). Streams the result, so a run never needs to live in\n
    memory all at once — only one record per run does.\n
  """
  iterators: list[Iterator[Record]] = [iter(run) for run in runs]

  # each heap entry pairs a record with its run index so equal records break
  # ties on the run they came from rather than ordering the iterators.
  heap: list[HeapNode[Record]] = []
  for run_index, iterator in enumerate(iterators):
    first = next(iterator, _SENTINEL)
    if not isinstance(first, _Sentinel):
      heapq.heappush(heap, HeapNode(first, run_index))

  while heap:
    # emit the global minimum front record.
    node = heapq.heappop(heap)
    yield node.record

    # refill the heap from exactly the run we just drained a record from.
    following = next(iterators[node.run_index], _SENTINEL)
    if not isinstance(following, _Sentinel):
      heapq.heappush(heap, HeapNode(following, node.run_index))

class _Sentinel:
  """
    A unique end-of-run marker distinct from any real record, including None.\n
  """

_SENTINEL: _Sentinel = _Sentinel()

class HeapNode(Generic[Record]):
  """
    One heap slot in the explicit k-way merger: a run's current front record\n
    paired with the index of the run it belongs to. Ordered by record, so a\n
    min-heap of these surfaces the global minimum front.\n
  """

  def __init__(self, record: Record, run_index: int) -> None:
    self.record: Record = record
    self.run_index: int = run_index

  def __lt__(self, other: HeapNode[Record], /) -> bool:
    if self.record < other.record:
      return True
    if other.record < self.record:
      return False
    return self.run_index < other.run_index

  def __repr__(self) -> str:
    return f"HeapNode({self.record!r}, run={self.run_index})"

class KWayMerger(Generic[Record]):
  """
    An explicit min-heap merger over k sorted runs.\n
    Mirrors the textbook k-Way-Merge: seed the heap with each run's head,\n
    then repeatedly extract the minimum and reload from its run. Useful when\n
    you want to inspect the heap (its size is the live fan-out) rather than\n
    just consume the merged stream.\n
  """

  def __init__(self, runs: Iterable[Iterable[Record]]) -> None:
    self._iterators: list[Iterator[Record]] = [iter(run) for run in runs]

    # seed the heap with the head record of each non-empty run.
    self._heap: list[HeapNode[Record]] = []
    for run_index, iterator in enumerate(self._iterators):
      head = next(iterator, _SENTINEL)
      if not isinstance(head, _Sentinel):
        heapq.heappush(self._heap, HeapNode(head, run_index))

  @property
  def heap_size(self) -> int:
    """
      The number of live run fronts currently buffered — the active fan-out.\n
    """
    return len(self._heap)

  def __iter__(self) -> Iterator[Record]:
    return self

  def __next__(self) -> Record:
    if not self._heap:
      raise StopIteration

    # pop the global minimum, then reload from the run it came from.
    node = heapq.heappop(self._heap)
    following = next(self._iterators[node.run_index], _SENTINEL)
    if not isinstance(following, _Sentinel):
      heapq.heappush(self._heap, HeapNode(following, node.run_index))
    return node.record

from collections.abc import Iterable, Iterator, Sequence
from typing import TypeVar
from comparable import Comparable
from k_way_merge import k_way_merge


Record = TypeVar("Record", bound=Comparable)


def form_runs(
  records: Iterable[Record], memory_size: int
) -> list[list[Record]]:
  """
    Phase 1 — run formation. Read the input one memory-load of up to\n
    `memory_size` records at a time, sort each load internally, and emit it\n
    as one sorted run. Yields about ceil(N / memory_size) runs.\n
  """
  if memory_size < 1:
    raise ValueError("memory_size must be at least 1")

  # fill a memory load, sort it into a run, then start the next load.
  runs: list[list[Record]] = []
  load: list[Record] = []
  for record in records:
    load.append(record)
    if len(load) == memory_size:
      runs.append(sorted(load))
      load = []

  # sort the trailing partial load, if any.
  if load:
    runs.append(sorted(load))
  return runs


def merge_pass(
  runs: Sequence[Sequence[Record]], fan_out: int
) -> list[list[Record]]:
  """
    One merge pass with fan-out `fan_out`. Group the runs into consecutive\n
    batches of at most `fan_out` and k-way merge each batch into a single\n
    run, dividing the run count by `fan_out`.\n
  """
  if fan_out < 2:
    raise ValueError("fan_out must be at least 2")

  # fold each consecutive batch of up to fan_out runs into one merged run.
  merged: list[list[Record]] = []
  for start in range(0, len(runs), fan_out):
    batch: Sequence[Sequence[Record]] = runs[start : start + fan_out]
    merged.append(list(k_way_merge(batch)))
  return merged


def external_merge_sort(
  records: Iterable[Record], memory_size: int, fan_out: int = 2
) -> list[Record]:
  """
    Sort `records` end to end with external merge sort: form memory-sized\n
    runs, then repeatedly `fan_out`-way merge until a single sorted run\n
    remains. Returns that run as a list.\n
  """
  # phase 1: form memory-sized sorted runs.
  runs: list[list[Record]] = form_runs(records, memory_size)
  if not runs:
    return []

  # phase 2: each pass replaces fan_out runs with one until just one survives.
  while len(runs) > 1:
    runs = merge_pass(runs, fan_out)
  return runs[0]


def streaming_external_sort(
  records: Iterable[Record], memory_size: int, fan_out: int = 2
) -> Iterator[Record]:
  """
    External merge sort that yields the final sorted stream lazily. The last\n
    merge pass is consumed on demand, so the fully merged output never needs\n
    to be materialized in memory all at once.\n
  """
  # phase 1: form memory-sized sorted runs.
  runs: list[list[Record]] = form_runs(records, memory_size)
  if not runs:
    return iter(())

  # merge down to at most `fan_out` runs eagerly, then stream the final fold.
  while len(runs) > fan_out:
    runs = merge_pass(runs, fan_out)
  return k_way_merge(runs)

from typing import Any, Protocol, TypeVar


class Comparable(Protocol):
  """
    Anything orderable with `<` (int, float, str, tuple, date, …).\n
  """

  # `other` is position-only so built-ins (int, str, …), whose dunder
  # operands are position-only, structurally satisfy the protocol.
  def __lt__(self, other: Any, /) -> bool: ...
  def __gt__(self, other: Any, /) -> bool: ...
  def __le__(self, other: Any, /) -> bool: ...
  def __ge__(self, other: Any, /) -> bool: ...

How many passes?

This is the figure of merit. Phase 1 produces $\lceil N/M\rceil$ runs. Each merging pass reads every run once and writes the merged output once, a full sequential sweep of the data, and replaces $k$ runs with one, dividing the run count by $k$. We are done when the count reaches $1$.

Each pass moves the entire file, $\lceil N/B\rceil$ reads and $\lceil N/B\rceil$ writes, so the total I/O cost is

The base of the logarithm is what matters. Doubling the fan-out from $k=2$ to $k=4$ halves the number of merge passes, because ${\log }_{4}x=\frac{1}{2}{\log }_{2}x$. On a terabyte file with gigabytes of memory the difference between a binary merge and a wide one is the difference between dozens of passes and two or three.

import math


def pass_count(record_count: int, memory_size: int, fan_out: int) -> int:
  """
    The number of sequential passes external merge sort makes over the data:\n
    one run-formation pass plus ceil(log_fan_out(N/M)) merge passes. Returns\n
    1 when everything fits in a single memory load (no merge needed).\n
  """
  if memory_size < 1:
    raise ValueError("memory_size must be at least 1")
  if fan_out < 2:
    raise ValueError("fan_out must be at least 2")
  if record_count <= 0:
    return 0

  # one memory load needs only the single run-formation pass.
  initial_runs: int = math.ceil(record_count / memory_size)
  if initial_runs <= 1:
    return 1

  # plus ceil(log_fan_out(initial_runs)) merge passes to fold down to one.
  merge_passes: int = math.ceil(math.log(initial_runs, fan_out))
  return 1 + merge_passes

Why not make k enormous?

If a bigger $k$ always helps, why not merge all $\lceil N/M\rceil$ runs in one pass? Because fan-out is bounded by memory. During a merge the $M$ records of memory are divided into one input buffer per run plus an output buffer; each buffer must hold at least one block of $B$ records to keep the disk traffic sequential. With $k$ input buffers and an output buffer we need roughly $(k+1)B\le M$, so

Push $k$ past this and the buffers shrink below a block, the merge starts seeking within each run, and the per-record I/O cost explodes, exactly the thrashing we set out to avoid. So $k$ is chosen near $M/B$: large enough that ${\log }_k(N/M)$ is typically $1$ or $2$, small enough that every buffer is at least one block. There is also a CPU-side tension, the heap costs $\Theta (N\log k)$ comparisons, but on disk-bound sorts the I/O term dominates and the comparison cost is secondary.

def max_fan_out(memory_size: int, block_size: int) -> int:
  """
    The largest safe fan-out for a merge with `memory_size` records of memory\n
    and `block_size` records per block: k <= memory_size/block_size - 1, one\n
    block-sized input buffer per run plus one output buffer. Pushing past\n
    this shrinks the buffers below a block and the merge starts seeking.\n
  """
  if block_size < 1:
    raise ValueError("block_size must be at least 1")
  return max(2, memory_size // block_size - 1)

Replacement selection: longer initial runs

The pass count is $\lceil {\log }_k(N/M)\rceil$, driven by the number of initial runs, $N/M$. Anything that makes the initial runs longer shrinks $N/M$ and can remove a whole pass. Phase 1 as described caps each run at $M$, the memory size. Replacement selection beats that bound, producing runs of average length $2M$.¹

The idea treats memory as a min-heap of $M$ records that continuously consumes input and emits output, rather than sorting in fixed batches. Fill the heap with $M$ records. Repeatedly extract the minimum and append it to the current run; then read the next input record and decide where it goes. If it is $\ge$ the record just emitted, it can still belong to the current run, so insert it into the heap. If it is smaller, it cannot, so set it aside (mark it frozen) to seed the next run. The current run grows as long as incoming records keep up with the output; only when the heap is entirely frozen records do we close the run and start fresh.

Why $2M$ on average? Picture the input as a stream and the heap as a window of $M$ records sliding along it. A new record extends the current run whenever it is no smaller than the last output, which for random data happens about half the time at any moment, but the records that do extend the run push the window forward and let still more records qualify. The classic snowplow argument makes it precise: a plow clears snow on a circular road while snow keeps falling uniformly; in steady state the plow always has about twice its own length of snow ahead of it. The same balance yields runs of expected length $2M$, and on already-sorted or nearly-sorted input a single run covering the entire file.²

Roughly halving the run count this way often saves exactly one merge pass, a sizable fraction of the disk traffic. Modern external sorts combine wide fan-out merging with replacement-selection run formation, and on real workloads the whole sort finishes in a small constant number of passes over the data.

from __future__ import annotations

import heapq
from collections.abc import Iterable, Iterator
from typing import Generic, TypeVar

from comparable import Comparable

Record = TypeVar("Record", bound=Comparable)

class _FrozenEntry(Generic[Record]):
  """
    A heap entry tagged with the run generation it belongs to. Entries of the\n
    current generation order ahead of frozen (next-generation) ones, and\n
    within a generation they order by record value — so the heap surfaces the\n
    smallest record still eligible for the current run.\n
  """

  def __init__(self, generation: int, record: Record) -> None:
    self.generation: int = generation
    self.record: Record = record

  def __lt__(self, other: _FrozenEntry[Record], /) -> bool:
    # current generation outranks frozen; ties break on record value.
    if self.generation != other.generation:
      return self.generation < other.generation
    return self.record < other.record

def replacement_selection(
  records: Iterable[Record], memory_size: int
) -> Iterator[list[Record]]:
  """
    Yield the sorted runs produced by replacement selection over `records`\n
    with an `memory_size`-record heap. Each yielded run is itself sorted;\n
    runs average about twice `memory_size` on random data and may be a single\n
    long run on nearly-sorted input.\n
  """
  if memory_size < 1:
    raise ValueError("memory_size must be at least 1")

  source: Iterator[Record] = iter(records)

  # fill the heap with the first memory_size records, all in generation 0.
  current_generation: int = 0
  heap: list[_FrozenEntry[Record]] = []
  for record in _take(source, memory_size):
    heapq.heappush(heap, _FrozenEntry(current_generation, record))

  if not heap:
    return

  current_run: list[Record] = []
  while heap:
    entry = heapq.heappop(heap)

    # a generation bump at the heap root means every remaining record is
    # frozen: close the current run and reopen on the next generation.
    if entry.generation != current_generation:
      yield current_run
      current_run = []
      current_generation = entry.generation

    # emit the smallest eligible record and remember it as the run's tail.
    current_run.append(entry.record)
    last_emitted: Record = entry.record

    # pull the next input record; nothing left means just drain the heap.
    incoming = next(source, _NO_MORE)
    if isinstance(incoming, _EndOfInput):
      continue

    # >= the tail can still join this run; smaller must seed the next one.
    record = incoming
    if record >= last_emitted:
      heapq.heappush(heap, _FrozenEntry(current_generation, record))
    else:
      heapq.heappush(heap, _FrozenEntry(current_generation + 1, record))

  # flush the run still open when the heap empties.
  if current_run:
    yield current_run

def _take(source: Iterator[Record], count: int) -> Iterator[Record]:
  """
    Yield up to `count` records from `source`, stopping early if it drains.\n
  """
  # yield records until the count is met or the source runs dry.
  for _ in range(count):
    record = next(source, _NO_MORE)
    if isinstance(record, _EndOfInput):
      return
    yield record

class _EndOfInput:
  """
    A unique end-of-input marker distinct from any real record, incl. None.\n
  """

_NO_MORE: _EndOfInput = _EndOfInput()

from typing import Any, Protocol, TypeVar


class Comparable(Protocol):
  """
    Anything orderable with `<` (int, float, str, tuple, date, …).\n
  """

  # `other` is position-only so built-ins (int, str, …), whose dunder
  # operands are position-only, structurally satisfy the protocol.
  def __lt__(self, other: Any, /) -> bool: ...
  def __gt__(self, other: Any, /) -> bool: ...
  def __le__(self, other: Any, /) -> bool: ...
  def __ge__(self, other: Any, /) -> bool: ...

The disk-block connection: B-trees

The same accounting that shapes external sorting shapes external search. When a sorted index is too big for memory, a balanced binary search tree is a poor fit: its height is ${\log }_{2}N$, and following each child pointer can cost a fresh block transfer, so a lookup costs $\Theta ({\log }_{2}N)$ seeks. A B-tree fixes this with the same fan-out idea external merge sort uses. Pack each node with as many keys as fill one block, $\Theta (B)$ of them, so each node has $\Theta (B)$ children instead of $2$. The tree's height collapses from ${\log }_{2}N$ to

and a search touches only $\Theta ({\log }_{B}N)$ blocks — the same ${\log }_k$ savings, with the block size $B$ playing the fan-out $k$. A wide node is to search what a wide merge is to sorting.

Past the two-parameter model

External merge sort is tuned to two parameters, memory size $M$ and block size $B$, that it must be told. Two developments push past that dependence, and one scales the idea across an entire datacenter.

The optimal I/O bound. The pass-count analysis gives $\Theta (\frac{N}{B}{\log }_{M/B}\frac{N}{B})$ block transfers with fan-out $k\approx M/B$. Aggarwal and Vitter (1988) proved this is optimal: no external sorting algorithm, comparison-based, can do asymptotically fewer I/Os. So external merge sort is to the block model what mergesort is to the comparison model — provably the best possible up to constants.

Cache-oblivious sorting. External merge sort needs $M$ and $B$ hard-coded to size its buffers. A cache-oblivious algorithm hits the same optimal I/O bound without knowing $M$ or $B$, so a single binary runs optimally across every level of the memory hierarchy at once — registers, L1, L2, RAM, disk — each with its own unknown block size. Funnelsort (Frigo, Leiserson, Prokop, and Ramachandran, 1999) achieves this by merging through recursively-built $k$-funnels whose sizes form a geometric ladder; the recursion automatically arranges that whatever the true block size turns out to be, the data movement near that scale is sequential. It is the same $k$-way-merge idea, made self-tuning.

Sorting a datacenter: TeraSort. When the data outgrows one machine's disks, the merge fans out across a cluster. The MapReduce sort (and the TeraSort benchmark built on it) is external merge sort's distributed cousin: a sampling pass picks $k-1$ splitter keys that partition the key range into $k$ roughly equal bands, each node range-partitions its shard by shipping records to the node owning their band (the shuffle), and each node sorts its band locally. Concatenating the bands in order yields a globally sorted file — a distributed bucket sort whose splitters play the role of bucket boundaries, and whose per-node local sort is itself an external merge sort. Sorting a petabyte this way is a standard industry benchmark.

Solid-state storage. The model here charges a flat cost per block transfer, which suited spinning disks where the seek dominated. Flash storage changes the constants: random reads are nearly as cheap as sequential, but writes are expensive, must happen in large erase blocks, and wear the device out. Modern external sorts on SSDs therefore optimize for write minimization and large sequential writes rather than seek avoidance — the accounting shifts, but the core strategy, few sequential passes with wide fan-out, carries over.¹

Takeaways

• On data larger than memory the cost model changes: we count block transfers (I/Os), not comparisons, and seeks dwarf everything, so good algorithms move the disk in long sequential passes.
• External merge sort has two phases: form $\lceil N/M\rceil$ sorted runs of memory size, then repeatedly $k$-way merge them until one run remains.
• The $k$-way merge uses a min-heap of the $k$ run fronts to emit the next smallest record in $O(\log k)$, giving $\Theta (N\log k)$ comparisons and a single sequential pass per merge.
• The number of passes is $1+\lceil {\log }_k(N/M)\rceil$; larger fan-out $k$ cuts passes, but $k\lesssim M/B$ because every run needs a block-sized buffer in memory, else the merge starts seeking.
• Replacement selection forms initial runs of average length $\approx 2M$ (double the naive bound), shrinking $N/M$ and often removing a whole pass.
• The same block-fan-out idea powers the B-tree: packing $\Theta (B)$ keys per node makes search $\Theta ({\log }_{B}N)$ seeks instead of $\Theta ({\log }_{2}N)$ — the same ${\log }_k$ savings applied to search.