Replacement Selection

Why Replacement Selection Matters

A file is much larger than the RAM available for sorting. You can read a fixed-size chunk, sort it, write a run, and repeat. Replacement selection offers another way to build those initial sorted runs: keep a bounded min-heap, emit its smallest eligible record, and replace the freed slot with the next input record whenever one exists.

If the new record can follow the last emitted record, it joins the current run. If it would make that run decrease, freeze it for the next run. The same storage slots serve both active and frozen records. A run can therefore grow beyond heap capacity without ever holding the whole run in memory.

This is the run-generation phase of external sorting. Its result is a sequence of individually sorted runs, not necessarily one globally sorted file. A later external multiway merge combines the runs. Longer initial runs can reduce that later work, but a benefit depends on the input order, the memory split, and whether fewer runs actually eliminate a merge pass.

You will learn the active/frozen invariant, trace every emission and replacement, implement a bounded tagged heap in Python, C++, and Java, and count actual page transfers in a storage simulator. Ordinary sorting supplies the heap idea; B-tree/B+ tree prerequisites supply the page-oriented storage model. The complete programs generate and inspect runs; they do not implement the later merge stage.

Core Concepts and Terminology

Let $N$ be the input length and $H$ the heap's maximum number of records. Let $B$ be the number of fixed-size records per simulated storage page. This lesson reserves one input frame and one output frame, each of capacity $B$, so the modeled record-buffer budget is $$ M=H+2B. $$ Here $H$ is the heap capacity; it is not the total memory budget. For a fixed available budget $M$, the heap can use at most $M-2B$ record slots in this layout. Tags, indices, container overhead, and scalar working temporaries need additional bytes in a real implementation.

Each input key becomes a record $(k,i)$, where $i$ is its original zero-based position. Records compare lexicographically by key and then original position. Write 5@0 for key five from input position zero. This makes duplicate handling deterministic and supplies a total order that a later merge can use to preserve stability.

An active record may still be emitted in the current run. A frozen record has been admitted from the input but must wait for the next run. If $x$ was just emitted and $y$ arrives, admit $y$ to the current run exactly when $y\ge x$ in this total order; otherwise freeze it. With fresh arrivals in original order, equality of keys is eligible because the new original index is larger.

The implementation uses one heap containing both populations. Its immutable priority is $$ (\text{generation},\text{key},\text{original index}). $$ Current-run records have generation $g$; frozen records have generation $g+1$. Every current-generation item outranks every frozen item, even when the frozen key is smaller. When the minimum generation changes, the active set is empty: close the old run and begin the next one. Absolute generation numbers avoid changing the heap comparator during a run transition.

Storage convention. The source is assumed already on disk, so constructing its simulated page image is uncounted. Every later page read or write costs one transfer. Readers refill only at page boundaries. Writers flush full output frames and any nonempty partial frame at every run boundary, and each run starts a fresh page. A partial page stores its valid record count rather than padding with ordinary keys. Metadata, allocation, deletion, and buffered comparisons cost no transfers in this model.

After generation, the programs separately read back every output page to return the run contents for inspection. Those reads are counted under readback. Simulated disk pages and returned observations live in host RAM outside the sorting-buffer budget; this is a transfer simulator rather than a real disk-backed sorter.

How Replacement Selection Works, Step by Step

1. Validate and initialize. The teaching API accepts $1\le H\le64$, $1\le B\le32$, at most $4096$ keys, and integer values from $-10^6$ through $10^6$. Pack the already-resident source image into pages of at most $B$ records. These finite limits keep demonstrations and counters manageable; replacement selection itself is not restricted to them.

2. Fill the initial heap. Read the first $\min(H,N)$ records through one input frame, tag them generation zero, and heapify. A page can contain a record not yet admitted to the heap; that record remains in the input frame until requested. The reader's prefetch does not make it a heap candidate early.

3. Check for a generation change. Before each emission, inspect the heap's minimum generation. If it is the next generation, all active items have been exhausted. Flush the current run's partial output page if nonempty, save the run's page list and length, clear its output metadata, and advance the current generation. The frozen records now become active through the generation change; their stored priorities do not change.

4. Emit one winner. Remove the heap minimum and append its record to the output frame. If the frame reaches $B$ records, write one page and clear the frame. Removing the winner frees one heap slot.

5. Admit one replacement. Ask the source reader for its next record. If this requires a new page, count one read. Compare the incoming record with the record just emitted. Tag it with the current generation when eligible, or the next generation when it is smaller; insert it into the freed heap slot. A frozen arrival is retained, not discarded. If the reader reports EOF, insert nothing, and the heap shrinks.

6. Continue past EOF. Repeat until the heap itself is empty. Current active items can remain after input ends, and frozen items can require another run. Finishing input is therefore different from finishing output. Close the final nonempty run and flush its last partial frame. An empty input creates no empty run and causes no transfers.

7. Observe the runs. Save generation read/write counters, release the source disk image, and use the input frame to read each generated run for display. This separate readback is optional in a real pipeline. The returned observer arrays do not count toward $M$.

Only the heap's size can grow when admitting a record, and each admission follows one removal. Hence the occupancy never exceeds its initial value $\min(H,N)$. The next figure shows a step that triggers both an output write and an input read; most emissions reuse buffered data and do neither.

flowchart TD
    A["Validate input; construct source image"] --> B["Read initial records; heapify with generation zero"]
    B --> C{"Heap empty?"}
    C -->|Yes| L["Close final nonempty run; optionally read back runs"]
    C -->|No| D{"Minimum generation changed?"}
    D -->|Yes| E["Flush and close current run; advance generation"]
    E --> F["Pop winner; emit; flush if full"]
    D -->|No| F
    F --> G{"Another source record?"}
    G -->|No| C
    G -->|Yes| H{"Arrival can follow emitted record?"}
    H -->|Yes| I["Insert with current generation"]
    H -->|No| J["Insert with next generation"]
    I --> C
    J --> C

Worked Example with a Full Trace

Use heap capacity $H=3$, page size $B=2$, and record-buffer budget $M=7$. The input keys are $$ [5,1,4,2,0,6,3,1,7,2,5,0]. $$ The six source pages P0 through P5 are [5@0,1@1], [4@2,2@3], [0@4,6@5], [3@6,1@7], [7@8,2@9], and [5@10,0@11].

Initial state. Reading P0 and P1 admits 5@0,1@1,4@2 and leaves 2@3 unread inside the input frame. Heapify them with generation zero. The active sorted view is [1@1,4@2,5@0], frozen is empty, and output is empty. Heap array layout is an implementation detail; every A/F list below is a sorted view of the eligible sets.

The full trace follows. Each step emits, flushes if full, then admits one replacement. A boundary flush occurs before the first emission of the next run. All original indices are retained so records with equal keys remain distinguishable.

1. Run 0: emit 1@1; arrival 2@3 → active. A = [2@3,4@2,5@0], F = []. Output [1@1]; use the existing input frame, with no transfer.

1. Run 0: emit 2@3; arrival 0@4 → frozen. A = [4@2,5@0], F = [0@4]. Write run 0 page [1@1,2@3]; clear output; read P2 to admit 0@4.

1. Run 0: emit 4@2; arrival 6@5 → active. A = [5@0,6@5], F = [0@4]. Output [4@2]; admit 6@5 from the existing frame, with no transfer.

1. Run 0: emit 5@0; arrival 3@6 → frozen. A = [6@5], F = [0@4,3@6]. Write run 0 page [4@2,5@0]; clear output; read P3 to admit 3@6.

1. Run 0: emit 6@5; arrival 1@7 → frozen. A = [], F = [0@4,1@7,3@6]. Output [6@5]. Active is now empty: before step 6, write this partial page and close run 0.

1. Run 1: emit 0@4; arrival 7@8 → active. A = [1@7,3@6,7@8], F = []. Begin run 1 with the promoted frozen set. Output [0@4]; read P4 to admit 7@8.

1. Run 1: emit 1@7; arrival 2@9 → active. A = [2@9,3@6,7@8], F = []. Write run 1 page [0@4,1@7]; clear output; admit 2@9 with no read.

1. Run 1: emit 2@9; arrival 5@10 → active. A = [3@6,5@10,7@8], F = []. Output [2@9]; read P5 to admit 5@10.

1. Run 1: emit 3@6; arrival 0@11 → frozen. A = [5@10,7@8], F = [0@11]. Write run 1 page [2@9,3@6]; clear output; admit 0@11 with no read.

1. Run 1: emit 5@10; EOF; insert nothing. A = [7@8], F = [0@11]. Output [5@10]; input reports EOF without a transfer.

1. Run 1: emit 7@8; EOF; insert nothing. A = [], F = [0@11]. Write run 1 page [5@10,7@8]; output is empty; input still reports EOF. Close run 1 before step 12; no extra flush is needed.

1. Run 2: emit 0@11; EOF; insert nothing. A = [], F = []. Begin run 2 with the last frozen record. Output [0@11]; EOF supplies no replacement. The heap is empty: write this partial page and close run 2.

All generated runs and their physical pages. Run 0 is [1@1,2@3]; [4@2,5@0]; [6@5]. Run 1 is [0@4,1@7]; [2@9,3@6]; [5@10,7@8]. Run 2 is [0@11]. The lengths are $[5,6,1]$, with page counts $[3,3,1]$. Every original ID from zero through eleven appears exactly once.

Concatenating the runs gives keys $[1,2,4,5,6,0,1,2,3,5,7,0]$, which is not globally sorted. A subsequent stable merge by $(k,i)$ would give [0@4,0@11,1@1,1@7,2@3,2@9,3@6,4@2,5@0,5@10,6@5,7@8]. That merge is an illustration of the next phase; its I/O is not included in this program's counters.

Exact transfers. Initial filling reads P0 and P1. Steps 2, 4, 6, and 8 read P2 through P5, for six generation reads total. Run 0 writes three pages, run 1 writes three, and run 2 writes one: seven generation writes. Generation therefore costs $6+7=13$ transfers. Readback adds seven reads, so reported totals are thirteen reads, seven writes, and twenty transfers.

The output needs one more page than the source because each run starts on a fresh page. The partial end of run 0 cannot be filled with run 1's first record under this layout. Peak heap occupancy is three; reserving two $B$-slot I/O frames gives seven record-buffer slots. Keeping all six source pages until generation finishes yields a peak of $6+7=13$ live disk pages.

Implementation in Python, C++, and Java

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 replacement-selection. Save Java as Main.java and run javac Main.java followed by java Main. Python 3.7 or newer supplies dataclasses; C++ requires C++17 for optional; Java uses ordinary classes rather than Java record syntax.

The functions are replacement_selection(keys, h, b) in Python/C++ and replacementSelection(keys, h, b) in Java. Supply a Python list, C++ vector<int>, or Java int[]; null/nonsequence inputs are outside the contract. Keys are not mutated. Python rejects booleans and floats by checking exact integer types; C++ and Java have integer-typed parameters. Parameters outside the stated limits raise PARAMETER, more than $4096$ keys raises SIZE, and out-of-range keys raise VALUE. Parameters are checked even for empty input; size is checked before key values.

How the pieces map to the model. Disk owns simulated source/output pages and increments counters only in its read/write methods. Reader keeps one input page and returns one record at a time; resetting it for readback reuses that frame. Item provides immutable generation/key/index priority. Python/C++ use local flush/finish helpers, while Java groups those helpers in Writer. All three perform the same emissions, classifications, page writes, and run transitions.

The result returns observed run records, total reads and writes, generation-only reads and writes, peak heap occupancy, peak reserved record-buffer slots, and peak live disk pages. generation_transfers excludes readback. For nonempty input, peak_buffer_records is heap_peak + 2*b: heap occupancy peaks during initialization, and both I/O frames reserve full capacity. It measures logical record slots, not bytes, object allocations, spare container capacity, metadata, or scalar record temporaries.

Simulation boundary. Input keys, simulated disk pages, run/page descriptors, and returned observer arrays all occupy host RAM. Overall host storage is $O(N+H+B)$ including these structures. The modeled sorting buffers fit $H+2B$ slots; the whole program is not constrained to that many host records. A real implementation would stream source records from files, serialize output pages, keep run metadata, and stream verification instead of returning all records. Writing a page into simulated disk storage represents the disk destination, not another sorting frame.

The generation loop maintains at most $H$ heap items and reads/writes only through buffered page methods. It never sorts all input keys as a shortcut. The displayed runs are reconstructed using counted output-page reads, so inspection has an explicit cost.

from dataclasses import dataclass
import heapq


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


@dataclass(frozen=True, order=True)
class Item:
    generation: int
    record: Record


@dataclass
class Run:
    pages: list
    length: int


@dataclass
class Result:
    runs: list
    reads: int
    writes: int
    generation_reads: int
    generation_writes: int
    heap_peak: int
    peak_buffer_records: int
    peak_live_pages: int


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

    def allocate(self, records):
        assert 1 <= len(records) <= self.b
        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 not destination and self.pages[page]
        self.reads += 1
        destination.extend(self.pages[page])

    def write(self, records):
        self.writes += 1
        return self.allocate(records.copy())

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


class Reader:
    def __init__(self, disk, run):
        self.disk = disk
        self.buffer = []
        self.reset(run)

    def reset(self, run):
        self.run = run
        self.page = self.pos = 0
        self.buffer.clear()

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


def replacement_selection(keys, h, b):
    if type(h) is not int or type(b) is not int:
        raise ValueError("PARAMETER")
    if not (1 <= h <= 64 and 1 <= b <= 32):
        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 = Disk(b)
    source = disk.source(keys)  # Source is assumed already on disk.
    reader = Reader(disk, source)
    heap = []
    for _ in range(min(h, len(keys))):
        heap.append(Item(0, reader.next_record()))
    heapq.heapify(heap)
    heap_peak = len(heap)
    generation = 0
    runs = []
    pages, output = [], []
    length = 0

    def flush():
        if output:
            pages.append(disk.write(output))
            output.clear()

    def finish():
        flush()
        runs.append(Run(pages.copy(), length))
        pages.clear()

    while heap:
        if heap[0].generation != generation:
            assert heap[0].generation == generation + 1
            finish()
            generation += 1
            length = 0
        item = heapq.heappop(heap)
        output.append(item.record)
        length += 1
        if len(output) == b:
            flush()
        incoming = reader.next_record()
        if incoming is not None:
            next_generation = generation + int(incoming < item.record)
            heapq.heappush(heap, Item(next_generation, incoming))
        assert len(heap) <= h
    if length:
        finish()
    generation_reads, generation_writes = disk.reads, disk.writes
    disk.erase(source)
    reader.buffer.clear()
    observed = []
    for run in runs:
        reader.reset(run)
        records = []  # Observer output is outside the sorting-buffer budget.
        while True:
            record = reader.next_record()
            if record is None:
                break
            records.append(record)
        assert len(records) == run.length
        observed.append(records)
    peak = heap_peak + 2 * b if keys else 0
    assert peak <= h + 2 * b
    return Result(observed, disk.reads, disk.writes,
                  generation_reads, generation_writes,
                  heap_peak, peak, disk.peak)


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


def main():
    keys = [5, 1, 4, 2, 0, 6, 3, 1, 7, 2, 5, 0]
    h, b = 3, 2
    result = replacement_selection(keys, h, b)
    print(f"H={h} B={b} M={h + 2 * b}")
    print("run_lengths=" + compact(len(run) for run in result.runs))
    for i, run in enumerate(result.runs):
        print(f"run{i}=" + compact(f"{a.key}@{a.order}" for a in run))
    print(f"generate reads={result.generation_reads} writes={result.generation_writes}")
    print(f"readback reads={result.reads - result.generation_reads} writes=0")
    print(f"reads={result.reads} writes={result.writes} "
          f"transfers={result.reads + result.writes} "
          f"generation_transfers={result.generation_reads + result.generation_writes}")
    print(f"heap_peak={result.heap_peak} 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 <optional>
#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 Item { int generation; Record record; };
struct Later {
    bool operator()(const Item& a, const Item& c) const {
        return tie(a.generation, a.record.key, a.record.order)
             > tie(c.generation, c.record.key, c.record.order);
    }
};
struct Run { vector<int> pages; int length; };
struct Result {
    vector<vector<Record>> runs;
    int reads, writes, generation_reads, generation_writes;
    int heap_peak, peak_buffer_records, peak_live_pages;
};
struct Disk {
    int b, reads = 0, writes = 0, live = 0, peak = 0;
    vector<vector<Record>> pages;
    explicit Disk(int b_) : b(b_) {}
    int allocate(vector<Record> records) {
        assert(!records.empty() && records.size() <= static_cast<size_t>(b));
        int id = static_cast<int>(pages.size());
        pages.push_back(move(records));
        peak = max(peak, ++live);
        return id;
    }
    Run source(const vector<int>& keys) {
        Run run; run.length = static_cast<int>(keys.size());
        for (int start = 0; start < run.length; start += b) {
            vector<Record> page;
            for (int i = start; i < min(start + b, run.length); ++i)
                page.push_back(Record{keys[i], i});
            run.pages.push_back(allocate(move(page)));
        }
        return run;
    }
    void read_into(int id, vector<Record>& destination) {
        assert(destination.empty() && !pages[id].empty());
        ++reads;
        destination.insert(destination.end(), pages[id].begin(), pages[id].end());
    }
    int write(const vector<Record>& records) {
        ++writes;
        return allocate(records);
    }
    void erase(const Run& run) {
        for (int id : run.pages) {
            assert(!pages[id].empty());
            vector<Record>().swap(pages[id]); --live;
        }
    }
};
struct Reader {
    Disk& disk;
    const Run* run = nullptr;
    size_t page = 0, pos = 0;
    vector<Record> buffer;
    Reader(Disk& d, const Run& r) : disk(d) { buffer.reserve(d.b); reset(r); }
    void reset(const Run& r) {
        run = &r; page = pos = 0; buffer.clear();
    }
    optional<Record> next_record() {
        if (pos == buffer.size()) {
            buffer.clear(); pos = 0;
            if (page == run->pages.size()) return nullopt;
            disk.read_into(run->pages[page++], buffer);
        }
        return buffer[pos++];
    }
};
Result replacement_selection(const vector<int>& keys, int h, int b) {
    if (h < 1 || h > 64 || b < 1 || b > 32) 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);
    Run source = disk.source(keys);
    Reader reader(disk, source);
    vector<Item> initial;
    for (int i = 0; i < min(h, static_cast<int>(keys.size())); ++i)
        initial.push_back(Item{0, *reader.next_record()});
    int heap_peak = static_cast<int>(initial.size());
    priority_queue<Item, vector<Item>, Later> heap(Later{}, move(initial));
    int generation = 0, length = 0;
    vector<Run> runs;
    vector<int> pages;
    vector<Record> output; output.reserve(b);
    auto flush = [&]() {
        if (!output.empty()) {
            pages.push_back(disk.write(output)); output.clear();
        }
    };
    auto finish = [&]() {
        flush(); runs.push_back(Run{pages, length}); pages.clear();
    };
    while (!heap.empty()) {
        if (heap.top().generation != generation) {
            assert(heap.top().generation == generation + 1);
            finish(); ++generation; length = 0;
        }
        Item item = heap.top(); heap.pop();
        output.push_back(item.record); ++length;
        if (static_cast<int>(output.size()) == b) flush();
        auto incoming = reader.next_record();
        if (incoming) {
            int next_generation = generation + (*incoming < item.record ? 1 : 0);
            heap.push(Item{next_generation, *incoming});
        }
        assert(heap.size() <= static_cast<size_t>(h));
    }
    if (length) finish();
    int generation_reads = disk.reads, generation_writes = disk.writes;
    disk.erase(source); reader.buffer.clear();
    vector<vector<Record>> observed;
    for (const Run& run : runs) {
        reader.reset(run);
        vector<Record> records;
        while (auto record = reader.next_record()) records.push_back(*record);
        assert(static_cast<int>(records.size()) == run.length);
        observed.push_back(move(records));
    }
    int peak = keys.empty() ? 0 : heap_peak + 2 * b;
    assert(peak <= h + 2 * b);
    return Result{move(observed), disk.reads, disk.writes,
                  generation_reads, generation_writes, heap_peak, peak, disk.peak};
}
string compact(const vector<int>& values) {
    string s = "[";
    for (size_t i = 0; i < values.size(); ++i) {
        if (i) s += ',';
        s += to_string(values[i]);
    }
    return s + "]";
}
int main() {
    vector<int> keys = {5,1,4,2,0,6,3,1,7,2,5,0};
    int h = 3, b = 2;
    Result r = replacement_selection(keys, h, b);
    cout << "H=" << h << " B=" << b << " M=" << h + 2 * b << '\n';
    vector<int> lengths;
    for (const auto& run : r.runs) lengths.push_back(static_cast<int>(run.size()));
    cout << "run_lengths=" << compact(lengths) << '\n';
    for (size_t i = 0; i < r.runs.size(); ++i) {
        cout << "run" << i << "=[";
        for (size_t j = 0; j < r.runs[i].size(); ++j) {
            if (j) cout << ',';
            cout << r.runs[i][j].key << '@' << r.runs[i][j].order;
        }
        cout << "]\n";
    }
    cout << "generate reads=" << r.generation_reads << " writes=" << r.generation_writes
         << "\nreadback reads=" << r.reads - r.generation_reads << " writes=0\n"
         << "reads=" << r.reads << " writes=" << r.writes
         << " transfers=" << r.reads + r.writes
         << " generation_transfers=" << r.generation_reads + r.generation_writes
         << "\nheap_peak=" << r.heap_peak << " peak_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 Item implements Comparable<Item> {
        final int generation;
        final Record record;
        Item(int generation, Record record) { this.generation = generation; this.record = record; }
        public int compareTo(Item other) {
            int c = Integer.compare(generation, other.generation);
            return c != 0 ? c : record.compareTo(other.record);
        }
    }
    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 Result {
        final List<List<Record>> runs;
        final int reads, writes, generation_reads, generation_writes;
        final int heap_peak, peak_buffer_records, peak_live_pages;
        Result(List<List<Record>> runs, int r, int w, int gr, int gw, int hp, int bp, int dp) {
            this.runs = runs; reads = r; writes = w;
            generation_reads = gr; generation_writes = gw;
            heap_peak = hp; peak_buffer_records = bp; peak_live_pages = dp;
        }
    }
    static final class Disk {
        final int b;
        final List<List<Record>> pages = 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 = pages.size(); pages.add(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 (!destination.isEmpty() || pages.get(id).isEmpty()) throw new AssertionError("read");
            ++reads;
            for (Record record : pages.get(id)) destination.add(record);
        }
        int write(List<Record> records) {
            ++writes;
            List<Record> page = new ArrayList<>(records.size());
            for (Record record : records) page.add(record);
            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<>()); --live;
            }
        }
    }
    static final class Reader {
        final Disk disk;
        final List<Record> buffer;
        Run run;
        int page, pos;
        Reader(Disk disk, Run run) {
            this.disk = disk; buffer = new ArrayList<>(disk.b); reset(run);
        }
        void reset(Run run) { this.run = run; page = pos = 0; buffer.clear(); }
        Record nextRecord() {
            if (pos == buffer.size()) {
                buffer.clear(); pos = 0;
                if (page == run.pages.size()) return null;
                disk.readInto(run.pages.get(page++), buffer);
            }
            return buffer.get(pos++);
        }
    }
    static final class Writer {
        final Disk disk;
        final List<Record> output;
        final List<Integer> pages = new ArrayList<>();
        final List<Run> runs = new ArrayList<>();
        int length;
        Writer(Disk disk) { this.disk = disk; output = new ArrayList<>(disk.b); }
        void flush() {
            if (!output.isEmpty()) { pages.add(disk.write(output)); output.clear(); }
        }
        void emit(Record record) {
            output.add(record); ++length;
            if (output.size() == disk.b) flush();
        }
        void finish() {
            flush(); runs.add(new Run(new ArrayList<>(pages), length));
            pages.clear(); length = 0;
        }
    }
    static Result replacementSelection(int[] keys, int h, int b) {
        if (h < 1 || h > 64 || b < 1 || b > 32) 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);
        Run source = disk.source(keys);
        Reader reader = new Reader(disk, source);
        List<Item> initial = new ArrayList<>();
        for (int i = 0; i < Math.min(h, keys.length); ++i)
            initial.add(new Item(0, reader.nextRecord()));
        PriorityQueue<Item> heap = new PriorityQueue<>(initial);
        int heapPeak = heap.size();
        initial.clear();
        int generation = 0;
        Writer writer = new Writer(disk);
        while (!heap.isEmpty()) {
            if (heap.peek().generation != generation) {
                if (heap.peek().generation != generation + 1) throw new AssertionError("generation");
                writer.finish(); ++generation;
            }
            Item item = heap.remove();
            writer.emit(item.record);
            Record incoming = reader.nextRecord();
            if (incoming != null) {
                int nextGeneration = generation + (incoming.compareTo(item.record) < 0 ? 1 : 0);
                heap.add(new Item(nextGeneration, incoming));
            }
            if (heap.size() > h) throw new AssertionError("heap budget");
        }
        if (writer.length != 0) writer.finish();
        int gr = disk.reads, gw = disk.writes;
        disk.erase(source); reader.buffer.clear();
        List<List<Record>> observed = new ArrayList<>();
        for (Run run : writer.runs) {
            reader.reset(run);
            List<Record> records = new ArrayList<>();
            Record record;
            while ((record = reader.nextRecord()) != null) records.add(record);
            if (records.size() != run.length) throw new AssertionError("run length");
            observed.add(records);
        }
        int peak = keys.length == 0 ? 0 : heapPeak + 2 * b;
        if (peak > h + 2 * b) throw new AssertionError("buffer budget");
        return new Result(observed, disk.reads, disk.writes, gr, gw, heapPeak, peak, disk.peak);
    }
    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 = {5,1,4,2,0,6,3,1,7,2,5,0};
        int h = 3, b = 2;
        Result r = replacementSelection(keys, h, b);
        System.out.println("H=" + h + " B=" + b + " M=" + (h + 2 * b));
        List<Integer> lengths = new ArrayList<>();
        for (List<Record> run : r.runs) lengths.add(run.size());
        System.out.println("run_lengths=" + compact(lengths));
        for (int i = 0; i < r.runs.size(); ++i) {
            StringBuilder s = new StringBuilder("run" + i + "=[");
            List<Record> run = r.runs.get(i);
            for (int j = 0; j < run.size(); ++j) {
                if (j != 0) s.append(',');
                Record a = run.get(j); s.append(a.key).append('@').append(a.order);
            }
            System.out.println(s.append(']'));
        }
        System.out.println("generate reads=" + r.generation_reads + " writes=" + r.generation_writes);
        System.out.println("readback reads=" + (r.reads - r.generation_reads) + " writes=0");
        System.out.println("reads=" + r.reads + " writes=" + r.writes
                           + " transfers=" + (r.reads + r.writes)
                           + " generation_transfers=" + (r.generation_reads + r.generation_writes));
        System.out.println("heap_peak=" + r.heap_peak + " peak_buffer_records=" + r.peak_buffer_records
                           + " peak_live_pages=" + r.peak_live_pages);
    }
}

Expected standard output, identical in all three languages:

• H=3 B=2 M=7
• run_lengths=[5,6,1]
• run0=[1@1,2@3,4@2,5@0,6@5]
• run1=[0@4,1@7,2@9,3@6,5@10,7@8]
• run2=[0@11]
• generate reads=6 writes=7
• readback reads=7 writes=0
• reads=13 writes=7 transfers=20 generation_transfers=13
• heap_peak=3 peak_buffer_records=7 peak_live_pages=13

Complexity, Correctness, and Invariants

Output contract and run-count bounds. Let the nonempty generated runs have lengths $n_1,\ldots,n_R$, summing to $N$. Every run except possibly the last contains at least $H$ records. Before EOF, a run starts with a full heap: each emission gets one replacement, and any un-emitted records are retained. All records active at a run's start must be emitted in that run. Once fewer than $H$ records remain at a new run's start, EOF has already occurred and no later replacement can create another run. Hence, for $N\gt0$, $$ 1\le R\le\left\lceil\frac{N}{H}\right\rceil. $$ If $N\le H$, only one run is needed. Empty input has $R=0$ and zero transfers.

Best, typical, and worst ordering. Nondecreasing keys produce one run of length $N$, including the all-equal case because equal arrivals are eligible. Strictly descending input produces runs of $H$ records, except a possibly shorter final run, attaining the upper bound. For a long stream of independent identically distributed keys from a continuous distribution, the classic steady-state run length is about $2H$. This is a distributional expectation, not a guaranteed run length or an exact mean for every finite file. Startup, the last short run, duplicates, and correlated input can change the observed average. The worked lengths $[5,6,1]$ do not imply a six-record guarantee.

The snowplow intuition is that records arriving ahead of a moving ordering threshold can join the current run, while arrivals behind it wait for the next sweep. In steady random input the current population is replenished while it is drained, allowing roughly twice the heap occupancy to be emitted per sweep. OpenDSA's external-sorting lesson develops this analogy. Here its heap-array capacity is called $H$, and the I/O frames are counted separately in $M$.

CPU cost. Put $K=\min(H,N)$. Initial heap construction is $O(K)$. There are exactly $N$ removals and $N-K$ later insertions, each taking at most $O(\log(K+1))$ time. Generation therefore has CPU upper bound $$ O\bigl(N\log(K+1)+K\bigr), $$ with linear page copying and readback in addition. For $H=1$ the heap work is constant per record. The same CPU upper bound applies to best, average, and worst key orders; the heap library does not skip its priority-queue work merely because input is sorted. It is a worst-case comparison bound, not an exact average-time formula. The number and lengths of runs are the input-sensitive quantities of greatest interest here.

Exact page transfers. Let $$ D=\left\lceil\frac{N}{B}\right\rceil, \qquad W=\sum_{j=1}^{R}\left\lceil\frac{n_j}{B}\right\rceil. $$ Generation reads each source page once and writes each output page once: $D$ reads plus $W$ writes. Readback adds $W$ reads. Thus the reported reads are $D+W$, writes are $W$, generation transfers are $D+W$, and reported transfers are $D+2W$. For nonempty input, $$ D\le W\le D+R-1,\qquad W\ge R. $$ The extra pages come from independently rounded run ends. Generation I/O is therefore $\Theta(D+R)$ under this layout; when $H\ge B$, the run-count bound makes it $\Theta(D)$. For $H\lt B$, many short runs can make fragmentation significant. Do not substitute $D$ for $W$ without checking run lengths or changing the storage layout.

For one ascending run, $W=D$. For strictly descending keys, write $N=qH+t$ with $0\le t\lt H$: then $$ W=q\left\lceil\frac{H}{B}\right\rceil+\left\lceil\frac{t}{B}\right\rceil. $$ The final term is zero when $t=0$. In the random-stream setting, $R\approx N/(2H)$ is only a rough run-count estimate; use measured $n_j$ to determine exact page writes. The sample's $D=6$ and $W=7$ give thirteen generation transfers and twenty including inspection.

Space. Active and frozen records share at most $H$ heap slots. Two I/O frames require $2B$ slots, so sorting record-buffer space is $O(H+B)$ and fits $M=H+2B$. For nonempty input, reported peak reservation is $\min(H,N)+2B$. Tags need $O(H)$ small metadata entries; source/output page descriptors and run descriptors add $O(D+W+R)$ metadata in the simulator. The source is retained until all output runs exist, so peak live disk pages are exactly $D+W$, not necessarily $2D$. Host storage also includes simulated disk payloads and observer output.

Correctness. Initially every admitted record is active, and the heap minimum begins a sorted run. Suppose the current prefix ends at $x$. Every other active item is at least $x$ because $x$ was the active minimum. A new arrival at least $x$ is safe to admit; a smaller arrival receives the next-generation tag. The heap's generation-first priority ensures frozen items cannot be emitted early. Thus every next emission preserves nondecreasing order in the current run.

When the minimum tag becomes $g+1$, no generation-$g$ record remains. Closing that run and advancing $g$ makes all waiting items eligible for a fresh prefix with no inherited threshold. Every source record is admitted exactly once and removed exactly once, and full/partial flushes preserve all emitted records. After EOF, the heap decreases on every iteration, so all active and frozen records are eventually drained. The resulting runs are a disjoint, individually sorted partition of the input. Their total pair order supports stable merging later; concatenation alone need not be sorted.

Effect on a full external sort. If a later merge supports fan-in $f\ge2$, $R$ runs require at most $\lceil\log_f R\rceil$ balanced merge passes for $R\gt1$. Heap space can be repurposed for merge input frames, so the merge budget must be calculated for that phase. Fewer generated runs eliminate a pass only when they cross an integer fan-in threshold. This lesson's transfer counters stop before merging.

Variants, Pitfalls, and Optimizations

Use the emitted record as the threshold. Compare an arrival with the record just output, not with the new heap root. After emitting key two, an arriving key three is eligible even if the remaining old minimum is four. Freezing it against four would unnecessarily shorten the run. Conversely, allowing a key smaller than the emitted key into the current generation can make the run decrease.

Frozen is a priority rule. A key-only heap will immediately prefer a frozen small key and break ordering. Two conceptual sets are useful for explanations, but their total record count must stay within $H$. A single tagged heap shares the capacity directly. Another classic implementation stores the active heap in the front of one array and frozen records in its shrinking suffix. At promotion it rebuilds a heap; it does not allocate a second full heap of records.

Keep comparison priorities immutable. This code orders by absolute generation before key and original index. Changing a global comparator interpretation while elements remain in a priority queue can invalidate heap order. Designs using alternating marks need a correct proof of priority at each transition and any necessary rebuilding. Absolute tags avoid that dependency in this teaching implementation.

EOF is not heap exhaustion. Stop requesting replacements when input ends, but continue emitting until no active or frozen item remains. The sample has 0@11 waiting for run 2 after EOF. Stopping at the last input read would lose several buffered records and possibly an entire run.

Flush every run boundary. If the output frame is partly full, write its valid records before beginning the next run. Otherwise a reader that assumes whole pages belong to one run can merge different runs accidentally. More compact formats can pack run ends into shared pages, but then descriptors must include record offsets and a different page/caching accounting rule. The formulas here assume fresh-page run starts.

Equality and stability. Use nondecreasing eligibility, not a strict-greater test, when equal keys are allowed in one run. With original-index tie breaking, a later equal-key arrival is always greater in the total record order. Retain that index, or an equivalent consistent stable tie policy, when merging the runs; stable ordering inside each run alone does not establish global stability without a merge rule.

Compare memory allocations fairly. The approximate $2H$ run-length advantage uses $H$ heap records. Under the total slot budget $M$, this version spends $2B$ on I/O frames first. A fixed-chunk sorter can allocate/reuse its buffers differently and may devote more of $M$ to each chunk. Heap operations and scattered accesses also have CPU/cache costs. Replacement selection is a useful option, not a guarantee of the fastest external sort.

Modern system choices. PostgreSQL's tuplesort source notes its historical use of replacement selection and its move to batch run generation using quicksort or radix sort. That is a concrete reason to distinguish the classical algorithm's teaching value from a claim that every current database uses it. Benchmark complete sorting, including merge passes and buffering, on the relevant workload.

Variants and research. Tournament or loser-tree structures can replace a binary heap. Direction-aware generators may make ascending and descending runs to exploit structure in the stream, but the merge must know each run's direction and storage layout. Run Generation Revisited studies online/offline run generation, alternating directions, and extra memory/lookahead. Those policy-specific guarantees do not automatically apply to the ascending-only code here.

From simulator to files. Replace in-memory disk pages with sequential page I/O, serialize only valid records, preserve run lengths/offsets and stable identifiers, and check short reads/writes. A completed-run manifest and cleanup strategy are needed if file creation fails. Prefetching, double buffering, or compression can help throughput but changes the byte budget and transfer interpretation. Do not infer elapsed time directly from the simulator's transfer count.

Practice Problems and Real-World Use

Verify the output as a collection of runs before checking performance: each run must be sorted by $(k,i)$, every input ID must occur exactly once, and the total record count must remain $N$. Compare the exact run partition with an independent implementation using unsorted active/frozen lists and linear minimum selection. This catches generation-tag mistakes without repeating the heap implementation.

Then compute expected I/O independently from the returned run lengths. Use $D=\lceil N/B\rceil$ source pages and $W=\sum_j\lceil n_j/B\rceil$ output pages, check generation $D+W$, readback $W$, and peak live pages $D+W$. Changing $B$ while holding $H$ and the key stream fixed should change transfers but preserve the logical run contents. Changing $H$ changes the selection window and can change the run partition.

All three implementations agree on $7555$ shared inputs: $7539$ valid cases and $16$ rejected cases. The oracle compares every record in every run and every reported counter. Tests cover exhaustive short arrays over three keys, all small distinct permutations, randomized inputs and buffer sizes, duplicate-heavy/sorted/reverse inputs, empty input, partial pages, EOF with frozen records, maximum size $4096$, and invalid limits. Python also passed $12$ exact-type rejection checks and $400$ page-size invariance comparisons. C++ ran with address and undefined-behavior sanitizers, and sample stdout is identical in every language.

For a realistic experiment, fix a byte budget, subtract I/O buffers and metadata, then determine the remaining heap capacity from the serialized record size. Measure generated run lengths, output-page fragmentation, later merge-pass count, CPU time, and total I/O. For random data, use several seeds and a long stream, report how startup/tail runs are treated, and compare the observed distribution with the approximate steady-state expectation rather than asserting every run has length $2H$.