7.2 外部排序、多路归并与 Merge Join

待排序数据超过内存预算后，执行器先写出一批有序段，再用有限缓冲区多路归并。本课从段数和归并路数推算临时 I/O，并验证重复值不会丢失。ORDER BY 写得再短，也不会让内存预算变大。

本课目标

• 推导 run generation 与 merge passes；
• 理解 memory、fan-in、temporary I/O 的关系；
• 区分 full sort、top-N、incremental sort；
• 解释 sort-based distinct/aggregation 与 merge join；
• 写一个有边界、能处理 duplicate 的 k-way merge。

Phase 1：生成 sorted runs

设输入 N pages，sort memory 可容纳约 M pages：

read up to M pages
sort records in memory
write one sorted run
repeat

initial run count ≈ ceil(N / M)

真实 memory 要分给 tuples、pointers/keys、comparison state 和 allocator，不能用配置 bytes 除 row width 得到精确 M。variable-length rows 与 abbreviated keys 也影响容量。

replacement selection 在随机输入下可能生成平均长于 memory 的 runs，但现代 engine 具体采用 quicksort、radix、replacement selection 或其他策略要看实现。

Phase 2：k-way merge

若有 B buffer pages，常用 B-1 input buffers + 1 output buffer，fan-in 约为 B-1，实际还受 file descriptors、prefetch、parallelism 与 per-run metadata 限制。

pass 0: R initial runs
pass 1: ceil(R / k) runs
pass 2: ceil(previous / k) runs
...

当 R > 1 且 k >= 2 时，merge passes 数量约为 ceil(log_k R)；零或一个 run 不需要归并。每个完整 materialized pass 读取 N、写出 N，约 2N page I/O。最后结果若直接 pipelined 给 parent operator，可省掉最终完整写出，具体由 plan 决定。

例如假设 N=10000、M=100，先得到 100 个 runs；k=9 时依次剩 12、2、1 个，需要三趟归并。若生成 runs 和每趟结果都落到临时存储，约需 2N × (1+3) = 80000 次页读写；最后直接输出则可少写 N 页。

外部排序不是 O(1) space：它使用 O(memory budget) RAM，并需要可能接近 input/output 数量级的 temporary disk；multi-pass 与 compression 会改变峰值。

一个正确的 k-way merge 骨架

以下 Python 3.10+ 程序实现一次 k 路归并：每个输入文件每行一个整数，且已按整数升序。它不生成初始 runs，也没有限制同时打开的文件数；输入很多时需按 fan-in 分批。保存为 merge_runs.py：

from __future__ import annotations

from contextlib import ExitStack
import heapq
from pathlib import Path
from typing import TextIO

def read_integer(handle: TextIO) -> int | None:
    line = handle.readline()
    if line == "":
        return None
    return int(line.strip())

def merge_integer_runs(inputs: list[Path], output: Path) -> None:
    if output in inputs:
        raise ValueError("output must not overwrite an input run")

    with ExitStack() as stack:
        handles = [
            stack.enter_context(path.open("r", encoding="utf-8"))
            for path in inputs
        ]
        destination = stack.enter_context(output.open("x", encoding="utf-8"))

        heap: list[tuple[int, int]] = []
        for run_id, handle in enumerate(handles):
            value = read_integer(handle)
            if value is not None:
                heapq.heappush(heap, (value, run_id))

        while heap:
            value, run_id = heapq.heappop(heap)
            destination.write(f"{value}\n")

            next_value = read_integer(handles[run_id])
            if next_value is not None:
                heapq.heappush(heap, (next_value, run_id))

把下面的入口接在同一文件末尾，运行 python3 merge_runs.py。预期输出 [1, 2, 3, 3, 3, 8]；重复值保留，空输入文件不产生记录：

if __name__ == "__main__":
    from tempfile import TemporaryDirectory

    with TemporaryDirectory() as directory:
        root = Path(directory)
        inputs = [root / "a.txt", root / "b.txt", root / "empty.txt"]
        for path, values in zip(inputs, ([1, 3, 3], [2, 3, 8], [])):
            path.write_text("".join(f"{value}\n" for value in values), encoding="utf-8")
        output = root / "merged.txt"
        merge_integer_runs(inputs, output)
        merged = [int(line) for line in output.read_text().splitlines()]
        assert merged == [1, 2, 3, 3, 3, 8]
        print(merged)

heap 使用 (value, run_id)，相同 value 时仍能比较，duplicate 会完整保留。生产实现还需要 buffered binary I/O、record serialization、checksum、temporary cleanup、disk-full handling、stable tie-breaker 和 multi-pass fan-in limit。

Double buffering 与 asynchronous prefetch

每个 run 可准备 active buffer 与 next buffer：CPU merge 当前 block 时，I/O 预取下一 block。它隐藏部分 latency，但仍受 storage bandwidth、queue、memory 与 scheduler 限制。

input runs 很多时，为每个 run 分配两个大 buffers 会挤压 sort memory 并降低 fan-in，需整体权衡。

Full sort 不总是必要

Top-N

SELECT item_id, value_cents
FROM catalog_items
ORDER BY value_cents DESC, item_id DESC
LIMIT 100;

若没有可直接提供 order 的 index，executor 可维护 bounded heap，只保留当前最佳 N rows，memory 约与 N 相关，而不是完整排序所有 rows。仍必须扫描候选输入，除非其他 access path/partition pruning 减少它。

OFFSET 很大时 bounded heap 可能需要保留 N + offset，成本上升。keyset pagination 更适合稳定深分页。

Incremental sort

若输入已按 (a) 排序，而目标是 (a, b)，executor 可在每个相同 a group 内排序 b，不必一次保存全部 input。收益取决于 prefix order 和 group sizes。

Index order

B+ tree index order 与 query order compatible 时，可按 index scan 输出，避免显式 sort。但 random heap fetch、coverage 与命中行数可能让 optimizer 仍选择 scan + sort。

DISTINCT 与 GROUP BY

sorted input 上，相邻相等 keys 可 deduplicate/aggregate：

sort by group key
scan once
accumulate current group
emit when key changes

hash aggregation 也可完成同一 logical task，并在 memory 不足时 partition/spill。哪种更好取决于 group cardinality、input order、memory、parallelism 和后续 order requirement。

“DISTINCT 数据大就一定外排序”不正确；planner 可以用 hash、sort、index uniqueness 或其他 operators。

Merge join

两侧按 join key 有序时，merge join 线性推进输入；计入输出对数 Z 后，总工作量通常写成 O(left_rows + right_rows + Z)：

left.key < right.key  -> advance left
left.key > right.key  -> advance right
equal                 -> output Cartesian matches of both equal-key groups

最后一点很重要。如果 left key=7 有 3 rows、right key=7 有 4 rows，inner join 要输出 12 pairs。不能只“双指针各前进一步”，否则漏掉 duplicate matches。大的重复键分组还可能需要缓冲、回扫或溢写，不能仅凭双指针就认定内存为常数。

merge join 的输入可来自：

• index/order-preserving scan；
• explicit sort；
• upstream operator 已有 order。

若需要先对两侧大表外排序，总成本可能高于 hash join；若结果还需同样 order，排序成本又可能被复用。

NULL equality、collation、outer join unmatched rows 和 inequality merge join 都需按 SQL semantics 处理。

PostgreSQL 中观察 spill

EXPLAIN (ANALYZE, BUFFERS, TIMING OFF)
SELECT item_id, value_cents
FROM catalog_items
ORDER BY value_cents, item_id;

这里需要 PostgreSQL 中已存在的 catalog_items(item_id, value_cents) 表；这是执行计划片段，不沿用 SQLite 连接。ANALYZE 会实际执行查询，Sort Method 和 Memory/Disk usage 描述本次执行的排序方式与 spill。work_mem 是每个 sort/hash operation 的预算起点，parallel workers 和同一 query 多个 nodes 可能各自使用，因此不能把它直接设为“服务器空闲内存”。配置应读取当前实例；哈希节点还受 hash_mem_multiplier 影响，不能简单按一个 work_mem 估算整条查询。PostgreSQL 内存配置

database-level temporary file counters 可用于趋势，但会汇总多个 sessions/queries，不能单独归因某一 plan。结合 log_temp_files、query stats 和 EXPLAIN 分析。

I/O 与稳定性边界

• temporary filesystem 空间不足会让 query 失败；
• 多个 concurrent spills 会争抢同一 device；
• sort comparison 受 collation/type/operator 影响；
• ORDER BY key 在 duplicate key 内没有稳定顺序，pagination 应加 unique tie-breaker；
• sort algorithm 是否 stable 是 implementation detail，SQL 不保证未列入 ORDER BY 的顺序；
• output rows 很大时，可排序 key + row reference，晚些 materialize payload。

验收清单

• [ ] 能根据 N、M、fan-in 估算 runs 与 passes；
• [ ] 不把 external sort 的 disk space 写成 O(1)；
• [ ] 能解释 top-N 为什么仍可能扫描全部 input；
• [ ] merge join 正确处理 duplicate groups；
• [ ] ORDER BY 含 unique tie-breaker；
• [ ] spill 测试同时记录 temp bytes、latency、concurrency 与 device；
• [ ] 修改 memory budget 前计算 per-node/per-worker 放大。

本章小结

hash index 放弃 key order 以优化 equality，external sort 则用 runs 与多路归并在有限内存中构造 order。下一章会把 scan、join、sort、aggregate 放进同一 iterator/vectorized execution model，并解释 optimizer 怎样选择 operators 与资源预算。