External Merge Sort

Why External Merge Sort Matters

Suppose a database must sort a log larger than its RAM allowance. Loading the entire log into an array fails, while repeatedly fetching individual records from storage wastes transfers. External merge sort turns the job into bounded-memory sorting followed by sequential merging: sort manageable chunks into temporary runs, then repeatedly merge several runs until one sorted run remains.

The expensive unit is a storage block, not an individual comparison. One page read brings several records into a memory buffer; many comparisons can then reuse them. A well-designed external sort does more sequential work per transfer and avoids needing all records in its sorting buffers at once.

This lesson develops the block model, run generation, buffered multiway merging, stability, exact I/O accounting, and a correctness argument. The complete Python, C++, and Java programs simulate disk pages and record buffers explicitly. They expose every block read and write, handle duplicate keys and partial pages, and print identical results.

The prerequisites are ordinary comparison sorting and the page-oriented view of B-trees/B+ trees. A B+ tree maintains order across updates; an external sort organizes a whole batch, often before bulk loading an index or executing a query. The simulator uses host RAM to represent its disk, so it teaches the transfer algorithm rather than acting as a production file sorter.

Core Concepts and Terminology

Let $N$ be the number of fixed-size records, $B$ the number of records per storage page, and $C$ the number of available record-buffer frames. The modeled data-buffer capacity is $M=BC$ records. The input already occupies $D=\lceil N/B\rceil$ disk pages; an empty input occupies zero pages. A partly filled page still costs one transfer.

The input is a sequence of integer keys. Each becomes a record $(k,i)$, where $i$ is its original zero-based position. All sorting compares the pair lexicographically: smaller key first, then smaller original position. The final keys are nondecreasing, every record appears once, and equal keys keep their original order. This supplies stability, even though the chunk sorter itself is heapsort.

A run is one sorted sequence, represented by its ordered page IDs and its record count. Its pages need not have adjacent physical IDs in this simulator. Run boundaries matter: two individually sorted runs cannot be concatenated into one sorted run without checking their key ranges or merging them.

A merge group of $g$ runs reserves one $B$-record input frame per run and one $B$-record output frame. Therefore $$ (g+1)B\le M,\qquad 2\le g\le f\le C-1, $$ where $f$ is the configured maximum fan-in. The heap stores run indices, whose comparisons inspect current buffered heads. Heap entries, page IDs, run descriptors, runtime objects, and scalar temporaries are metadata outside this record-slot budget. A real byte budget must also pay for them and for record representation.

Accounting convention. Each actual call to the simulated disk's read or write method counts one transfer. Creating the already-resident source image, allocating/freeing pages, manipulating metadata, and comparing buffered keys do not count. A page is read again if it is needed in another pass; the model has no cross-pass cache. The sort materializes its final disk run, then performs a separately labeled, counted readback for display. Returned observer records are outside the sorting-buffer budget.

For $N=17$, $B=2$, $C=3$, and $f=2$, three frames hold six records. Two input frames plus one output frame fit exactly. Increasing fan-in to three would require four frames and is rejected, even though a heap of three indices itself is tiny.

How External Merge Sort Works, Step by Step

1. Validate the model. Require $1\le B\le32$, $3\le C\le32$, and $2\le f\le C-1$. The teaching API accepts at most $4096$ integer keys between $-10^6$ and $10^6$, inclusive. These finite caps keep traces and counters manageable; they are not mathematical restrictions of external sorting. Invalid parameters are rejected even for empty input.

2. Build the uncounted source image. Pack input records in original order into pages of size $B$, with a possibly partial last page. Give every record its original index. This represents a file that was on disk before the timed sort began.

3. Generate initial runs. Read at most $C$ consecutive source pages, giving at most $M$ records. Reserve only the chunk's valid record count, heapsort the chunk in place by $(k,i)$, and write its records in $B$-sized pages. The sorted chunk itself supplies these writes; generation does not need an additional output record frame. Continue until the source ends, then release all source pages. Keeping the source until generation finishes makes the peak disk space easy to audit.

4. Form a merge group. Take up to $f$ consecutive runs. Carry a singleton group forward unchanged. For a real merge, reserve $(g+1)B$ slots, create one reader per input run, and read each first page. Insert each reader's index into the min-heap. Every run is nonempty, so its first head exists.

5. Emit the smallest head. Remove the heap's winning index and append that reader's head to the output frame. Write and clear the frame whenever it reaches $B$ records. Then advance that reader: continue within its current page, or clear/refill its frame from its next page. Reinsert its index only if another valid record exists. No other reader changes, so the other heap comparisons remain valid. Mutating a reader while its index is still in the heap would break this argument.

6. Finish the group and pass. When the heap is empty, flush any nonempty partial output frame. Record the new run's page list and total length. Only after its output exists, erase the input group pages and release its reservation. Repeat for every group. The next pass sees only the newly merged runs and any carried singleton.

7. Stop at one run, then observe it. Repeated passes reduce the run count until it is at most one. Use one input frame to read the final run into the returned demonstration list, counting these reads under readback. Empty input produces no pages, no records, and no transfers; its history still contains runs and readback phases.

The figure isolates one emission that fills an output page and exhausts an input page. These are independent events: an output flush is a write, and an input refill is a read.

flowchart TD
    A["Validate parameters and keys"] --> B["Pack the already-resident source image"]
    B --> C["Read and sort bounded chunks; write runs"]
    C --> D["Release source pages"]
    D --> E{"More than one run?"}
    E -->|No| O["Read back final run and report counters"]
    E -->|Yes| F["Begin next pass"]
    F --> G{"Another run group?"}
    G -->|No| H["Replace runs with successor runs"]
    H --> E
    G -->|Yes| I{"Singleton group?"}
    I -->|Yes| J["Carry run metadata unchanged"]
    J --> G
    I -->|No| K["Reserve frames; read heads into buffers"]
    K --> L{"Heap empty?"}
    L -->|No| M["Emit minimum; flush if full; advance and reinsert"]
    M --> L
    L -->|Yes| N["Flush partial page; release input pages and frames"]
    N --> G

Worked Example with a Full Trace

Use $B=2$, $C=3$, $M=6$, and $f=2$. The keys are $$ [9,1,5,3,5,2,8,0,7,3,6,4,2,9,1,8,0]. $$ Write key@index for a record. The nine source pages, in order, are [9@0,1@1], [5@2,3@3], [5@4,2@5], [8@6,0@7], [7@8,3@9], [6@10,4@11], [2@12,9@13], [1@14,8@15], and [0@16].

Initial run generation. Read three source pages per full chunk. Sorting and writing them gives these exact pages; a semicolon separates pages:

• $R_0$, length $6$: [1@1,2@5]; [3@3,5@2]; [5@4,9@0].
• $R_1$, length $6$: [0@7,3@9]; [4@11,6@10]; [7@8,8@6].
• $R_2$, length $5$: [0@16,1@14]; [2@12,8@15]; [9@13].

The chunks cost $3+3+3=9$ reads and $3+3+3=9$ writes. Run $R_2$ has five valid records spread across three pages. The live disk reaches $9+9=18$ pages before the source is deleted; the largest chunk reserves six record slots.

First merge pass. Merge $R_0$ and $R_1$. The winning record sequence is 0@7,1@1,2@5,3@3,3@9,4@11,5@2,5@4,6@10,7@8,8@6,9@0. Pairing consecutive winners yields six output pages:

[0@7,1@1]; [2@5,3@3]; [3@9,4@11]; [5@2,5@4]; [6@10,7@8]; [8@6,9@0].

This group reads its six input pages once and writes six output pages. Carry $R_2$ unchanged, giving lengths $[12,5]$. That carry incurs zero reads and zero writes. The sorted contents of every run after this pass are now explicitly known.

Second merge pass. Call the twelve-record run L and the carried five-record run R. Name L's pages L0 through L5, and R's pages R0 through R2 in the orders just shown. Initially read L0 and R0: two reads. The following list records every winning head, output-buffer state, flush, and subsequent refill. Each step first emits and possibly flushes, then advances the winning reader.

1. L wins 0@7; output [0@7]; no refill.
2. R wins 0@16; write [0@7,0@16]; output empty; no refill.
3. L wins 1@1; output [1@1]; read L1, headed by 2@5.
4. R wins 1@14; write [1@1,1@14]; output empty; read R1, headed by 2@12.
5. L wins 2@5; output [2@5]; no refill.
6. R wins 2@12; write [2@5,2@12]; output empty; no refill.
7. L wins 3@3; output [3@3]; read L2, headed by 3@9.
8. L wins 3@9; write [3@3,3@9]; output empty; no refill.
9. L wins 4@11; output [4@11]; read L3, headed by 5@2.
10. L wins 5@2; write [4@11,5@2]; output empty; no refill.
11. L wins 5@4; output [5@4]; read L4, headed by 6@10.
12. L wins 6@10; write [5@4,6@10]; output empty; no refill.
13. L wins 7@8; output [7@8]; read L5, headed by 8@6.
14. L wins 8@6; write [7@8,8@6]; output empty; no refill.
15. R wins 8@15; output [8@15]; read R2, headed by 9@13.
16. L wins 9@0; write [8@15,9@0]; output empty; L is exhausted.
17. R wins 9@13; output [9@13]; R is exhausted. The heap is empty: write the final partial page and clear the frame.

This pass has $2+5+2=9$ reads and nine writes: eight full pages plus one partial page. The nine flushed pages above are the final disk run. Their concatenation is $$ [0,0,1,1,2,2,3,3,4,5,5,6,7,8,8,9,9]. $$ For each repeated key, the original indices increase. Both sorted order and stability are visible, and all seventeen original IDs remain.

Final readback and totals. Read those nine final pages once for display: nine reads, no writes. Sorting alone used $9+9+6+6+9+9=48$ transfers. Including readback gives $33$ reads, $24$ writes, and $57$ transfers. Peak record-buffer reservation is six slots, and peak live disk storage is eighteen pages.

Implementation in Python, C++, and Java

Each panel is a complete program. Save the Python panel as main.py and run python3 main.py; save C++ as main.cpp and compile with g++ -std=c++17 -O2 main.cpp -o external-sort; save Java as Main.java and run javac Main.java followed by java Main. Python needs a version with dataclasses, available from Python 3.7 onward. Java uses ordinary classes, so no Java record feature is required.

The public teaching functions are external_sort(keys, b, memory_blocks, fan_in) in Python/C++, and externalSort(keys, b, memoryBlocks, fanIn) in Java. They do not mutate the caller's key sequence. Pass a Python list, C++ vector<int>, or Java int[]; null/nonsequence arguments are outside the API contract. Python checks exact integer types and rejects booleans and floats. C++ and Java expose integer-typed parameters. Across the shared integer input domain, invalid model parameters raise PARAMETER, more than $4096$ keys raises SIZE, and a key outside the declared range raises VALUE. Size is checked before key values.

Implementation map. Disk owns simulated pages and counters. Its source setup is uncounted; each later read/write method increments exactly one counter. Freed page slots are reused. Reader keeps one input page and refills only at its boundary. Ram tracks reserved record slots and asserts that reservations fit. heap_sort/heapSort sorts generation chunks in place, and the merge heap stores reader indices, comparing their buffered heads.

The returned result contains the observed records, read/write totals, peak record-buffer reservation, peak live disk pages, and a history with each phase's output run lengths and counter deltas. The last history entry is the optional readback. Subtracting its reads from total transfers gives sort_transfers.

Simulator boundary. Simulated disk pages, input keys, output observations, page IDs, heap indices, and history are stored in host memory. This implementation therefore uses $O(N)$ host storage overall, plus metadata; it is not an actual RAM-capped disk sorter. The $M$ limit applies to reserved sorting record buffers. In a real implementation, replace Disk with page I/O, budget metadata and runtime overhead in bytes, and stream final records to a consumer instead of accumulating the observer list. Copying a page into simulated disk storage represents a write destination, not an extra algorithm input/output frame.

from dataclasses import dataclass
import heapq


@dataclass(frozen=True, order=True)
class Record:
    key: int
    order: int


@dataclass
class Run:
    pages: list
    length: int


@dataclass
class Phase:
    name: str
    lengths: list
    reads: int
    writes: int


@dataclass
class Result:
    records: list
    reads: int
    writes: int
    peak_buffer_records: int
    peak_live_pages: int
    history: list


class Ram:
    def __init__(self, capacity):
        self.capacity = capacity
        self.used = self.peak = 0

    def reserve(self, slots):
        self.used += slots
        assert self.used <= self.capacity
        self.peak = max(self.peak, self.used)

    def release(self, slots):
        self.used -= slots
        assert self.used >= 0


class Disk:
    def __init__(self, block_size):
        self.b = block_size
        self.pages = []
        self.free = []
        self.reads = self.writes = self.live = self.peak = 0

    def allocate(self, records):
        assert 1 <= len(records) <= self.b
        if self.free:
            page = self.free.pop()
            self.pages[page] = records
        else:
            page = len(self.pages)
            self.pages.append(records)
        self.live += 1
        self.peak = max(self.peak, self.live)
        return page

    def source(self, keys):
        ids = []
        for start in range(0, len(keys), self.b):
            ids.append(self.allocate([
                Record(keys[i], i)
                for i in range(start, min(start + self.b, len(keys)))
            ]))
        return Run(ids, len(keys))

    def read_into(self, page, destination):
        assert self.pages[page]
        self.reads += 1
        destination.extend(self.pages[page])

    def write(self, records, start, count):
        self.writes += 1
        return self.allocate(records[start:start + count])

    def erase(self, run):
        for page in run.pages:
            assert self.pages[page]
            self.pages[page] = []
            self.free.append(page)
            self.live -= 1


class Reader:
    def __init__(self, disk, run):
        self.disk, self.run = disk, run
        self.page = self.pos = 0
        self.buffer = []
        self.load()

    def load(self):
        self.buffer.clear()
        self.pos = 0
        if self.page == len(self.run.pages):
            return False
        self.disk.read_into(self.run.pages[self.page], self.buffer)
        self.page += 1
        return True

    def head(self):
        return self.buffer[self.pos]

    def advance(self):
        self.pos += 1
        return self.pos < len(self.buffer) or self.load()


def heap_sort(a):
    def sift(root, end):
        while 2 * root + 1 < end:
            child = 2 * root + 1
            if child + 1 < end and a[child] < a[child + 1]:
                child += 1
            if not a[root] < a[child]:
                return
            a[root], a[child] = a[child], a[root]
            root = child
    for root in range(len(a) // 2 - 1, -1, -1):
        sift(root, len(a))
    for end in range(len(a) - 1, 0, -1):
        a[0], a[end] = a[end], a[0]
        sift(0, end)


def merge_group(disk, group, b, ram):
    slots = (len(group) + 1) * b
    ram.reserve(slots)
    readers = [Reader(disk, run) for run in group]

    class Head:
        def __init__(self, run):
            self.run = run

        def __lt__(self, other):
            return readers[self.run].head() < readers[other.run].head()

    heap = [Head(i) for i in range(len(group))]
    heapq.heapify(heap)
    output, pages = [], []
    while heap:
        item = heapq.heappop(heap)
        reader = readers[item.run]
        output.append(reader.head())
        if len(output) == b:
            pages.append(disk.write(output, 0, len(output)))
            output.clear()
        if reader.advance():
            heapq.heappush(heap, item)
    if output:
        pages.append(disk.write(output, 0, len(output)))
    result = Run(pages, sum(run.length for run in group))
    ram.release(slots)
    for run in group:
        disk.erase(run)
    return result


def external_sort(keys, b, memory_blocks, fan_in):
    if any(type(x) is not int for x in (b, memory_blocks, fan_in)):
        raise ValueError("PARAMETER")
    if not (1 <= b <= 32 and 3 <= memory_blocks <= 32
            and 2 <= fan_in <= memory_blocks - 1):
        raise ValueError("PARAMETER")
    if len(keys) > 4096:
        raise ValueError("SIZE")
    if any(type(x) is not int or not -1000000 <= x <= 1000000 for x in keys):
        raise ValueError("VALUE")
    disk, ram = Disk(b), Ram(b * memory_blocks)
    source = disk.source(keys)
    runs, history = [], []
    for first in range(0, len(source.pages), memory_blocks):
        ids = source.pages[first:first + memory_blocks]
        length = min(b * memory_blocks, len(keys) - first * b)
        ram.reserve(length)
        chunk = []
        for page in ids:
            disk.read_into(page, chunk)
        heap_sort(chunk)
        pages = []
        for start in range(0, length, b):
            pages.append(disk.write(chunk, start, min(b, length - start)))
        runs.append(Run(pages, length))
        chunk.clear()
        ram.release(length)
    history.append(Phase("runs", [r.length for r in runs], disk.reads, disk.writes))
    disk.erase(source)
    number = 0
    while len(runs) > 1:
        before_r, before_w = disk.reads, disk.writes
        next_runs = []
        for start in range(0, len(runs), fan_in):
            group = runs[start:start + fan_in]
            next_runs.append(group[0] if len(group) == 1
                             else merge_group(disk, group, b, ram))
        runs = next_runs
        number += 1
        history.append(Phase("merge" + str(number), [r.length for r in runs],
                             disk.reads - before_r, disk.writes - before_w))
    before_r = disk.reads
    records = []
    if runs:
        ram.reserve(b)
        reader = Reader(disk, runs[0])
        while True:
            records.append(reader.head())
            if not reader.advance():
                break
        ram.release(b)
    history.append(Phase("readback", [r.length for r in runs],
                         disk.reads - before_r, 0))
    assert ram.used == 0
    return Result(records, disk.reads, disk.writes, ram.peak, disk.peak, history)


def compact(values):
    return "[" + ",".join(map(str, values)) + "]"


def main():
    keys = [9, 1, 5, 3, 5, 2, 8, 0, 7, 3, 6, 4, 2, 9, 1, 8, 0]
    b, blocks, fan_in = 2, 3, 2
    result = external_sort(keys, b, blocks, fan_in)
    print(f"B={b} memory_blocks={blocks} M={b * blocks} fan_in={fan_in}")
    for phase in result.history:
        print(f"{phase.name} lengths={compact(phase.lengths)} "
              f"reads={phase.reads} writes={phase.writes}")
    print("sorted=" + compact(f"{r.key}@{r.order}" for r in result.records))
    total = result.reads + result.writes
    print(f"reads={result.reads} writes={result.writes} transfers={total} "
          f"sort_transfers={total - result.history[-1].reads}")
    print(f"peak_buffer_records={result.peak_buffer_records} "
          f"peak_live_pages={result.peak_live_pages}")


if __name__ == "__main__":
    main()

#include <algorithm>
#include <cassert>
#include <iostream>
#include <queue>
#include <stdexcept>
#include <string>
#include <tuple>
#include <vector>
using namespace std;

struct Record {
    int key, order;
    bool operator<(const Record& r) const {
        return tie(key, order) < tie(r.key, r.order);
    }
};
struct Run { vector<int> pages; int length; };
struct Phase { string name; vector<int> lengths; int reads, writes; };
struct Result {
    vector<Record> records;
    int reads, writes, peak_buffer_records, peak_live_pages;
    vector<Phase> history;
};
struct Ram {
    int capacity, used = 0, peak = 0;
    explicit Ram(int capacity_) : capacity(capacity_) {}
    void reserve(int slots) {
        used += slots;
        assert(used <= capacity);
        peak = max(peak, used);
    }
    void release(int slots) { used -= slots; assert(used >= 0); }
};
struct Disk {
    int b, reads = 0, writes = 0, live = 0, peak = 0;
    vector<vector<Record>> pages;
    vector<int> free_pages;
    explicit Disk(int block_size) : b(block_size) {}
    int allocate(vector<Record> records) {
        assert(!records.empty() && records.size() <= static_cast<size_t>(b));
        int id;
        if (free_pages.empty()) {
            id = static_cast<int>(pages.size());
            pages.push_back(move(records));
        } else {
            id = free_pages.back(); free_pages.pop_back();
            pages[id] = move(records);
        }
        peak = max(peak, ++live);
        return id;
    }
    Run source(const vector<int>& keys) {
        Run result; result.length = static_cast<int>(keys.size());
        for (int start = 0; start < result.length; start += b) {
            vector<Record> page;
            for (int i = start; i < min(start + b, result.length); ++i)
                page.push_back(Record{keys[i], i});
            result.pages.push_back(allocate(move(page)));
        }
        return result;
    }
    void read_into(int id, vector<Record>& destination) {
        assert(!pages[id].empty());
        ++reads;
        destination.insert(destination.end(), pages[id].begin(), pages[id].end());
    }
    int write(const vector<Record>& records, int start, int count) {
        ++writes;
        return allocate(vector<Record>(records.begin() + start,
                                       records.begin() + start + count));
    }
    void erase(const Run& run) {
        for (int id : run.pages) {
            assert(!pages[id].empty());
            vector<Record>().swap(pages[id]);
            free_pages.push_back(id); --live;
        }
    }
};
struct Reader {
    Disk* disk;
    const Run* run;
    size_t page = 0, pos = 0;
    vector<Record> buffer;
    Reader(Disk& d, const Run& r) : disk(&d), run(&r) {
        buffer.reserve(d.b); load();
    }
    bool load() {
        buffer.clear(); pos = 0;
        if (page == run->pages.size()) return false;
        disk->read_into(run->pages[page++], buffer);
        return true;
    }
    const Record& head() const { return buffer[pos]; }
    bool advance() { return ++pos < buffer.size() || load(); }
};
void sift(vector<Record>& a, int root, int end) {
    while (2 * root + 1 < end) {
        int child = 2 * root + 1;
        if (child + 1 < end && a[child] < a[child + 1]) ++child;
        if (!(a[root] < a[child])) return;
        swap(a[root], a[child]); root = child;
    }
}
void heap_sort(vector<Record>& a) {
    int n = static_cast<int>(a.size());
    for (int root = n / 2 - 1; root >= 0; --root) sift(a, root, n);
    for (int end = n - 1; end > 0; --end) {
        swap(a[0], a[end]); sift(a, 0, end);
    }
}
Run merge_group(Disk& disk, const vector<Run>& group, int b, Ram& ram) {
    int slots = (static_cast<int>(group.size()) + 1) * b;
    ram.reserve(slots);
    vector<Reader> readers; readers.reserve(group.size());
    for (const Run& run : group) readers.emplace_back(disk, run);
    auto greater_head = [&readers](int a, int c) {
        return readers[c].head() < readers[a].head();
    };
    priority_queue<int, vector<int>, decltype(greater_head)> heap(greater_head);
    for (int i = 0; i < static_cast<int>(group.size()); ++i) heap.push(i);
    Run result; result.length = 0;
    vector<Record> output; output.reserve(b);
    while (!heap.empty()) {
        int id = heap.top(); heap.pop();
        output.push_back(readers[id].head());
        if (static_cast<int>(output.size()) == b) {
            result.pages.push_back(disk.write(output, 0, b)); output.clear();
        }
        if (readers[id].advance()) heap.push(id);
    }
    if (!output.empty())
        result.pages.push_back(disk.write(output, 0, static_cast<int>(output.size())));
    for (const Run& run : group) { result.length += run.length; disk.erase(run); }
    ram.release(slots);
    return result;
}
vector<int> lengths(const vector<Run>& runs) {
    vector<int> result;
    for (const Run& run : runs) result.push_back(run.length);
    return result;
}
Result external_sort(const vector<int>& keys, int b, int memory_blocks, int fan_in) {
    if (b < 1 || b > 32 || memory_blocks < 3 || memory_blocks > 32
        || fan_in < 2 || fan_in > memory_blocks - 1)
        throw invalid_argument("PARAMETER");
    if (keys.size() > 4096) throw invalid_argument("SIZE");
    for (int key : keys)
        if (key < -1000000 || key > 1000000) throw invalid_argument("VALUE");
    Disk disk(b); Ram ram(b * memory_blocks);
    Run source = disk.source(keys);
    vector<Run> runs; vector<Phase> history;
    for (int first = 0; first < static_cast<int>(source.pages.size()); first += memory_blocks) {
        int length = min(b * memory_blocks, static_cast<int>(keys.size()) - first * b);
        ram.reserve(length);
        vector<Record> chunk; chunk.reserve(length);
        for (int p = first; p < min(first + memory_blocks, static_cast<int>(source.pages.size())); ++p)
            disk.read_into(source.pages[p], chunk);
        heap_sort(chunk);
        Run run; run.length = length;
        for (int start = 0; start < length; start += b)
            run.pages.push_back(disk.write(chunk, start, min(b, length - start)));
        runs.push_back(move(run));
        ram.release(length);
    }
    history.push_back(Phase{"runs", lengths(runs), disk.reads, disk.writes});
    disk.erase(source);
    int number = 0;
    while (runs.size() > 1) {
        int before_r = disk.reads, before_w = disk.writes;
        vector<Run> next;
        for (int start = 0; start < static_cast<int>(runs.size()); start += fan_in) {
            int end = min(start + fan_in, static_cast<int>(runs.size()));
            if (end - start == 1) next.push_back(runs[start]);
            else {
                vector<Run> group(runs.begin() + start, runs.begin() + end);
                next.push_back(merge_group(disk, group, b, ram));
            }
        }
        runs = move(next);
        history.push_back(Phase{"merge" + to_string(++number), lengths(runs),
                                disk.reads - before_r, disk.writes - before_w});
    }
    int before_r = disk.reads;
    vector<Record> records;
    if (!runs.empty()) {
        ram.reserve(b);
        Reader reader(disk, runs[0]);
        do { records.push_back(reader.head()); } while (reader.advance());
        ram.release(b);
    }
    history.push_back(Phase{"readback", lengths(runs), disk.reads - before_r, 0});
    assert(ram.used == 0);
    return Result{move(records), disk.reads, disk.writes, ram.peak, disk.peak, move(history)};
}
string compact(const vector<int>& values) {
    string result = "[";
    for (size_t i = 0; i < values.size(); ++i) {
        if (i) result += ",";
        result += to_string(values[i]);
    }
    return result + "]";
}
int main() {
    vector<int> keys = {9,1,5,3,5,2,8,0,7,3,6,4,2,9,1,8,0};
    int b = 2, blocks = 3, fan_in = 2;
    Result r = external_sort(keys, b, blocks, fan_in);
    cout << "B=" << b << " memory_blocks=" << blocks << " M=" << b * blocks
         << " fan_in=" << fan_in << '\n';
    for (const Phase& p : r.history)
        cout << p.name << " lengths=" << compact(p.lengths)
             << " reads=" << p.reads << " writes=" << p.writes << '\n';
    cout << "sorted=[";
    for (size_t i = 0; i < r.records.size(); ++i) {
        if (i) cout << ',';
        cout << r.records[i].key << '@' << r.records[i].order;
    }
    int total = r.reads + r.writes;
    cout << "]\nreads=" << r.reads << " writes=" << r.writes << " transfers=" << total
         << " sort_transfers=" << total - r.history.back().reads
         << "\npeak_buffer_records=" << r.peak_buffer_records
         << " peak_live_pages=" << r.peak_live_pages << '\n';
    return 0;
}

import java.util.ArrayList;
import java.util.List;
import java.util.PriorityQueue;

public class Main {
    static final class Record implements Comparable<Record> {
        final int key, order;
        Record(int key, int order) { this.key = key; this.order = order; }
        public int compareTo(Record r) {
            int c = Integer.compare(key, r.key);
            return c != 0 ? c : Integer.compare(order, r.order);
        }
    }
    static final class Run {
        final List<Integer> pages;
        final int length;
        Run(List<Integer> pages, int length) { this.pages = pages; this.length = length; }
    }
    static final class Phase {
        final String name;
        final List<Integer> lengths;
        final int reads, writes;
        Phase(String n, List<Integer> l, int r, int w) {
            name = n; lengths = l; reads = r; writes = w;
        }
    }
    static final class Result {
        final List<Record> records;
        final int reads, writes, peak_buffer_records, peak_live_pages;
        final List<Phase> history;
        Result(List<Record> a, int r, int w, int buffer, int disk, List<Phase> h) {
            records = a; reads = r; writes = w;
            peak_buffer_records = buffer; peak_live_pages = disk; history = h;
        }
    }
    static final class Ram {
        final int capacity;
        int used, peak;
        Ram(int capacity) { this.capacity = capacity; }
        void reserve(int slots) {
            used += slots;
            if (used > capacity) throw new AssertionError("RAM budget");
            peak = Math.max(peak, used);
        }
        void release(int slots) {
            used -= slots;
            if (used < 0) throw new AssertionError("RAM release");
        }
    }
    static final class Disk {
        final int b;
        final List<List<Record>> pages = new ArrayList<>();
        final List<Integer> free = new ArrayList<>();
        int reads, writes, live, peak;
        Disk(int b) { this.b = b; }
        int allocate(List<Record> records) {
            if (records.isEmpty() || records.size() > b) throw new AssertionError("page size");
            int id;
            if (free.isEmpty()) { id = pages.size(); pages.add(records); }
            else { id = free.remove(free.size() - 1); pages.set(id, records); }
            peak = Math.max(peak, ++live);
            return id;
        }
        Run source(int[] keys) {
            List<Integer> ids = new ArrayList<>();
            for (int start = 0; start < keys.length; start += b) {
                List<Record> page = new ArrayList<>(b);
                for (int i = start; i < Math.min(start + b, keys.length); ++i)
                    page.add(new Record(keys[i], i));
                ids.add(allocate(page));
            }
            return new Run(ids, keys.length);
        }
        void readInto(int id, List<Record> destination) {
            if (pages.get(id).isEmpty()) throw new AssertionError("dead page");
            ++reads;
            // Append directly: do not allocate an extra temporary block array.
            for (Record record : pages.get(id)) destination.add(record);
        }
        int write(List<Record> records, int start, int count) {
            ++writes;
            List<Record> page = new ArrayList<>(count);
            for (int i = start; i < start + count; ++i) page.add(records.get(i));
            return allocate(page);
        }
        void erase(Run run) {
            for (int id : run.pages) {
                if (pages.get(id).isEmpty()) throw new AssertionError("dead page");
                pages.set(id, new ArrayList<>());
                free.add(id); --live;
            }
        }
    }
    static final class Reader {
        final Disk disk;
        final Run run;
        final List<Record> buffer;
        int page, pos;
        Reader(Disk disk, Run run) {
            this.disk = disk; this.run = run;
            buffer = new ArrayList<>(disk.b); load();
        }
        boolean load() {
            buffer.clear(); pos = 0;
            if (page == run.pages.size()) return false;
            disk.readInto(run.pages.get(page++), buffer);
            return true;
        }
        Record head() { return buffer.get(pos); }
        boolean advance() { return ++pos < buffer.size() || load(); }
    }
    static void swap(List<Record> a, int i, int j) {
        Record t = a.get(i); a.set(i, a.get(j)); a.set(j, t);
    }
    static void sift(List<Record> a, int root, int end) {
        while (2 * root + 1 < end) {
            int child = 2 * root + 1;
            if (child + 1 < end && a.get(child).compareTo(a.get(child + 1)) < 0) ++child;
            if (a.get(root).compareTo(a.get(child)) >= 0) return;
            swap(a, root, child); root = child;
        }
    }
    static void heapSort(List<Record> a) {
        for (int root = a.size() / 2 - 1; root >= 0; --root) sift(a, root, a.size());
        for (int end = a.size() - 1; end > 0; --end) {
            swap(a, 0, end); sift(a, 0, end);
        }
    }
    static Run mergeGroup(Disk disk, List<Run> group, int b, Ram ram) {
        int slots = (group.size() + 1) * b;
        ram.reserve(slots);
        List<Reader> readers = new ArrayList<>();
        for (Run run : group) readers.add(new Reader(disk, run));
        PriorityQueue<Integer> heap = new PriorityQueue<>((a, c) ->
            readers.get(a).head().compareTo(readers.get(c).head()));
        for (int i = 0; i < group.size(); ++i) heap.add(i);
        List<Integer> pages = new ArrayList<>();
        List<Record> output = new ArrayList<>(b);
        while (!heap.isEmpty()) {
            int id = heap.remove();
            Reader reader = readers.get(id);
            output.add(reader.head());
            if (output.size() == b) {
                pages.add(disk.write(output, 0, b)); output.clear();
            }
            if (reader.advance()) heap.add(id);
        }
        if (!output.isEmpty()) pages.add(disk.write(output, 0, output.size()));
        int length = 0;
        for (Run run : group) { length += run.length; disk.erase(run); }
        ram.release(slots);
        return new Run(pages, length);
    }
    static List<Integer> lengths(List<Run> runs) {
        List<Integer> result = new ArrayList<>();
        for (Run run : runs) result.add(run.length);
        return result;
    }
    static Result externalSort(int[] keys, int b, int memoryBlocks, int fanIn) {
        if (b < 1 || b > 32 || memoryBlocks < 3 || memoryBlocks > 32
            || fanIn < 2 || fanIn > memoryBlocks - 1)
            throw new IllegalArgumentException("PARAMETER");
        if (keys.length > 4096) throw new IllegalArgumentException("SIZE");
        for (int key : keys)
            if (key < -1000000 || key > 1000000) throw new IllegalArgumentException("VALUE");
        Disk disk = new Disk(b);
        Ram ram = new Ram(b * memoryBlocks);
        Run source = disk.source(keys);
        List<Run> runs = new ArrayList<>();
        List<Phase> history = new ArrayList<>();
        for (int first = 0; first < source.pages.size(); first += memoryBlocks) {
            int length = Math.min(b * memoryBlocks, keys.length - first * b);
            ram.reserve(length);
            List<Record> chunk = new ArrayList<>(length);
            for (int p = first; p < Math.min(first + memoryBlocks, source.pages.size()); ++p)
                disk.readInto(source.pages.get(p), chunk);
            heapSort(chunk);
            List<Integer> pages = new ArrayList<>();
            for (int start = 0; start < length; start += b)
                pages.add(disk.write(chunk, start, Math.min(b, length - start)));
            runs.add(new Run(pages, length));
            chunk.clear(); ram.release(length);
        }
        history.add(new Phase("runs", lengths(runs), disk.reads, disk.writes));
        disk.erase(source);
        int number = 0;
        while (runs.size() > 1) {
            int beforeR = disk.reads, beforeW = disk.writes;
            List<Run> next = new ArrayList<>();
            for (int start = 0; start < runs.size(); start += fanIn) {
                List<Run> group = runs.subList(start, Math.min(start + fanIn, runs.size()));
                next.add(group.size() == 1 ? group.get(0) : mergeGroup(disk, group, b, ram));
            }
            runs = next;
            history.add(new Phase("merge" + (++number), lengths(runs),
                                  disk.reads - beforeR, disk.writes - beforeW));
        }
        int beforeR = disk.reads;
        List<Record> records = new ArrayList<>();
        if (!runs.isEmpty()) {
            ram.reserve(b);
            Reader reader = new Reader(disk, runs.get(0));
            do { records.add(reader.head()); } while (reader.advance());
            ram.release(b);
        }
        history.add(new Phase("readback", lengths(runs), disk.reads - beforeR, 0));
        if (ram.used != 0) throw new AssertionError("RAM leak");
        return new Result(records, disk.reads, disk.writes, ram.peak, disk.peak, history);
    }
    static String compact(List<Integer> values) {
        StringBuilder s = new StringBuilder("[");
        for (int i = 0; i < values.size(); ++i) {
            if (i != 0) s.append(',');
            s.append(values.get(i));
        }
        return s.append(']').toString();
    }
    public static void main(String[] args) {
        int[] keys = {9,1,5,3,5,2,8,0,7,3,6,4,2,9,1,8,0};
        int b = 2, blocks = 3, fanIn = 2;
        Result r = externalSort(keys, b, blocks, fanIn);
        System.out.println("B=" + b + " memory_blocks=" + blocks + " M=" + b * blocks
                           + " fan_in=" + fanIn);
        for (Phase p : r.history)
            System.out.println(p.name + " lengths=" + compact(p.lengths)
                               + " reads=" + p.reads + " writes=" + p.writes);
        StringBuilder sorted = new StringBuilder("sorted=[");
        for (int i = 0; i < r.records.size(); ++i) {
            if (i != 0) sorted.append(',');
            Record a = r.records.get(i); sorted.append(a.key).append('@').append(a.order);
        }
        System.out.println(sorted.append(']'));
        int total = r.reads + r.writes;
        System.out.println("reads=" + r.reads + " writes=" + r.writes + " transfers=" + total
                           + " sort_transfers=" + (total - r.history.get(r.history.size() - 1).reads));
        System.out.println("peak_buffer_records=" + r.peak_buffer_records
                           + " peak_live_pages=" + r.peak_live_pages);
    }
}

Expected standard output, identical in all three languages:

• B=2 memory_blocks=3 M=6 fan_in=2
• runs lengths=[6,6,5] reads=9 writes=9
• merge1 lengths=[12,5] reads=6 writes=6
• merge2 lengths=[17] reads=9 writes=9
• readback lengths=[17] reads=9 writes=0
• sorted=[0@7,0@16,1@1,1@14,2@5,2@12,3@3,3@9,4@11,5@2,5@4,6@10,7@8,8@6,8@15,9@0,9@13]
• reads=33 writes=24 transfers=57 sort_transfers=48
• peak_buffer_records=6 peak_live_pages=18

Complexity, Correctness, and Invariants

Run counts and passes. For nonempty input, generation produces $$ R_0=\lceil N/M\rceil. $$ A pass maps $R$ runs to $\lceil R/f\rceil$ runs, including any carried singleton. Thus the number of merge passes is $h=0$ when $R_0\le1$, and otherwise $h=\lceil\log_f R_0\rceil$. Handle $N=0$ separately: it uses no transfers. The fan-in is at least two, so the pass count is finite.

Exact I/O. Generation reads and writes $D=\lceil N/B\rceil$ pages because chunk boundaries align to page boundaries. For a real merge group with run lengths $n_1,\ldots,n_g$, the exact counts are $$ \text{reads}=\sum_{j=1}^{g}\left\lceil\frac{n_j}{B}\right\rceil, \qquad \text{writes}=\left\lceil\frac{\sum_{j=1}^{g}n_j}{B}\right\rceil. $$ A singleton group contributes zero. Readback adds $D$ reads. All runs except possibly the last have page-aligned record lengths at each pass, so each full pass costs at most $2D$ transfers in this implementation. Therefore sort-only I/O is at most $2D(1+h)$, and reported I/O at most $D(3+2h)$. These are upper bounds; the worked example saves six transfers by carrying a three-page run in pass one.

Asymptotically, sort-only I/O is $O(D(1+h))$. Taking $f$ close to $M/B-1$ yields the familiar external-sort scale $O((N/B)(1+\log_{M/B}(N/M)))$ for $N\gt M$, ignoring ceilings and constant factors; use the explicit $D$, $R_0$, and $f$ formulas for small inputs and exact counts. A larger fan-in reduces passes only when it crosses a run-count threshold. The CMU sorting lecture presents bounded-memory run generation and merging; its buffer notation differs from this lesson's $B$ records per page.

CPU work. In-place heapsort generation costs $O(N\log(\min(N,M)+1))$ comparisons as an upper bound. Each heap-based merge pass costs $O(N\log f)$, giving $O(N\log(\min(N,M)+1)+Nh\log f)$ total CPU work, plus linear setup/readback and metadata traversal. In the worst case this is $O(N\log(N+1))$. The code does not detect already sorted input or exploit equal keys to skip passes. For fixed $N,B,C,f$, its I/O counts and buffer reservations are the same for best, average, and worst key orders; comparison counts may differ, but all share these CPU upper bounds. There is no assumed input distribution or claim of an exact average comparison count.

Buffer and storage space. Generation uses at most $M$ record slots. A merge reserves $(g+1)B\le M$, and readback reserves one frame when nonempty. Heap metadata has $O(f)$ indices; run and page descriptors can occupy $O(D)$ metadata in this simulator, with phase-history storage also linear in the number of descriptors across its geometrically shrinking passes. For nonempty input the peak live disk payload is exactly $2D$ pages: generation retains the $D$ source pages while writing $D$ initial-run pages. Later group writes cannot exceed the pages they replace, so they never exceed that peak. Freed payload pages are released and page slots reused. The returned observer list stores $N$ records in host memory. Metadata, object layout, spare container capacity, and garbage collection are not measured by the record-slot tracker.

Correctness of generation. The max-heap chunk sorter places the largest remaining pair at the end on each extraction. Its finished suffix is ordered and contains exactly the extracted records; at termination the entire chunk is sorted. Reading disjoint source page chunks and writing each sorted permutation preserves every original record exactly once across initial runs.

Correctness of a merge. At the start of each emission, every nonexhausted reader exposes the smallest unconsumed record of its sorted run, and the heap contains exactly those reader indices. The smallest current head is therefore the smallest record remaining anywhere in the group. Appending it preserves the sorted output prefix. Advancing only that reader restores the head invariant, and exhausted readers are omitted. Full-page writes and the final partial flush partition this prefix without omissions or duplication. Every iteration consumes one record, so the merge terminates with the group's exact sorted union.

Whole-sort argument. Generation establishes sorted, disjoint runs covering the input. Merging replaces a group by its sorted union; carrying preserves the same run. Each pass keeps exact record coverage and reduces the run count until at most one remains. Pair ordering puts equal keys in increasing original-index order, proving stability. Readback returns the final run without reordering it.

Variants, Pitfalls, and Optimizations

Partial pages and sentinels. A run's last page may contain fewer than $B$ valid records. Preserve that length and flush a nonempty output buffer after the heap empties. Padding a block with an ordinary key, especially zero, can fabricate records or corrupt order. An exhausted reader should leave the heap; do not use a legal maximum key as an infinity sentinel.

Run boundaries and carry groups. A lone last run can be carried by metadata without any transfers. Do not delete its pages when cleaning up merged inputs, and do not discard it because it lacks a partner. The configured fan-in is a maximum: the last real merge can have fewer inputs. Use actual group size when reserving frames and computing merge costs.

Stability and heap mutation. Equal-key tie handling must be consistent through chunk sorting and every merge. This code decorates each record with a unique original index; merely using an unstable chunk sort on keys would lose the guarantee. Pop a reader index before changing its head, then reinsert it. Comparator arithmetic should compare keys directly rather than subtracting them, which can overflow in a wider input domain.

Buffer the storage access. Fetching a record should usually mean advancing inside a loaded page, not issuing one disk call per key comparison. Multiple pages per run, prefetching, double buffering, and asynchronous I/O can improve throughput, but consume more memory and change the budget. The maximum feasible fan-in need not give the best latency when it leaves too little buffering per stream or exceeds file-descriptor limits.

Production choices. PostgreSQL's tuplesort source describes sorted temporary runs and balanced multiway merging, with prereading to improve locality. It can perform a final merge on demand when random access is unnecessary, avoiding a materialize-and-read cycle. This lesson intentionally materializes the final run and separately counts readback so every phase is visible; an integrated query pipeline may choose differently.

Longer initial runs. Replacement selection maintains a heap while streaming input and can produce runs averaging about $2M$ records under suitable independent random-key assumptions. That is not a worst-case guarantee: descending inputs can yield runs near $M$, while favorable order can yield much longer runs. Natural-run detection can exploit existing order. Either changes the run schedule and must be reflected in the I/O analysis.

Merge selection and scheduling. A loser tree can reduce comparison overhead for multiway merging. Unequal run lengths can motivate length-aware scheduling; historical tape-oriented polyphase schemes distribute runs differently from balanced passes. These are separate implementation choices, not permission to ignore buffer counts or valid run lengths. Parallel run generation and merging also need an aggregate memory and temporary-space budget.

Actual files and failures. A production implementation needs record serialization, stable record identifiers or equivalent tie rules, reliable page reads/writes, disk-space management, cleanup on failure, and publication of a finished output only after it is complete. Retaining originals until replacement output is complete protects against premature deletion but is not by itself a crash-recovery protocol. This simulator has no file errors, OS caching, device latency, compression, or concurrency.

Choose the operation you need. If the input fits comfortably in memory, sort it there. If a suitable ordered index already exists, scan it. If only the largest few records are required, consider a top-$k$ method. External radix/distribution sorting can suit fixed-width keys. External merge sort is particularly useful when a complete comparison-ordered batch must be produced under a buffer budget.

Practice Problems and Real-World Use

Start with three independent properties: final pairs equal an independently sorted copy, every original ID appears exactly once, and equal keys retain increasing IDs. Sortedness alone cannot detect lost records. Then derive expected transfers from run lengths rather than from the simulator's heap operations: generation costs $D$ reads and $D$ writes, a real group uses the ceiling formulas above, a singleton costs zero, and readback adds $D$ reads.

Compare configurations using the same input. Reduce fan-in while keeping $M$ fixed to isolate merge-pass effects; increase $C$ to change both chunk capacity and feasible fan-in. Test a length exactly at $M$, just over $M$, and a run count congruent to one modulo $f$ to exercise a carry. Use all-equal, increasing, decreasing, mixed-sign, and duplicate-heavy keys; fixed-length inputs should have identical transfer histories for the same parameters.

The three implementations were checked on $3633$ shared inputs: $3616$ valid cases and $17$ rejected cases. The independent oracle compares every output record, phase name, run-length vector, read/write count, buffer peak, and live-page peak. Cases include exhaustive short arrays over three keys, randomized configurations, empty and partial pages, singleton carries, maximum fan-in, length $4096$, and invalid limits. Python also passed $16$ exact-type rejection checks and $400$ direct chunk-sort comparisons. C++ ran with address and undefined-behavior sanitizers; all sample outputs match byte for byte.

For a real storage experiment, replace simulated pages with fixed-size file blocks, stream the final output to a verifier, and measure actual bytes, requests, and elapsed time separately. Vary buffer allocation and prefetch depth as well as fan-in. The abstract count predicts data movement, while device layout, caching, and concurrency determine how quickly those transfers complete.