K-Way Merge

Why K-Way Merge Matters

Several files are already sorted: perhaps temporary runs from an external sort, log streams from separate services, or ordered batches from storage. Concatenating them does not usually produce sorted output. Sorting their union again works, but discards useful information. K-way merge combines these sorted inputs by repeatedly selecting the smallest current head.

A min-heap maintains one candidate per nonempty input. When a candidate wins, only its own source advances. The replacement head returns to the heap, and the process continues until all sources are exhausted. The output preserves duplicates and every input record; deduplication is a separate operation.

This lesson connects that frontier idea to storage pages. The Python, C++, and Java programs merge explicitly buffered simulated disk runs, resolve equal keys deterministically, reject unsorted input, enforce a frame budget, and print identical block-transfer counts. A complete trace shows empty runs, ties, refills, exhaustion, and the final partial output page.

Ordinary sorting supplies comparisons and heaps. B-tree/B+ tree prerequisites supply pages, buffers, and ordered scans. K-way merge is the combining phase that follows run generation; it assumes each source is already sorted under a compatible order. If all data is unsorted, first generate runs or use another full sorting method.

Core Concepts and Terminology

Let $k$ be the number of supplied runs, $n_i$ the length of run $i$, and $N=\sum_i n_i$ the total record count. Let $q$ be the number of nonempty runs. Empty runs count toward $k$ but need no heap entry, disk page, or input frame. If $q=0$, the output is empty.

The output is the sorted multiset union: all $N$ records appear exactly once, including duplicates. To make ties deterministic, attach source identity and compare $$ (\text{key},\text{source run},\text{position within source}). $$ Write 4@2:1 for key four at position one of source run two. Equal keys come from lower source-run indices first, with within-run order preserved. This is stable with respect to the concatenation of the supplied runs. If runs came from interleaved original records, such as replacement selection, retain and compare a true original global index to obtain stability relative to that earlier stream; source-run order alone does not supply it.

The heap contains reader indices, not all remaining records. Each index compares its reader's current buffered head. Only the winning reader changes, and its index is removed before that change, then reinserted if it still has data. No heap comparator performs disk I/O.

For the storage model, $B$ is records per page, $C$ is available record-buffer frames, and $M=BC$ is the record-slot budget. A nonempty merge reserves $q$ input frames and one output frame: $$ (q+1)B\le M,\qquad q\le C-1. $$ Heap indices, page IDs, run descriptors, scalar record temporaries, and runtime/container overhead are outside this record-buffer count. A real byte budget must include them. The code rejects a one-pass merge that does not fit; a multipass external sort would instead group inputs into feasible merges.

Transfer convention. Input runs begin on separate simulated pages, each holding at most $B$ valid records. Their initial source image is assumed already on disk and is constructed without counting writes. Every later page read and write costs one transfer; partial pages still cost one. Allocation, deletion, validation of supplied host arrays, and buffered comparisons are outside the simulated I/O count.

The merge materializes one packed output run. A separately counted readback phase reads that output for display. Returned observer records and simulated disk payloads occupy host memory outside the merging-frame budget. The program demonstrates I/O behavior rather than operating on real files under a hard RAM cap.

How K-Way Merge Works, Step by Step

1. Check parameters and sorted-input preconditions. The teaching API accepts at most $32$ runs and $4096$ total integer keys, with each key between $-10^6$ and $10^6$. Require $1\le B\le32$ and $2\le C\le33$. Verify that each host input array is nondecreasing. Count nonempty sources and require $q+1\le C$ when any records exist. These caps bound demonstrations, not the general algorithm.

2. Create source pages. Pack each run separately into pages, preserving its source index even when earlier runs are empty. Pages never mix different source runs. Source construction represents data already resident on disk and is uncounted.

3. Initialize readers and candidates. Create one reader per nonempty run, reserve one input frame for each, and read its first page. Insert its reader index into the heap using the buffered record comparison. These initial $q$ page reads are included in merge I/O. Python/C++ heapify the initial index collection; Java inserts initial indices into its priority queue. Both fit the final complexity bound.

4. Emit the minimum. Pop one winning reader index, append its head to the output frame, and write/clear that frame if it reaches $B$ records. The emitted record is globally next under the sorted-input assumption and tie rule.

5. Advance exactly one reader. If more records remain in its current page, move its position without any read. Otherwise clear the frame and load the next page if one exists, counting one read. Reinsert the index only when the reader has a valid next head. Exhausted readers leave the heap, so no sentinel value is needed.

6. Finish and observe. When the heap empties, flush any nonempty partial output frame. The merge helper returns, releasing input-frame storage, and source pages are deleted after the complete output exists. Read back the output through one input frame for display, with its reads counted separately. If every run is empty, no heap item is created, no page is read/written, and all counters remain zero.

Do not advance all readers after each output: that would skip records. Do not mutate a reader's head while its index remains in the heap: that can invalidate priority-queue order. Pop, emit, advance, and reinsert keep changes local and auditable.

flowchart TD
    A["Validate parameters, sorted runs, and frame budget"] --> B["Construct source images; skip empty runs"]
    B --> C["Read initial pages and heap their reader indices"]
    C --> D{"Heap empty?"}
    D -->|No| E["Pop minimum reader; emit its head"]
    E --> F{"Output frame full?"}
    F -->|Yes| G["Write output page and clear frame"]
    F -->|No| H["Advance only the winning reader"]
    G --> H
    H --> I{"Input page exhausted?"}
    I -->|Yes| J["Refill from next page if available"]
    I -->|No| K{"Reader still has a head?"}
    J --> K
    K -->|Yes| L["Reinsert reader index"]
    K -->|No| D
    L --> D
    D -->|Yes| M["Flush partial output and release source pages"]
    M --> N["Read back output; report phase counters"]

Worked Example with a Full Trace

Use $B=2$, $C=4$, and $M=8$. The five supplied runs are R0 = $[1,4,4,9,11]$, R1 = $[]$, R2 = $[0,4,8]$, R3 = $[2,4,4,10,12]$, and R4 = $[]$. Only three sources are nonempty, so three input frames plus one output frame fit exactly. Total length is $N=13$.

Source pages. R0:P0 is [1@0:0,4@0:1], R0:P1 is [4@0:2,9@0:3], and R0:P2 is [11@0:4]. R2:P0 is [0@2:0,4@2:1] and R2:P1 is [8@2:2]. R3:P0 is [2@3:0,4@3:1], R3:P1 is [4@3:2,10@3:3], and R3:P2 is [12@3:4]. R1 and R4 have no pages.

Initialization. Read R0:P0, R2:P0, and R3:P0, costing three reads. The head set is [0@2:0,1@0:0,2@3:0] in priority order; output is empty. The following trace lists every emission, resulting head set, output-buffer change, and cumulative merge read/write count. Each step emits and possibly flushes before advancing/refilling its winning reader.

1. Emit 0@2:0. Output frame = [0@2:0]. No input refill. Heads afterward = [1@0:0,2@3:0,4@2:1]. Cumulative reads/writes: $3/0$.

1. Emit 1@0:0. Write page [0@2:0,1@0:0] and clear output. No input refill. Heads afterward = [2@3:0,4@0:1,4@2:1]. Cumulative reads/writes: $3/1$.

1. Emit 2@3:0. Output frame = [2@3:0]. No input refill. Heads afterward = [4@0:1,4@2:1,4@3:1]. Cumulative reads/writes: $3/1$.

1. Emit 4@0:1. Write page [2@3:0,4@0:1] and clear output. Read R0:P1. Heads afterward = [4@0:2,4@2:1,4@3:1]. Cumulative reads/writes: $4/2$.

1. Emit 4@0:2. Output frame = [4@0:2]. No input refill. Heads afterward = [4@2:1,4@3:1,9@0:3]. Cumulative reads/writes: $4/2$.

1. Emit 4@2:1. Write page [4@0:2,4@2:1] and clear output. Read R2:P1. Heads afterward = [4@3:1,8@2:2,9@0:3]. Cumulative reads/writes: $5/3$.

1. Emit 4@3:1. Output frame = [4@3:1]. Read R3:P1. Heads afterward = [4@3:2,8@2:2,9@0:3]. Cumulative reads/writes: $6/3$.

1. Emit 4@3:2. Write page [4@3:1,4@3:2] and clear output. No input refill. Heads afterward = [8@2:2,9@0:3,10@3:3]. Cumulative reads/writes: $6/4$.

1. Emit 8@2:2. Output frame = [8@2:2]. No input refill. Heads afterward = [9@0:3,10@3:3]. Cumulative reads/writes: $6/4$.

1. Emit 9@0:3. Write page [8@2:2,9@0:3] and clear output. Read R0:P2. Heads afterward = [10@3:3,11@0:4]. Cumulative reads/writes: $7/5$.

1. Emit 10@3:3. Output frame = [10@3:3]. Read R3:P2. Heads afterward = [11@0:4,12@3:4]. Cumulative reads/writes: $8/5$.

1. Emit 11@0:4. Write page [10@3:3,11@0:4] and clear output. No input refill. Heads afterward = [12@3:4]. Cumulative reads/writes: $8/6$.

1. Emit 12@3:4. Output frame = [12@3:4]. No input refill. Heads afterward = []. Cumulative reads/writes: $8/6$.

After step 13, write the final one-record page [12@3:4]. The output pages, in write order, are [0@2:0,1@0:0], [2@3:0,4@0:1], [4@0:2,4@2:1], [4@3:1,4@3:2], [8@2:2,9@0:3], [10@3:3,11@0:4], and [12@3:4].

The merged keys are $[0,1,2,4,4,4,4,4,8,9,10,11,12]$. All five fours retain the declared tie order: the two from R0, then R2's four, then the two from R3. Every source-position pair appears once. R2 is exhausted after step 9, R0 after step 12, and R3 after step 13; none is reinserted after exhaustion.

Counter audit. Initialization costs three reads; refills after steps 4, 6, 7, 10, and 11 add five. Total merge reads are $3+5=8$. Six full output pages and one final partial page give seven writes, so merge-only cost is $8+7=15$ transfers. Readback reads seven output pages, producing fifteen total reads, seven writes, and twenty-two reported transfers.

Source runs occupy $3+2+3=8$ pages, while the packed output occupies seven. Keeping the complete source image until output is finished yields fifteen peak live disk pages. The heap peaks at three reader indices and the record-buffer reservation at eight slots. Both empty runs retain their original indices but contribute no input frames or transfers.

Implementation in Python, C++, and Java

Save Python as main.py and run python3 main.py. Save C++ as main.cpp and compile with g++ -std=c++17 -O2 main.cpp -o k-way-merge. Save Java as Main.java and run javac Main.java followed by java Main. Python 3.7 or newer provides dataclasses; the C++ panel uses C++17; Java uses ordinary classes and a comparator-based priority queue.

The public functions are k_way_merge(runs, b, memory_blocks) in Python/C++ and kWayMerge(runs, b, memoryBlocks) in Java. Supply a list of lists, vector<vector<int>>, or int[][]; null/nonsequence sources are outside the contract. Input keys and run order are not mutated. Python checks exact integer types and rejects booleans/floats. C++/Java expose integer-typed parameters.

Validation order is: PARAMETER for invalid $B,C$; RUNS for more than $32$ sources; SIZE for more than $4096$ total keys; VALUE for any invalid key; ORDER for a descending adjacent pair; then MEMORY when the nonempty sources plus output exceed $C$ frames. These checks apply before any simulated disk access. The loops check all values before sortedness so error precedence is consistent across languages.

What is measured. Disk stores simulated pages and counts actual read/write method calls. Reader preloads one page for a nonempty run and refills only after consuming it. merge_to_disk/mergeToDisk heaps reader indices; a comparison inspects buffered records. It returns a complete disk run, after which input pages are released and readback reconstructs the observer list.

The result exposes merged records, merge_reads, merge_writes, readback_reads, heap_peak, peak_buffer_records, and peak_live_pages. Total reads are merge reads plus readback reads; total writes are merge writes. A single nonempty input is deliberately copied to materialized output under the same rules; a production pipeline could reuse or stream that run instead. Zero nonempty inputs produce no output pages or readback.

Model limits. peak_buffer_records reserves full $B$-slot input and output frames, excluding heap indices, page/run metadata, container spare capacity, object overhead, and scalar record temporaries. Source arrays, simulated disk payloads, and returned observer records occupy host memory, so the program is not a real RAM-capped file merger. Overall host storage is $O(N+k+qB)$ including those structures. Replace Disk with page I/O and stream final records to use the algorithm on actual files.

Sortedness validation scans the already supplied host arrays outside the I/O simulation. It must not be interpreted as a free validation scan of real files. A real merger can trust a run generator's contract, check order during the counted input reads, or pay for a separate validation pass. A streaming order failure invalidates any partial output already produced.

from dataclasses import dataclass
import heapq


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


@dataclass
class Run:
    pages: list
    length: int


@dataclass
class Result:
    records: list
    merge_reads: int
    merge_writes: int
    readback_reads: int
    heap_peak: int
    peak_buffer_records: int
    peak_live_pages: int


class Disk:
    def __init__(self, b):
        self.b = b
        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, run_id):
        pages = []
        for start in range(0, len(keys), self.b):
            pages.append(self.allocate([
                Record(keys[i], run_id, i)
                for i in range(start, min(start + self.b, len(keys)))
            ]))
        return Run(pages, 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, self.run = disk, run
        self.page = self.position = 0
        self.buffer = []
        self.load()

    def load(self):
        self.buffer.clear()
        self.position = 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.position]

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


def merge_to_disk(disk, sources, b):
    readers = [Reader(disk, run) for run in sources if run.length]

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

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

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


def k_way_merge(runs, b, memory_blocks):
    if type(b) is not int or type(memory_blocks) is not int:
        raise ValueError("PARAMETER")
    if not (1 <= b <= 32 and 2 <= memory_blocks <= 33):
        raise ValueError("PARAMETER")
    if len(runs) > 32:
        raise ValueError("RUNS")
    if sum(map(len, runs)) > 4096:
        raise ValueError("SIZE")
    for keys in runs:
        if any(type(x) is not int or not -1000000 <= x <= 1000000 for x in keys):
            raise ValueError("VALUE")
    for keys in runs:
        if any(keys[i] > keys[i + 1] for i in range(len(keys) - 1)):
            raise ValueError("ORDER")
    q = sum(bool(keys) for keys in runs)
    if q and q + 1 > memory_blocks:
        raise ValueError("MEMORY")
    disk = Disk(b)
    sources = [disk.source(keys, i) for i, keys in enumerate(runs)]
    output = merge_to_disk(disk, sources, b)
    merge_reads, merge_writes = disk.reads, disk.writes
    for run in sources:
        disk.erase(run)
    records = []  # Observer output is outside the merge-buffer budget.
    if output.length:
        reader = Reader(disk, output)
        while True:
            records.append(reader.head())
            if not reader.advance():
                break
    peak = (q + 1) * b if q else 0
    assert peak <= memory_blocks * b
    return Result(records, merge_reads, merge_writes,
                  disk.reads - merge_reads, q, peak, disk.peak)


def main():
    runs = [[1, 4, 4, 9, 11], [], [0, 4, 8], [2, 4, 4, 10, 12], []]
    b, blocks = 2, 4
    r = k_way_merge(runs, b, blocks)
    print(f"B={b} memory_blocks={blocks} M={b * blocks} nonempty={r.heap_peak}")
    print("merged=[" + ",".join(f"{a.key}@{a.run}:{a.position}" for a in r.records) + "]")
    print(f"merge reads={r.merge_reads} writes={r.merge_writes}")
    print(f"readback reads={r.readback_reads} writes=0")
    reads = r.merge_reads + r.readback_reads
    print(f"reads={reads} writes={r.merge_writes} transfers={reads + r.merge_writes} "
          f"merge_transfers={r.merge_reads + r.merge_writes}")
    print(f"heap_peak={r.heap_peak} peak_buffer_records={r.peak_buffer_records} "
          f"peak_live_pages={r.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, run, position;
    bool operator<(const Record& r) const {
        return tie(key, run, position) < tie(r.key, r.run, r.position);
    }
};
struct Run { vector<int> pages; int length; };
struct Result {
    vector<Record> records;
    int merge_reads, merge_writes, readback_reads;
    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, int run_id) {
        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], run_id, 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;
    size_t page = 0, position = 0;
    vector<Record> buffer;
    Reader(Disk& d, const Run& r) : disk(&d), run(&r) { buffer.reserve(d.b); load(); }
    bool load() {
        buffer.clear(); position = 0;
        if (page == run->pages.size()) return false;
        disk->read_into(run->pages[page++], buffer); return true;
    }
    const Record& head() const { return buffer[position]; }
    bool advance() { return ++position < buffer.size() || load(); }
};
Run merge_to_disk(Disk& disk, const vector<Run>& sources, int b) {
    vector<Reader> readers; readers.reserve(sources.size());
    for (const Run& run : sources) if (run.length) readers.emplace_back(disk, run);
    auto later = [&readers](int a, int c) { return readers[c].head() < readers[a].head(); };
    vector<int> initial;
    for (int i = 0; i < static_cast<int>(readers.size()); ++i) initial.push_back(i);
    priority_queue<int, vector<int>, decltype(later)> heap(later, move(initial));
    vector<Record> output; output.reserve(b);
    Run result; result.length = 0;
    for (const Run& run : sources) result.length += run.length;
    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)); output.clear();
        }
        if (readers[id].advance()) heap.push(id);
    }
    if (!output.empty()) result.pages.push_back(disk.write(output));
    return result;
}
Result k_way_merge(const vector<vector<int>>& runs, int b, int memory_blocks) {
    if (b < 1 || b > 32 || memory_blocks < 2 || memory_blocks > 33)
        throw invalid_argument("PARAMETER");
    if (runs.size() > 32) throw invalid_argument("RUNS");
    size_t n = 0;
    for (const auto& keys : runs) {
        if (keys.size() > 4096 - n) throw invalid_argument("SIZE");
        n += keys.size();
    }
    for (const auto& keys : runs) for (int key : keys)
        if (key < -1000000 || key > 1000000) throw invalid_argument("VALUE");
    for (const auto& keys : runs)
        if (!is_sorted(keys.begin(), keys.end())) throw invalid_argument("ORDER");
    int q = 0;
    for (const auto& keys : runs) if (!keys.empty()) ++q;
    if (q && q + 1 > memory_blocks) throw invalid_argument("MEMORY");
    Disk disk(b); vector<Run> sources; sources.reserve(runs.size());
    for (int i = 0; i < static_cast<int>(runs.size()); ++i)
        sources.push_back(disk.source(runs[i], i));
    Run output = merge_to_disk(disk, sources, b);
    int merge_reads = disk.reads, merge_writes = disk.writes;
    for (const Run& run : sources) disk.erase(run);
    vector<Record> records;
    if (output.length) {
        Reader reader(disk, output);
        do { records.push_back(reader.head()); } while (reader.advance());
    }
    int peak = q ? (q + 1) * b : 0;
    assert(peak <= memory_blocks * b);
    return Result{move(records), merge_reads, merge_writes, disk.reads - merge_reads,
                  q, peak, disk.peak};
}
int main() {
    vector<vector<int>> runs;
    runs.push_back({1,4,4,9,11}); runs.push_back({});
    runs.push_back({0,4,8}); runs.push_back({2,4,4,10,12}); runs.push_back({});
    int b = 2, blocks = 4;
    Result r = k_way_merge(runs, b, blocks);
    cout << "B=" << b << " memory_blocks=" << blocks << " M=" << b * blocks
         << " nonempty=" << r.heap_peak << "\nmerged=[";
    for (size_t i = 0; i < r.records.size(); ++i) {
        if (i) cout << ',';
        const Record& a = r.records[i]; cout << a.key << '@' << a.run << ':' << a.position;
    }
    int reads = r.merge_reads + r.readback_reads;
    cout << "]\nmerge reads=" << r.merge_reads << " writes=" << r.merge_writes
         << "\nreadback reads=" << r.readback_reads << " writes=0\n"
         << "reads=" << reads << " writes=" << r.merge_writes
         << " transfers=" << reads + r.merge_writes
         << " merge_transfers=" << r.merge_reads + r.merge_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, run, position;
        Record(int key, int run, int position) {
            this.key = key; this.run = run; this.position = position;
        }
        public int compareTo(Record r) {
            int c = Integer.compare(key, r.key);
            if (c == 0) c = Integer.compare(run, r.run);
            return c != 0 ? c : Integer.compare(position, r.position);
        }
    }
    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<Record> records;
        final int merge_reads, merge_writes, readback_reads;
        final int heap_peak, peak_buffer_records, peak_live_pages;
        Result(List<Record> records, int mr, int mw, int rr, int hp, int bp, int dp) {
            this.records = records; merge_reads = mr; merge_writes = mw; readback_reads = rr;
            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, int runId) {
            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], runId, 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 Run run;
        final List<Record> buffer;
        int page, position;
        Reader(Disk disk, Run run) {
            this.disk = disk; this.run = run; buffer = new ArrayList<>(disk.b); load();
        }
        boolean load() {
            buffer.clear(); position = 0;
            if (page == run.pages.size()) return false;
            disk.readInto(run.pages.get(page++), buffer); return true;
        }
        Record head() { return buffer.get(position); }
        boolean advance() { return ++position < buffer.size() || load(); }
    }
    static Run mergeToDisk(Disk disk, List<Run> sources, int b) {
        List<Reader> readers = new ArrayList<>();
        for (Run run : sources) if (run.length != 0) 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 < readers.size(); ++i) heap.add(i);
        List<Record> output = new ArrayList<>(b);
        List<Integer> pages = new ArrayList<>();
        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)); output.clear(); }
            if (reader.advance()) heap.add(id);
        }
        if (!output.isEmpty()) pages.add(disk.write(output));
        int length = 0;
        for (Run run : sources) length += run.length;
        return new Run(pages, length);
    }
    static Result kWayMerge(int[][] runs, int b, int memoryBlocks) {
        if (b < 1 || b > 32 || memoryBlocks < 2 || memoryBlocks > 33)
            throw new IllegalArgumentException("PARAMETER");
        if (runs.length > 32) throw new IllegalArgumentException("RUNS");
        int n = 0;
        for (int[] keys : runs) {
            if (keys.length > 4096 - n) throw new IllegalArgumentException("SIZE");
            n += keys.length;
        }
        for (int[] keys : runs) for (int key : keys)
            if (key < -1000000 || key > 1000000) throw new IllegalArgumentException("VALUE");
        for (int[] keys : runs) for (int i = 1; i < keys.length; ++i)
            if (keys[i - 1] > keys[i]) throw new IllegalArgumentException("ORDER");
        int q = 0;
        for (int[] keys : runs) if (keys.length != 0) ++q;
        if (q != 0 && q + 1 > memoryBlocks) throw new IllegalArgumentException("MEMORY");
        Disk disk = new Disk(b);
        List<Run> sources = new ArrayList<>();
        for (int i = 0; i < runs.length; ++i) sources.add(disk.source(runs[i], i));
        Run output = mergeToDisk(disk, sources, b);
        int mr = disk.reads, mw = disk.writes;
        for (Run run : sources) disk.erase(run);
        List<Record> records = new ArrayList<>();
        if (output.length != 0) {
            Reader reader = new Reader(disk, output);
            do { records.add(reader.head()); } while (reader.advance());
        }
        int peak = q == 0 ? 0 : (q + 1) * b;
        if (peak > memoryBlocks * b) throw new AssertionError("buffer budget");
        return new Result(records, mr, mw, disk.reads - mr, q, peak, disk.peak);
    }
    public static void main(String[] args) {
        int[][] runs = new int[5][];
        runs[0] = new int[]{1,4,4,9,11}; runs[1] = new int[0];
        runs[2] = new int[]{0,4,8}; runs[3] = new int[]{2,4,4,10,12}; runs[4] = new int[0];
        int b = 2, blocks = 4;
        Result r = kWayMerge(runs, b, blocks);
        System.out.println("B=" + b + " memory_blocks=" + blocks + " M=" + (b * blocks)
                           + " nonempty=" + r.heap_peak);
        StringBuilder s = new StringBuilder("merged=[");
        for (int i = 0; i < r.records.size(); ++i) {
            if (i != 0) s.append(',');
            Record a = r.records.get(i);
            s.append(a.key).append('@').append(a.run).append(':').append(a.position);
        }
        System.out.println(s.append(']'));
        System.out.println("merge reads=" + r.merge_reads + " writes=" + r.merge_writes);
        System.out.println("readback reads=" + r.readback_reads + " writes=0");
        int reads = r.merge_reads + r.readback_reads;
        System.out.println("reads=" + reads + " writes=" + r.merge_writes
                           + " transfers=" + (reads + r.merge_writes)
                           + " merge_transfers=" + (r.merge_reads + r.merge_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:

• B=2 memory_blocks=4 M=8 nonempty=3
• merged=[0@2:0,1@0:0,2@3:0,4@0:1,4@0:2,4@2:1,4@3:1,4@3:2,8@2:2,9@0:3,10@3:3,11@0:4,12@3:4]
• merge reads=8 writes=7
• readback reads=7 writes=0
• reads=15 writes=7 transfers=22 merge_transfers=15
• heap_peak=3 peak_buffer_records=8 peak_live_pages=15

Complexity, Correctness, and Invariants

CPU work. Only nonempty sources enter the heap. Initialization inspects $k$ descriptors and sets up $q$ readers. Python/C++ heap construction is $O(q)$; Java's initial insertions are $O(q\log(q+1))$. The loop performs exactly $N$ removals and $N-q$ reinserts. Each priority-queue operation costs at most $O(\log(q+1))$, yielding $$ O\bigl(k+N\log(q+1)\bigr) $$ CPU time overall, including validation and linear page copying/readback. This handles $q=1$ as linear work and $q=0$ as descriptor validation; using $\log q$ without these cases would misdescribe them.

The same upper bound holds for best, average, and worst key arrangements among valid sorted inputs. Heap occupancy can shrink as sources exhaust, reducing actual comparisons. There is no specified random distribution from which to derive one exact average comparison count. For balanced nonempty runs, the comparison worst case requires $\Omega(N\log q)$ comparisons when $q\ge2$: many interleavings remain possible despite each source's internal order. For known nonoverlapping key ranges or one nonempty run, special handling can reduce work; the general heap does not infer a range-based concatenation shortcut.

Exact single-pass I/O. Define $$ D=\sum_{i=0}^{k-1}\left\lceil\frac{n_i}{B}\right\rceil, \qquad O=\left\lceil\frac{N}{B}\right\rceil. $$ Empty sources contribute zero. The letter $O$ here is a named output-page count; $O(\cdot)$ denotes asymptotic upper bounds elsewhere. Each input page is read once and every packed output page is written once, so $$ T_{\mathrm{merge}}=D+O, \qquad T_{\mathrm{reported}}=D+2O. $$ Readback adds $O$ reads and no writes. For $N\gt0$, separate input-run tails imply $$ O\le D\le O+q-1,\qquad D\ge q. $$ Thus single-pass merge transfers are $\Theta(D+O)=\Theta(N/B+q)$, with ceilings understood. For fixed run lengths and $B$, best, average, and worst key interleavings have exactly the same transfer counts. Heap comparisons affect CPU work but do not trigger extra page reads.

The materialized merge reaches the minimum data-page reads/writes in this model: every input page must be consumed and every output page produced. That statement assumes the sorted-input contract, this page layout, and a materialized output. Extra prevalidation scans, different run packing, caching, or pipelining change the accounting. Counted readback is optional observation, not a requirement of the merge algorithm.

Record buffers and metadata. For nonempty input, the reserved record-buffer peak is $(q+1)B\le M=CB$. Input frames remain reserved throughout the merge even as readers exhaust. The heap contains at most $q$ indices, with $O(q)$ reader/heap metadata. The simulator stores input and output page-ID lists using $O(D+O)$ metadata, source descriptors using $O(k)$, and observer output using $O(N)$ host storage. Metadata and representation overhead require additional real bytes.

Disk space. This implementation retains all $D$ source pages until the entire $O$-page output exists, so its peak is exactly $D+O$ live payload pages. It then releases the input payloads and retains the output. Empty input peaks at zero. A production implementation may reclaim input blocks earlier, but only if its ownership and recovery rules allow it.

Frontier invariant. Before an emission, every nonexhausted source has exactly one heap index and its buffered head is its smallest unconsumed record. Because the source is sorted, every hidden later record is at least that head. Taking the minimum head therefore selects the smallest remaining record globally.

Inductive step. Emit that minimum, preserving the sorted output prefix. Only its source changes. Popping its index before advancing avoids changing a priority still resident in the heap. Reinsert its next head when present; otherwise omit the exhausted source. All other heads remain valid. Writing a full frame or final partial frame preserves record order and exact counts; it merely moves an output prefix to disk.

Coverage, termination, and ties. Each iteration consumes one previously unconsumed source record, so after exactly $N$ iterations no records remain and the heap is empty. No record is skipped, duplicated, or created. Comparing $(k,i,j)$ by key, source index, and source position preserves the declared order for duplicates; sorted input keys plus increasing within-run positions ensure each run is sorted under that total order. This proves the complete output contract.

When all runs do not fit. A one-pass fan-in $f$ with one frame per input plus one output requires $(f+1)B\le M$. Group runs into feasible merges and repeat; for $q\gt1$ and fixed $f\ge2$, at most $\lceil\log_f q\rceil$ balanced merge passes suffice. Extra passes add input/output transfers. The programs reject insufficient memory so they never silently exceed the stated frame budget.

Variants, Pitfalls, and Optimizations

Sortedness is essential. A heap of heads does not sort arbitrary input runs. If a source is [1,100,0], its final zero can remain hidden while larger values are emitted. The supplied API rejects this before simulation. Real stream validation can discover such an error later; its partial result must then be rejected. Every source must also use the same comparator, including string collation or null ordering when those are relevant.

Empty and exhausted sources. Empty runs contribute no head and must never be dereferenced. When a reader reaches the end of its final partial page, remove it instead of inventing an infinity record. A legal maximum key cannot safely serve as a sentinel unless it is distinguished from real data. Keep original source IDs when filtering empty runs, since the tie contract uses those IDs.

Stable ties depend on the intended original order. Key-only heaps may choose equal records in unspecified order. This lesson explicitly uses source index and within-run position. That is appropriate for stability relative to concatenated runs. If earlier run generation rearranged records across runs, carry original global IDs through the merge to recover stability relative to the original input stream. Do not replace those IDs with the new run number.

Keep heap keys consistent. Pop an index before advancing its reader, then push it back if still valid. Comparators read buffered heads, so mutating a head while its index remains in the heap can silently invalidate order. Comparators should use direct comparisons rather than subtraction to avoid overflow in wider key domains.

Flush and count partial pages. Each input run can have a short last page. The output's final short page must also be written. Preserve valid lengths; padding with ordinary zeros or extrema fabricates data. Distinguish $D=\sum_i\lceil n_i/B\rceil$ from $\lceil N/B\rceil$: separately rounded source tails can use more pages than packed output.

Fan-in versus buffering. The largest feasible number of streams can reduce merge passes but leaves less room for prereading or double buffering. File-descriptor limits, cache behavior, seek patterns, and asynchronous reads may favor a smaller fan-in with larger per-source buffers. PostgreSQL's tuplesort source describes multiway head selection, prereading for locality, and an on-demand final merge that can avoid materializing and rereading final output. Those are practical choices beyond this one-frame simulator.

Library and selection variants. Python's heapq.merge documentation describes a lazy merge of already sorted iterables without pulling all records into memory. The explicit programs here expose page access, counters, validation, and tie policy for teaching. Root replacement can combine pop/push work when a successor exists. A loser tree can lower comparison constants; specialized two-way merge is often simpler for exactly two sources. These choices retain the same frontier proof.

Aggregation and top results. To deduplicate, compare adjacent emitted keys and define which payload survives; do not silently drop equal heap heads. Group-by aggregation can similarly combine a complete equal-key group. A top-$t$ consumer can stop after $t$ outputs, but total order, tie rules, source closure, and pages already prefetched determine its partial-I/O behavior. The full-output counters here do not apply unchanged to early stopping.

Streaming and parallel systems. A live sorted stream may temporarily lack its next record without being exhausted. Absence is not an infinity head: the merger must wait or use a valid watermark/lower-bound protocol to know that no unseen earlier key can arrive. Distributed merging also needs consistent comparison rules and backpressure. Parallel merge partitions can use key boundaries or rank-based splits, with care around duplicate ranges and shared storage bandwidth.

Practice Problems and Real-World Use

An independent oracle can flatten records into triples $(\text{key},\text{run},\text{position})$ and sort those triples. Compare the complete merged list against it: sortedness alone would miss a lost duplicate or a skipped source record. Also assert that caller inputs remain unchanged and that the heap/frame peaks fit the declared budget.

Derive expected counters from lengths alone, independently of the heap: $D$ input reads, $O$ output writes, $O$ optional readback reads, $(q+1)B$ reserved slots for nonempty input, and $D+O$ peak live pages under this retention policy. The exact output order should be invariant under changes in $B$ and excess available frames, provided the merge remains feasible.

The three programs agree on $6037$ shared inputs: $5679$ valid cases and $358$ rejected cases. The independent oracle compares every stable record triple and every reported transfer/storage counter. Cases include exhaustive combinations of short sorted runs, randomized source lengths and key overlaps, all-equal and mixed-sign keys, empty runs interspersed with nonempty runs, single-source output, partial pages, thirty-two sources, the total-size limit $4096$, unsorted sources, invalid values/parameters, and insufficient frames. Python also passed $12$ exact-type rejection checks and $400$ page-size invariance comparisons. C++ ran with address and undefined-behavior sanitizers; sample stdout matches byte for byte in all languages.

To extend the experiment, compare one-pass merging with balanced multipass merging under the same byte budget. Track record reads/writes and intermediate pages, not merely comparator calls. Then replace simulated storage with actual files and stream a verifier, keeping measured device requests, bytes, elapsed time, and logical page-transfer counts separate.