实验 13：在线聚合——不等扫完就给出答案

> `SELECT max(b) FROM stuff` 要扫全表才知道确切结果，但用户往往等不了那么久：本实验实现"边扫边报数"的在线聚合（online aggregation），给运行中的估计值配上置信区间，并弄清它对哪种聚合有效、对哪种无效。

一、实验目标

完成本实验后，你应当能够：

1. 能解释全量聚合（full-blown aggregation）与在线聚合的区别，说出在线聚合原始论文的出处（Hellerstein、Haas、Wang，SIGMOD 1997）。
2. 能用 Python 生成器（`yield`）把一次性聚合改造成逐行产出运行结果的流式聚合。
3. 能实现带 95% 置信区间的均值型在线估计，并验证置信区间宽度按 1/√m 收缩（m 为已扫描行数）。
4. 能说明扫描顺序（随机序与有序序）对在线估计正确性的影响。
5. 能区分在线聚合与两类"长得像"的技术：DuckDB 的查询进度条（只报完成百分比与 ETA，不做统计推断）和 BigQuery 的 APPROX_* 函数（基于 sketch 的近似计算，另一条技术路线）。

二、预备知识

2.1 聚合的两副面孔

```sql
SELECT max(b) FROM stuff;
```

全量聚合的做法：从第一行扫到最后一行，维护当前最大值，扫完才返回。行数 n = 2×10⁵ 时机器毫秒级就能扫完，但数据到 10⁹ 行、或扫描本身要读盘要网络时，用户可能盯着黑屏等几分钟。在线聚合的做法：**每扫过一部分就给出当前答案，随扫描进行不断修正**，让用户扫到 10% 时就能对最终结果有个靠谱的判断——是"再等等"，还是"答案已经够用，直接停"。

2.2 逐行产出：Python 生成器

普通函数 `return` 一次给全部结果；**生成器函数（generator）**用 `yield` 逐个吐出中间结果，调用方拿到的是可以边循环边取的迭代器。这正是"边扫边报"的最小实现骨架：

```python
def online(xs):
state = None
for x in xs:
state = update(state, x) # 每处理一行更新一次状态
yield state # 立刻把当前状态交给调用方
```

聚合函数大多只需 O(1) 状态：MAX 记一个当前最大值，SUM 记一个累加和，AVG 记 (和, 计数)。所以"全量聚合"与"在线聚合"的差别不在算什么，而在**何时把结果交出去**。

2.3 均值型估计的置信区间

对"平均值"这类聚合，扫到第 m 行时的估计就是这 m 行的样本均值：

```text
x̄ₘ = (x₁ + x₂ + … + xₘ) / m
```

只要这 m 行是总体的一个随机样本，统计学的中心极限定理告诉我们：估计误差大致服从正态分布，标准误差（standard error）为

```text
SE = s / √m
```

其中 s 是这 m 个样本的样本标准差（总体标准差未知时用它代替），m 是已扫描行数。95% 置信区间（confidence interval，CI）取：

```text
[x̄ₘ − 1.96·SE, x̄ₘ + 1.96·SE]
```

1.96 是标准正态分布 95% 分位数。区间宽度的近似式是本实验的核心公式：

```text
W(m) ≈ 2 × 1.96 × s / √m
```

每个符号：W 单位与数据相同；s 与数据同单位；m 是行数（无量纲）。要点：**W 与 √m 成反比**。宽度减半需要 m 扩大到 4 倍；从扫 1% 到扫 100%，宽度只缩到 1/10。误差下降是越到后面越"贵"的。

2.4 MAX 是另一回事

均值有 SE 公式可用的前提是"部分样本能代表总体"。MAX 没有这种代表性质：扫过的最大值永远不大于真值（只会往上爬），扫到 99% 时可能还差一个藏在最后 1% 里的巨值。不给分布假设，就无法为 MAX 写出统计意义上的置信区间。所以本实验两条线分开走：

• **MAX**：运行值是确定的下界，随扫描单调不减、最后一步跳到真值；它给出"至少有多大"，不承诺"有多大把握"。

• **AVG**：运行值 + 置信区间，可以量化"有多大的把握、误差多大"，从而支持"提前停止"。

2.5 扫描顺序为什么必须是随机的

CI 公式把已扫描的 m 行当作随机样本。如果表恰好按 b 降序存放（例如刚做过建索引的排序），扫前 10% 拿到的全是最大值，估计严重偏高；升序则严重偏低。所以在线聚合要求**随机顺序扫描（random-order scan）**，或在读取时打乱。物理布局不受你控制时，这一点要在心里打个问号——本实验会专门用一组对照验证它。

三、实验原理

3.1 在线聚合的三要素

按 Hellerstein、Haas、Wang 1997 年的原始论文《Online Aggregation》（SIGMOD 1997），一个在线聚合系统要提供：

1. **运行估计（running estimate）**：随扫描不断更新的当前答案。
2. **置信区间（confidence bound）**：当前答案可能偏离真值的范围及置信水平。
3. **停止规则（stopping rule）**：当区间宽度小于用户要求的精度时提前结束扫描。

本实验用生成器实现第 1 项，用 2.3 节公式实现第 2 项，第 3 项留成实验步骤里的一个判断。

3.2 参考实测数据

本机实测（CPython 3.13 + Windows；n = 200000，Gamma 分布数据，`default_rng(0)` 固定种子，随机顺序扫描，估计对象为均值型聚合，95% CI；由本节步骤 5–6 的代码逐位复现）：

| 扫描进度 m/n | 运行估计 | 备注 |
| --- | --- | --- |
| 10% | 50.4435 | 与真值相差 0.344 |
| 100% | 50.0996 | 即总体真值 |
| 100% | — | 95% CI 宽度 0.3102 |

由 2.3 节公式反推总体标准差：W = 2 × 1.96 × s/√m，取 W = 0.3102、m = 200000，得

```text
s ≈ 0.3102 × √200000 / 3.92 ≈ 35.4
```

均值约 50、标准差约 35 的分布，实验代码用 Gamma 分布（形状参数 shape = 2、尺度参数 scale = 25，理论均值 50、标准差 √(2×25²) ≈ 35.4）生成。上表数字由步骤 5 的固定种子（`default_rng(0)`）实测得出，逐位可复现；换随机种子或改用 notebook 的 Faker 数据后，具体数字会略有差异，判断标准是"数量级与趋势"。

区间宽度的收缩验证：W(m) ∝ 1/√m，故

```text
W(20%) ≈ W(100%) / √0.2 ≈ 0.3090 / 0.447 ≈ 0.69
W(10%) ≈ W(100%) / √0.1 ≈ 0.3090 / 0.316 ≈ 0.98
```

扫到 10% 时区间宽约 1，扫到 100% 时收窄到 0.31——比例 1/√10。把你的实测填进观测表后应能直接看出这一关系。


3.3 与两类"长得像"的技术的边界

• **DuckDB 的进度条**：打印已完成的百分比和预计剩余时间（ETA）。它不改变估计的统计性质，也不做抽样推断——是"告诉你还要等多久"，不是"提前告诉你答案大概是什么"。在线聚合三要素里它一项都不占。

• **BigQuery 的 APPROX_* 函数**（如 APPROX_COUNT_DISTINCT）：基于 sketch（概要数据结构）的一次遍历近似，牺牲精度换速度，但**结果不随时间流式细化、通常不附带置信区间**。它回答"能不能算得更快"，在线聚合回答"能不能边算边信"。两者是不同的路线，文献里注意别混引。

3.4 提前停止值多少

设扫描速度恒定，扫完需要时间 T。若用户只需"95% 置信下宽度不超过 0.5"的答案，由 W(m) = W(n)·√(n/m) 反解：

```text
m = n × (W(n) / 0.5)² = 200000 × (0.3090 / 0.5)² ≈ 76000 行 ≈ 全表的 38%
```

也就是约 62% 的扫描时间被省掉，代价是答案带一个明确的误差声明。这个"精度换时间"的交易是在线聚合的实际价值所在；本实验最后一步会把它做成可运行的停止规则。

四、实验环境与准备

1. 进入仓库根目录 `explaindb-main`。
2. 安装依赖：

```bash
pip install faker numpy matplotlib jupyter
```

3. 打开配套 notebook：

```bash
jupyter notebook notebooks/Online-Aggregation.ipynb
```

4. notebook 用 40 行小数据演示两种算法与单元测试；第五章在 20 万行规模上加入置信区间与扫描顺序对照。需要 Python 3.10 及以上。

五、实验步骤

步骤 1：造数据（对应 notebook 的小规模版）

```python
from dataclasses import dataclass

@dataclass
class Stuff:
a: int
b: int

小规模演示（与 notebook 一致的规模）

import random
rng = random.Random(42)
data_small = [Stuff(rng.randrange(1000), rng.randrange(2000)) for _ in range(40)]
print(data_small[:5])
```

步骤 2：全量聚合（对应 notebook 算法 1）

```python
import math

def aggregate(rows, attribute):
"""一次性聚合：扫完才返回结果。"""
current_max = -math.inf
for el in rows:
current_max = max(current_max, el.__dict__[attribute])
return current_max

print(aggregate(data_small, "b"))
```

步骤 3：在线聚合（对应 notebook 算法 2）

```python
def online_aggregation(rows, attribute):
"""逐行产出运行最大值：每处理一行 yield 一次。"""
current_max = -math.inf
for el in rows:
current_max = max(current_max, el.__dict__[attribute])
yield current_max

snaps = online_aggregation(data_small, "b")
for i, v in enumerate(snaps, 1):
if i in (1, 10, 20, 40):
print(f"扫过 {i:>2} 行: max = {v}")
```

注意 `online_aggregation` 的函数体与 `aggregate` 几乎逐行相同，只把 `return` 换成了循环内的 `yield`。运行值只升不降，最后一行跳到真值。

步骤 4：单元测试确认两法等价

```python
import unittest

class TestAggregation(unittest.TestCase):
def test_end_to_end(self):
for trial in range(10):
r = random.Random(trial)
rows = [Stuff(r.randrange(20), r.randrange(20)) for _ in range(1000)]
res1 = aggregate(rows, "b")
*_, res2 = online_aggregation(rows, "b") # 取生成器最后一个值
self.assertEqual(res1, res2)

unittest.main(argv=["ignored", "-v"], defaultTest="TestAggregation", exit=False)
```

步骤 5：放大到 20 万行，构造随机顺序扫描

```python
import numpy as np

rng = np.random.default_rng(0)
N = 200_000

Gamma(2, 25)：均值 50，标准差约 35.4，与参考实测同分布

values = rng.gamma(shape=2.0, scale=25.0, size=N)

order = rng.permutation(N) # 随机扫描顺序：在线估计正确性的前提
scan = values[order]

true_mean = values.mean() # 全量聚合的真值（见 3.2 节参考实测表：50.0996）
print("真值:", round(true_mean, 4))
```

步骤 6：带置信区间的在线均值估计

```python
import math

def online_mean_with_ci(xs, checkpoints=(0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 1.0)):
"""边扫边报：运行均值、样本标准差、95% CI 及其宽度。
W(m) = 2 × 1.96 × s / √m，随 m 按 1/√m 收缩。"""
n = len(xs)
total = 0.0
marks = {max(1, int(c * n)) for c in checkpoints}
for m, x in enumerate(xs, 1):
total += x
if m in marks:
mean = total / m
s = xs[:m].std(ddof=1) # 样本标准差
half = 1.96 * s / math.sqrt(m)
print(f"m = {m:>6} ({m/n:>4.0%}) 估计 = {mean:.4f} "
f"95% CI = [{mean - half:.4f}, {mean + half:.4f}] 宽度 = {2*half:.4f}")

online_mean_with_ci(scan)
```

对照 3.2 节：m = 10% 时估计与真值相差约 0.34、宽度约 1；m = 100% 时宽度收窄到 0.31 附近（参考实测：10% 估计 50.4435，100% 估计 50.0996、宽度 0.3102）。

步骤 7：验证 1/√m 收缩

```python
import math

widths = {}
total = 0.0
for m, x in enumerate(scan, 1):
total += x
if m in (2_000, 20_000, 50_000, 200_000):
widths[m] = 2 * 1.96 * scan[:m].std(ddof=1) / math.sqrt(m)

print("m 宽度 宽度×√m")
for m, w in widths.items():
print(f"{m:>7} {w:>8.4f} {w * math.sqrt(m):>8.2f}")
```

理论预言最后一列近似常数（都约等于 2 × 1.96 × s）。若你的四行最后一列彼此接近，就实测验证了 W ∝ 1/√m。

步骤 8：有序序对照（反例）

```python
bad_scan = np.sort(values) # 升序扫描
print("升序扫描，m = 10% 时估计:", round(bad_scan[:20_000].mean(), 4), "（真值约 50）")
print("降序扫描，m = 10% 时估计:", round(np.sort(values)[::-1][:20_000].mean(), 4))
```

升序时前 10% 全是最小的值，估计远低于真值且 CI 完全失真——区间"罩不住"真值。这说明 2.5 节的随机性不是可选项。

步骤 9：提前停止规则

```python
import math

TARGET_WIDTH = 0.5 # 用户要求的精度：95% CI 宽度 ≤ 0.5
total, stopped_at = 0.0, None
for m, x in enumerate(scan, 1):
total += x
if m % 1000 == 0:
mean = total / m
w = 2 * 1.96 * scan[:m].std(ddof=1) / math.sqrt(m)
if w <= TARGET_WIDTH:
stopped_at = m
break

print(f"在第 {stopped_at} 行（{stopped_at/N:.0%}）停止，"
f"估计 {total/stopped_at:.4f}，真值 {true_mean:.4f}")
```

参考 3.4 节的推算，停止点应在全表 38% 附近。省下的时间就是在线聚合卖给你的东西。

六、观测记录表

**表 1：MAX 的运行轨迹（把每次 yield 的值填进表格，记 5 个时点）**

| 已扫描比例 | 运行最大值 | 与真值之差 |
| --- | --- | --- |
| 10% | | |
| 25% | | |
| 50% | | |
| 75% | | |
| 100% | | |

**表 2：均值的在线估计（随机顺序扫描）**

| m/n | 运行估计 | 95% CI 下界 | 95% CI 上界 | CI 宽度 | 宽度 × √m |
| --- | --- | --- | --- | --- | --- |
| 10% | | | | | |
| 20% | | | | | |
| 50% | | | | | |
| 100% | | | | | |

**表 3：扫描顺序对照（m = 10% 时的估计）**

| 扫描顺序 | 运行估计 | 偏离真值多少 |
| --- | --- | --- |
| 随机顺序 | | |
| 升序 | | |
| 降序 | | |

**表 4：提前停止（TARGET_WIDTH 从 1.0 收紧到 0.3）**

| 目标宽度 | 停止位置 m/n | 运行估计 | 真值 |
| --- | --- | --- | --- |
| 1.0 | | | |
| 0.5 | | | |
| 0.3 | | | |

七、思考题

1. `aggregate` 与 `online_aggregation` 的函数体几乎一样，只差 `yield`。从内存占用与首结果延迟（time to first result）两个角度比较两者。
2. MAX 的运行值单调不减、且永远不大于真值。为什么它天然不需要置信区间也可以用？用户看到"运行值 1937 且 5 分钟没变"能推出什么、不能推出什么？
3. 由 W(100%) = 0.3090 反推数据的样本标准差；再把宽度压到 0.1 需要扫多少行？在 n = 200000 的表上做得到吗？这说明什么？
4. 为什么升序扫描会让置信区间失效，而不是仅仅"估计偏低"？提示：CI 公式的前提是哪一条被破坏了？
5. 若把 95% 换成 99%，分位数 1.96 要换成约 2.58。同样 20 万行，CI 宽度变为多少倍？精度与把握之间你如何取舍？
6. DuckDB 查询进度条与在线聚合的本质区别是什么？如果一个系统只输出"已完成 37%"，用户能对最终结果做出什么判断、不能做出什么判断？
7. BigQuery 的 APPROX_COUNT_DISTINCT 与在线聚合都在"没扫完就要答案"，两者的交换物分别是什么？（提示：一个换的是精度不换时间结构，一个换的是时间结构。）
8. 停止规则实验里，目标宽度 0.3 时停止位置比理论推算（0.3090/0.3)² × n ≈ 106% 更晚甚至扫完了。解释为什么目标宽度接近 W(n) 时提前停止失去意义。

八、思考题答案与解析

1. 内存：两者都是 O(1) 状态（一个当前最大值），差别不在内存。首结果延迟：`aggregate` 必须扫完 n 行才有第一个也是唯一一个结果；`online_aggregation` 处理完第 1 行就产出第一个结果。流式接口还允许调用方在任意时点"止损"放弃扫描，这正是 3.4 节停止规则的接口基础。
2. 运行值是确定性的下界：无论剩下什么数据，真值只可能更大，所以"至少 1937"这个命题永远成立，不需要概率论。5 分钟没变只说明"没遇到更大的值"，不能推出"真值就是 1937"——除非对 b 的分布有假设（例如已知上界），否则最后 1% 藏一个巨值的可能性无法排除。
3. 由 W = 2 × 1.96 × s/√m：s = 0.3090 × √200000 / 3.92 ≈ 35.2。宽度压到 0.1 需要 m = (3.92 × 35.2 / 0.1)² ≈ 1.9 × 10⁶ 行，约为本表行数的 9.5 倍——在这张表上做不到，只能换更大的表或接受更宽的区间。这说明 1/√m 的收缩非常"贵"：精度要求每提高 10 倍，数据量要扩大 100 倍。
4. CI 公式的成立前提是"已扫描的 m 行构成随机样本"，于是样本均值围绕真值波动、波动幅度由 SE 刻画。升序扫描的前 m 行是最小的 m 个值，样本均值系统性偏低（偏差，不是方差），且这个偏差不随 m 增大而消失——扫到 99% 仍偏低。区间是以"无偏估计 ± 1.96·SE"为中心构造的，中心错了，区间整体错位，无论多宽都罩不住真值。这叫估计量有偏，偏差无法靠加宽区间弥补。
5. 宽度变为 2.58/1.96 ≈ 1.32 倍，即同样的数据把握提高 4 个百分点，代价是区间放宽三成。取舍取决于应用：报表初筛可要 90%（1.64）；涉及钱和承诺的数字要 99% 以上，或者干脆扫完拿精确值。
6. 进度条提供的是执行状态（扫到哪了），在线聚合提供的是结果的统计性质（答案大概是多少、误差多大）。看到"已完成 37%"，用户能判断"还要等多久"，不能判断"结果会不会和最终值差很多"；在线聚合的 CI 恰好补上后半句。原论文（Hellerstein、Haas、Wang，SIGMOD 1997）的三要素——运行估计、置信区间、停止规则——进度条一项都不具备。
7. APPROX_* 交换的是精度：一次遍历、常数内存给出有误差保证的近似值（sketch 路线），运行时间结构与精确计算相同（都是扫一遍），只是每步更便宜。在线聚合交换的是时间结构：扫描可以随时中断、答案随中断点越来越准（CI 越来越窄），单行成本未必更省。一句话：前者"算得快但不能边算边看"，后者"边算边看但未必算得快"。
8. W(n) = 0.3090 是全表扫完时的宽度下限。目标 0.5、1.0 都大于它，所以在扫完之前就能达标；目标 0.3 小于 W(n)，任何中途时刻都不可能达标，停止条件永不触发，退化成全表扫描。使用停止规则前要先算好理论可达精度，或让系统在"扫完仍不达标"时给出明确提示。

九、扩展任务（选做）

1. **Bootstrap 置信区间。** 把 2.3 节的正态近似换成 bootstrap：在每个检查点对已扫描的 m 个值有放回重采样 B = 1000 次，取样本均值的 2.5% 与 97.5% 分位数作为 CI。对比两种 CI 在 m = 10% 时的宽度与覆盖率，讨论小 m 时哪种更可信。提示：`np.random.default_rng().choice(xs[:m], size=(B, m), replace=True).mean(axis=1)`。
2. **分位数聚合的在线化。** 均值有 SE 公式，中位数没有闭式解。实现运行中位数（维护已扫描值的分位数，可用 `np.quantile`），用 bootstrap 给它配 CI，观察 m = 50% 前 CI 的行为。提示：这是"在线聚合能推广到哪类聚合函数"的入门题，天平会倒向样本量需求更大的一侧。
3. **多表连接下的在线聚合。** 原论文的核心贡献其实在连接（join）后的聚合。把两表内连接的结果流式化：一边流式产出匹配对、一边喂给步骤 6 的在线均值。观察连接顺序（小表build大表probe）对早期估计质量的影响。提示：不需要真的建哈希表，用两个 `defaultdict(list)` 模拟即可。

十、常见错误与排查

| 现象 | 原因 | 处理 |
| --- | --- | --- |
| 生成器第二次遍历没有内容 | 生成器只能消费一次，耗尽即空 | 需要重跑时重新调用 `online_aggregation(...)`，或改用 `list(...)` 缓存 |
| `*_, res2 = online_aggregation(...)` 报解包错误 | 数据为空，生成器一次也没 yield | 先确认输入非空；或给生成器加"空输入"分支 |
| CI 宽度不随 1/√m 收缩 | 扫描顺序没有随机化（用了有序数据） | 步骤 5 的 `rng.permutation` 必须执行；先跑步骤 8 的反例自查 |
| 估计值离真值很远且不收敛 | 数据里混入了极端值或分布不是预期分布 | 打印 `values.mean()`、`values.std()` 与理论值 50、35.4 对照 |
| 步骤 6 每个检查点都很慢 | `xs[:m].std(ddof=1)` 每次重扫前 m 个元素 | 教学规模（20 万行）可接受；提速可用 Welford 在线算法维护均值与方差 |
| 停止规则从未触发 | 目标宽度小于 W(n)（见思考题 8） | 把 `TARGET_WIDTH` 放宽到 0.5 以上，或增大 n |
| 99% CI 比参考实测宽很多 | 分位数写错（如写成 2.96） | 95% 用 1.96，99% 用约 2.58，90% 用约 1.64 |
| notebook 的 `faker` 数据与本实验数字对不上 | notebook 只有 40 行且是整数，不产生统计意义 | 40 行用于理解接口；统计实验用第五章的 20 万行 Gamma 数据 |