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

待排序的账册已经大到装不进内存，档案员只好把问题拆成一批磁盘上的有序段。

当待排序数据超过 operator memory budget，执行器会把部分有序 runs 写入 temporary storage，再多路归并。外部排序的核心不是换一个排序算法，而是让每一趟尽量顺序读写，并用有限 buffers 控制 merge fan-in。

本课目标

• 推导 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
...

merge passes 数量约为 ceil(log_k R)。每个完整 materialized pass 读取 N、写出 N，约 2N page I/O。最后结果若直接 pipelined 给 parent operator，可省掉最终完整写出，具体由 plan 决定。

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

一个正确的 k-way merge 骨架

以下程序假设每个 input file 每行一个 integer，且已按整数升序：

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))

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 vault_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 线性推进：

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 vault_items
ORDER BY value_cents, item_id;

计划中的 Sort Method、Memory/Disk usage 能说明该 sort node 是否 spill。work_mem 是每个 sort/hash operation 的预算起点，parallel workers 和同一 query 多个 nodes 可能各自使用，因此不能把它直接设为“服务器空闲内存”。默认值也取决于当前配置，不应硬编码为 4 MB。

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 与资源预算。