# 实验 11：外部归并排序

> 当一张表大到装不进内存时，如何用有限的缓冲页把它排好序，并把代价用"读写了多少页"算清楚。

## 一、实验目标

1. 理解为什么数据库的排序不能只靠内存排序算法（如快排、归并排序的数组版本）。
2. 掌握外部归并排序（external merge sort）的两个阶段：顺串生成（run generation）与多路归并（multi-way merge）。
3. 会用页数（page）而不是记录条数来估算 I/O 代价，能用公式算出归并趟数与总 I/O。
4. 能亲手实现一个带 I/O 计数的外部归并排序，并把实测值与理论公式对上。
5. 理解缓冲页数 B 对代价的影响：内存不是越多越线性变好，趟数是阶梯式下降的。

## 二、预备知识

### 2.1 页（page）、块（block）与缓冲页（buffer frame）

磁盘（含 SSD）与内存之间交换数据的最小单位叫**页**（page），也叫块（block）。一条记录可能只有几十字节，但读写一次至少是一整页。因此数据库里算 I/O 代价时，单位是"页"，不是"条"。

- N：待排序数据占的页数（本实验 N = 500 页）。
- 页大小（page size）：一页能放多少条记录（本实验每页 100 条）。
- B：排序可用的**缓冲页数**（buffer frames）。这是本实验里最稀缺的资源。

数据库管理系统（DBMS）会在内存中划出一小块区域作为缓冲区（buffer pool），排序只能使用其中的 B 页。超出这个额度就必须把中间结果写到磁盘上。

### 2.2 顺串（run）

**顺串**（run）指一段已经排好序的记录序列。顺串可以放在内存里，也可以写在磁盘上。外部排序的第一阶段就是把原始数据切成若干段，每段装进 B 页内存、排好、写回磁盘，得到若干个有序顺串。

原始数据（未排序）：

```
[ 7 3 9 1 5 8 2 6 4 ]      共 9 条，假设 B 页内存一次只能装 3 条
```

切成 3 段各自排序后：

```
run1 = [1 3 7]   run2 = [2 5 8]   run3 = [4 6 9]
```

### 2.3 趟（pass）与归并路数（fan-in）

**趟**（pass）：把全部数据从头到尾读一遍、再写一遍，算一趟。

**归并路数**（fan-in）：一趟归并中能同时合并多少个顺串。归并时需要给每个输入顺串留一页输入缓冲，还要给输出留至少一页输出缓冲，所以：

```
fan_in = B − 1
```

这就是"留 1 页输出"的经典约定（出处：CMU 15-445 课程 "Sorting & Aggregations" 讲义）。

```
       输入缓冲（B−1 页）                      输出缓冲（1 页）
   ┌────┬────┬────┬────┐                    ┌──────────┐
   │run1│run2│run3│run4│   ── k 路归并 ──▶  │  输出页  │ ── 写满一页就落盘
   └────┴────┴────┴────┘                    └──────────┘
        B = 5 页  →  fan_in = 4
```

### 2.4 k 路归并怎么做到"每趟只读一次"

k 路归并用一个大小 k 的**小顶堆**（min-heap）：每路顺串取当前最小的一条入堆，反复弹出堆顶（全局最小）写入输出，再从对应顺串补一条进堆。Python 标准库 `heapq.merge` 就是这个算法的现成实现。

### 2.5 I/O 代价模型

本实验采用教学口径的计数规则：

- 从磁盘**读**一页记 1 次 I/O；
- 向磁盘**写**一页记 1 次 I/O；
- 顺序读写与随机读写在本实验中不做区分（真实系统里随机 I/O 贵得多，这也是排序优于"建索引后随机查找"的原因之一）。

一整趟归并的代价 = 读入全部 N 页 + 写出全部 N 页 = **2N**。

## 三、实验原理

### 3.1 两个阶段

**阶段 1：顺串生成（run generation）**

```
   磁盘（N 页原始数据）          内存缓冲区（B 页）            磁盘（顺串文件）
 ┌────────────────────┐      ┌───┬───┬───┬───┐          ┌──────────────────┐
 │ 页1 页2 … 页N      │ 读▶ │   │   │   │   │  排好写▶ │ run1 run2 … runK │
 └────────────────────┘      └───┴───┴───┴───┘          └──────────────────┘
                                                        K = ⌈N / B⌉ 个顺串
```

每读满 B 页就排序并写回，得到 K = ⌈N/B⌉ 个顺串。这一阶段读写各一遍全部数据，代价 **2N**。

**阶段 2：多路归并（merge）**

每一趟把 fan-in = B−1 个顺串合并成 1 个，顺串数量按 (B−1) 倍缩减：

```
K → ⌈K/(B−1)⌉ → ⌈K/(B−1)²⌉ → … → 1
```

每趟代价 2N。

### 3.2 代价公式

归并趟数（不含阶段 1 的生成趟，通常把阶段 1 记作第 1 趟）：

```
总趟数 P = 1 + ⌈ log_{B−1} ⌈N/B⌉ ⌉
总 I/O  = 2N × P
```

符号说明：

| 符号 | 含义 | 单位 |
|---|---|---|
| N | 待排序数据的页数 | 页 |
| B | 排序可用缓冲页数 | 页 |
| B−1 | 归并路数 fan-in（留 1 页输出） | 个顺串/趟 |
| ⌈N/B⌉ | 阶段 1 生成的顺串个数 | 个 |
| P | 总趟数（含顺串生成那一趟） | 趟 |
| 2N | 一整趟的 I/O（读 N 页 + 写 N 页） | 页 |

出处：CMU 15-445 "Sorting & Aggregations" 讲义。

**公式的省略项**：如果排序结果是直接流式（streaming）交给上层算子（比如 Top-k、聚合），最后一趟的输出可以不必落盘，能省掉 N 次写。本实验为了与公式对齐，把最后一趟的写回也计算在内。

### 3.3 内存多大才算够

PostgreSQL 的 `work_mem` 参数（官方文档 19.4 Resource Consumption）默认 4 MB，它是每个排序/哈希操作在落盘前的内存上限。把它换算成页数就能直接套用上面的公式：内存越大 → 顺串越少 → 归并趟数越少 → I/O 越少。

### 3.4 本实验的实测结论

下图是本机实测结果：横轴是缓冲页数 B，左图对比"真实执行的 I/O 页数"与"公式值"，右图是归并趟数随 B 的变化。可以看到两条柱子完全重合，说明教学模型与公式是自洽的。

![外部归并排序实测](../figures/fig_extsort.png)

## 四、实验环境与准备

- 操作系统：Windows 11（本指导书所有数字在 CPython 3.13 + Windows 上实测获得）
- Python ≥ 3.9；需要 `numpy`（造数据）、`heapq`（标准库，多路归并）、`math`（标准库，向上取整）
- 可选：`pandas`、`matplotlib`（画观测图）
- 本仓库目录（用于选做部分）：`system/sorting.py` 提供 `RunGenerator`、`SingleStreamMerge`、`ExternalMergeSort` 三个类；`system/queues/` 提供 `external_queues.py`（落盘队列）、`list_queues.py`（内存队列）与 `queue_factories.py`（队列工厂）。

安装依赖（如已安装可跳过）：

```bash
python -m pip install numpy pandas matplotlib
```

准备实验数据目录：把本指导书配套的代码存成一个 `.py` 文件，例如 `lab11_extsort.py`，在仓库根目录下运行，这样选做部分才能 `import system`。

## 五、实验步骤

### 步骤 1：造一份"大到装不下"的数据

本实验的规模：50,000 条整数，按每页 100 条计，共 **N = 500 页**。这个规模在 Python 里跑得快，同时又能体现出多趟归并。

```python
import numpy as np

PAGE_SIZE = 100                      # 每页 100 条记录
N_PAGES = 500                        # 数据共 500 页
rng = np.random.default_rng(0)       # 固定随机种子，保证可复现
data = rng.integers(0, 10**9, N_PAGES * PAGE_SIZE).tolist()

print("记录条数:", len(data))
print("数据页数:", -(-len(data) // PAGE_SIZE))
```

### 步骤 2：算法 1 —— 内存内排序（对照组）

先看看"如果能全装进内存"该怎么做，作为外部排序的对照基准。

```python
sorted_in_memory = sorted(data)
print(sorted_in_memory[:5])
```

内存内排序的代价是 O(n log n) 次比较，几乎不产生磁盘 I/O；但它的前提是内存装得下。真实的大表排序（几十 GB）根本没有这个前提。

### 步骤 3：搭一个按页计数的 I/O 计数器

真实的磁盘读写我们用一个计数器模拟：读一页 `+1`，写一页 `+1`。

```python
import math

class IOCounter:
    """按页计数：读一页 +1，写一页 +1。"""
    def __init__(self):
        self.read_pages = 0
        self.write_pages = 0

    def read(self, n_tuples):
        pages = math.ceil(n_tuples / PAGE_SIZE)
        self.read_pages += pages
        return pages

    def write(self, n_tuples):
        pages = math.ceil(n_tuples / PAGE_SIZE)
        self.write_pages += pages
        return pages

    @property
    def total(self):
        return self.read_pages + self.write_pages
```

### 步骤 4：实现外部归并排序（带 I/O 计数）

关键点：阶段 1 按 B 页切块；阶段 2 每趟最多合并 `B−1` 个顺串。

```python
import heapq

def external_merge_sort(data, B, counter):
    """外部归并排序。返回 (排序结果, 归并趟数, 阶段1生成的顺串数)。

    :param data: 待排序的记录列表
    :param B: 可用缓冲页数
    :param counter: IOCounter 实例
    """
    # ------- 阶段 1：顺串生成 -------
    runs = []
    for start in range(0, len(data), B * PAGE_SIZE):
        chunk = data[start:start + B * PAGE_SIZE]
        counter.read(len(chunk))      # 读入这一块（B 页）
        chunk.sort()                  # 在内存中排序
        counter.write(len(chunk))     # 写回一个顺串
        runs.append(chunk)
    n_runs_initial = len(runs)

    # ------- 阶段 2：多路归并 -------
    fan_in = max(2, B - 1)            # B=2 时兜底，保证至少能两两归并
    passes = 1                        # 阶段 1 记作第 1 趟
    while len(runs) > 1:
        new_runs = []
        for i in range(0, len(runs), fan_in):
            group = runs[i:i + fan_in]
            for r in group:
                counter.read(len(r))            # 读入待归并的顺串
            merged = list(heapq.merge(*group))  # k 路归并（小顶堆）
            counter.write(len(merged))          # 写回合并结果
            new_runs.append(merged)
        runs = new_runs
        passes += 1
    return (runs[0] if runs else []), passes, n_runs_initial
```

### 步骤 5：对不同的 B 跑一遍，填观测表

```python
Bs = [3, 5, 9, 17, 33, 65]
rows = []
for B in Bs:
    counter = IOCounter()
    out, passes, n_runs = external_merge_sort(list(data), B, counter)
    assert out == sorted(data), "排序结果不正确！"      # 正确性校验
    N = math.ceil(len(data) / PAGE_SIZE)
    formula = 2 * N * (1 + math.ceil(math.log(max(2, math.ceil(N / B)), max(2, B - 1))))
    rows.append({
        "缓冲页数 B": B,
        "初始顺串数 ⌈N/B⌉": n_runs,
        "fan-in (B−1)": max(2, B - 1),
        "实测趟数": passes,
        "公式趟数": 1 + math.ceil(math.log(max(2, math.ceil(N / B)), max(2, B - 1))),
        "实测 I/O 页数": counter.total,
        "公式 I/O 页数": formula,
    })

for r in rows:
    print(r)
```

本机实测输出（N = 500 页，每页 100 条）：

| 缓冲页数 B | 初始顺串数 ⌈N/B⌉ | fan-in | 趟数 | 实测 I/O 页数 | 公式 I/O 页数 |
|---|---|---|---|---|---|
| 3 | 167 | 2 | 9 | 9000 | 9000 |
| 5 | 100 | 4 | 5 | 5000 | 5000 |
| 9 | 56 | 8 | 3 | 3000 | 3000 |
| 17 | 30 | 16 | 3 | 3000 | 3000 |
| 33 | 16 | 32 | 2 | 2000 | 2000 |
| 65 | 8 | 64 | 2 | 2000 | 2000 |

实测与公式完全一致。

### 步骤 6：把结果画出来（可选）

```python
import matplotlib.pyplot as plt

io_real = [r["实测 I/O 页数"] for r in rows]
io_formula = [r["公式 I/O 页数"] for r in rows]
passes = [r["实测趟数"] for r in rows]

plt.rcParams["font.sans-serif"] = ["Microsoft YaHei", "SimHei", "DejaVu Sans"]
plt.rcParams["axes.unicode_minus"] = False

fig, axes = plt.subplots(1, 2, figsize=(12, 4.6))
x = range(len(Bs))
w = 0.38
axes[0].bar([i - w / 2 for i in x], io_real, w, label="实测 I/O")
axes[0].bar([i + w / 2 for i in x], io_formula, w, label="公式 I/O")
axes[0].set_xticks(list(x))
axes[0].set_xticklabels([f"B={b}" for b in Bs])
axes[0].set_ylabel("I/O 页数")
axes[0].set_title("实测与公式一致")
axes[0].legend()
axes[0].grid(axis="y", ls=":", alpha=0.4)

axes[1].plot(Bs, passes, "o-", color="#C00000")
axes[1].set_xscale("log")
axes[1].set_xlabel("缓冲页数 B（对数轴）")
axes[1].set_ylabel("归并趟数")
axes[1].set_title("趟数随 B 阶梯式下降")
axes[1].grid(ls=":", alpha=0.4)

plt.tight_layout()
plt.show()
```

### 步骤 7：自己改参数再看一次

把 `PAGE_SIZE`、`N_PAGES`、`Bs` 各改一遍，重新跑步骤 5。重点观察两件事：

1. 实测 I/O 是否始终等于公式值；
2. 趟数是否出现"内存翻倍但代价不变"的平台（例如上表中 B=9 与 B=17 都是 3 趟）。

### 步骤 8（选做）：用仓库里的真实实现跑一遍

仓库 `system/sorting.py` 里的 `ExternalMergeSort` 用的是真实的落盘队列（`system/queues/external_queues.py`），不再是我们这个"内存列表模拟磁盘"的教学模型。可以用内存队列（`list_queues.py`）先跑通，再换成外部队列观察行为差异。

```python
import sys, pathlib
for _p in (pathlib.Path.cwd(), *pathlib.Path.cwd().parents):
    if (_p / "system").is_dir():
        sys.path.insert(0, str(_p))
        break

from system.sorting import ExternalMergeSort
from system.queues.external_queues import ExternalQueueFactory

# number_of_tuples_in_main_memory 相当于"内存里能放多少条"，
# 它除以每页条数就对应实验里的缓冲页数 B
sorter = ExternalMergeSort(
    input_data=iter(data[:2000]),
    queue_factory=ExternalQueueFactory(),
    number_of_tuples_in_main_memory=300,
    fan_in=2,
)
print(list(sorter)[:5])
```

## 六、观测记录表

请把自己机器上的运行结果填入下表（前两列已给出本指导书编写时的实测值作为参照）。

| B | 初始顺串数 ⌈N/B⌉ | fan-in (B−1) | 实测趟数 | 公式趟数 | 实测 I/O 页数 | 公式 I/O 页数 | 是否一致 |
|---|---|---|---|---|---|---|---|
| 3 | 167 | 2 | 9 | 9 | 9000 | 9000 | |
| 5 | 100 | 4 | 5 | 5 | 5000 | 5000 | |
| 9 | 56 | 8 | 3 | 3 | 3000 | 3000 | |
| 17 | 30 | 16 | 3 | 3 | 3000 | 3000 | |
| 33 | 16 | 32 | 2 | 2 | 2000 | 2000 | |
| 65 | 8 | 64 | 2 | 2 | 2000 | 2000 | |
| （自定 B=____） | | | | | | | |

换一组数据量之后再填一次：

| 数据量（页） | B | 趟数 | 实测 I/O | 公式 I/O | 差值 |
|---|---|---|---|---|---|
| | | | | | |
| | | | | | |

## 七、思考题

1. 已知 N = 500 页、B = 3，请手算：阶段 1 生成多少个顺串？归并需要几趟？总 I/O 是多少页？
2. 归并路数为什么是 B−1 而不是 B？如果强行用 B 路归并会发生什么？
3. 上表中 B=9 和 B=17 的实测 I/O 都是 3000 页，为什么内存几乎翻倍了代价却没变？
4. 如果排序结果直接流式交给上层算子（最后一趟输出不落盘），总 I/O 能省多少？请以 N=500、B=5 为例计算。
5. 假设页大小 8 KB，排序可用内存按 PostgreSQL `work_mem` 默认值 4 MB 计算：缓冲页数 B 是多少？在"只需 2 趟"（即 P = 2）的前提下，最多能排多少页、约合多少 GB 的数据？
6. 阶段 1 如果用"替换选择"（replacement selection）代替简单的"装满 B 页再排序"，顺串的平均长度会变成多少？这对总趟数有什么影响？
7. 输入数据本身已经有序时，外部归并排序的代价是多少？数据库里有什么办法能少做这次排序？
8. 写出 B=2 时的趟数与总 I/O 表达式；此时 `max(2, B−1)` 这个兜底为什么是必要的？

## 八、思考题答案与解析

**第 1 题**

⌈N/B⌉ = ⌈500/3⌉ = 167 个顺串。fan-in = B−1 = 2。
归并趟数 = ⌈log₂ 167⌉ = ⌈7.38⌉ = 8 趟（167→84→42→21→11→6→3→2→1）。
总趟数 P = 1 + 8 = 9。
总 I/O = 2N × P = 2 × 500 × 9 = **9000 页**。与实测表第一行一致。

**第 2 题**

归并时每个输入顺串都要占一页输入缓冲，否则无法预读、无法做 k 路堆归并。剩下的一页必须留给输出：输出页写满才能整页落盘，否则就要对磁盘做小于一页的随机写，代价更高甚至硬件不支持。所以 fan-in = B−1。若强行用 B 路归并，就没有输出缓冲区，输出只能"来一条写一条"，实际会退化为大量小块写，性能远差于整页写。

**第 3 题**

因为趟数是`向上取整`的对数，属于阶梯函数：
B=9：⌈500/9⌉ = 56 个顺串，⌈log₈ 56⌉ = ⌈1.94⌉ = 2 → P = 3；
B=17：⌈500/17⌉ = 30 个顺串，⌈log₁₆ 30⌉ = ⌈1.23⌉ = 2 → P = 3。
两者都停在 3 趟，所以总 I/O 都是 2 × 500 × 3 = 3000 页。结论：加内存要加到"跨过台阶"才有收益，这就是调 `work_mem` 时常见的"加了没效果"现象。

**第 4 题**

B=5 时 P = 5，总 I/O = 5000 页。最后一趟是纯输出，若流式消费掉，就省掉最后一趟的 N 次写 = 500 页。
省后代价 = 5000 − 500 = **4500 页**，节省 10%。一般形式：总 I/O = 2N×P − N（最后一趟不落盘）。

**第 5 题**

B = 4 MB ÷ 8 KB = 4096 KB ÷ 8 KB = **512 页**。
P = 2 要求 ⌈log_{B−1} ⌈N/B⌉⌉ = 1，即 ⌈N/B⌉ ≤ B−1 = 511，所以 N ≤ 512 × 511 = **261,632 页**。
折算：261,632 × 8 KB = 2,093,056 KB ≈ 2044 MB ≈ **2 GB**。
也就是说 4 MB 的 `work_mem` 大约能把 2 GB 的数据在两趟内排完（这是纯 I/O 模型下的上界，实际还要扣掉页表等开销）。

**第 6 题**

替换选择（replacement selection）用一个大小等于内存容量的堆持续吐出最小元素，平均能生成长度约为 **2B 页**的顺串（简单切块法生成的顺串长度是 B 页）。
顺串数从 ⌈N/B⌉ 降到约 ⌈N/(2B)⌉，归并趟数变为 1 + ⌈log_{B−1} ⌈N/(2B)⌉⌉。
以 N=500、B=9 为例：顺串从 56 降到约 28，⌈log₈ 28⌉ = ⌈1.60⌉ = 2，仍是 3 趟——所以替换选择省的是"顺串数量"，能否少一趟还要看有没有跨过台阶。

**第 7 题**

若输入已有序，阶段 1 可以只做一次**验证扫描**（检查相邻记录是否有序）：读 N 页、不写，代价 N。若确认有序就直接交给上层，代价接近 **N 页**（只扫描一遍）。
数据库里真正的规避办法是：让数据一开始就按需要的顺序存放（聚簇索引 clustered index / 索引组织表），或者让上层算子使用索引扫描（index scan）直接拿到有序流，从而不做排序。

**第 8 题**

B=2 时 ⌈N/B⌉ = ⌈N/2⌉ 个顺串，fan-in 用兜底值 2：
P = 1 + ⌈log₂ ⌈N/2⌉⌉，总 I/O = 2N × P。
以 N=500 为例：⌈500/2⌉ = 250，⌈log₂ 250⌉ = 8，P = 9，总 I/O = 9000 页。
`max(2, B−1)` 是必要的：B=2 时 B−1 = 1，若 fan-in 取 1，归并循环会"每次只合并 1 个顺串"，顺串数量永不减少，程序陷入死循环。

## 九、扩展任务（选做）

1. **替换选择（replacement selection）**：把步骤 4 的阶段 1 换成堆式替换选择，统计生成的顺串个数与平均长度，与"装满再排"对比。
2. **最后一趟不落盘**：实现流式输出版本，实测总 I/O 是否比公式少 N 页。
3. **随机 I/O 对比**：在计数模型里区分顺序页与随机页（顺序 1 次、随机按 `random_page_cost / seq_page_cost` 折算），观察 B 很小时随机 I/O 占比的变化。
4. **真实落盘**：用 `system/sorting.py` 的 `ExternalMergeSort` 配合 `ExternalQueueFactory`，把 100 万条数据真写到磁盘文件上排序，用系统监视器观察磁盘写入量。
5. **双缓冲（double buffering）**：思考如何在 B 页中预读下一块，让 CPU 排序与磁盘 I/O 重叠。这是真实 DBMS 提升外排序吞吐的常用手段，可用两个线程模拟。

## 十、常见错误与排查

| 现象 | 原因 | 处理 |
|---|---|---|
| 实测 I/O 与公式差 500 或 1000 页 | 最后一块数据不足 B 页，向上取整方式与公式假设不同 | 统一用 `math.ceil(len(chunk)/PAGE_SIZE)` 计数，公式里 N 也取 `ceil(总条数/页大小)` |
| 程序卡死不退出 | B=2 时 fan-in = B−1 = 1，每次只"合并"1 个顺串，数量不减少 | 归并路数写成 `max(2, B - 1)` |
| 排序结果不对 | 阶段 1 的切片排序后忘了把结果放回 `runs`，或用了原地排序的返回值（Python 的 `list.sort()` 返回 `None`） | 用 `chunk.sort()` 后 `append(chunk)`，不要用 `runs.append(chunk.sort())` |
| 归并后结果长度变少 | 手写的 k 路归并里有某一路提前结束就被丢弃 | 用 `heapq.merge(*group)`，k 路归并必须处理"某一路耗尽"的情况 |
| `assert out == sorted(data)` 失败但看不出哪里错 | 造数据时用了可变对象并在多处复用同一引用 | 每次 `external_merge_sort(list(data), ...)` 传入 `data` 的拷贝 |
| 加内存（调大 B）后 I/O 完全不变 | 趟数是阶梯函数，没跨过台阶 | 打印每一步的顺串数，确认 `⌈log_{B−1}⌉` 是否真的变小 |
| 选做部分 `ModuleNotFoundError: No module named 'system'` | 运行目录不在仓库根，或仓库根没进 `sys.path` | 在仓库根目录运行，或照步骤 8 的代码把仓库根插入 `sys.path` |
| 中文标签在图里显示为方框 | 缺中文字体 | 设置 `plt.rcParams["font.sans-serif"]`，或用英文标签 |