【分布式 OLAP 查询引擎】向量化批处理与 Morsel-Driven 并行

第 6 篇 里的 Volcano 逐行 next() 在 OLAP 扫描百万行时，函数调用与分支预测成本会压过磁盘带宽。columnar-engine 第 4 篇 从 ClickHouse Block 与 IProcessor 讲了 存储对齐的执行 batch；本篇从 分布式 OLAP 查询引擎 视角回答：Trino、DuckDB 如何把「一批列向量」嵌进执行框架，morsel-driven 并行与 Page 流如何分工。

一、为什么必须批量化

Boncz 等 MonetDB/X100: Hyper-Pipelining Query Execution（CIDR 2005）论证：列存 OLAP 若仍逐行 dispatch，CPU 大量耗在 interpretation overhead 而非数据运算。向量化执行的核心等式：

\[ \text{吞吐} \approx \frac{\text{每批行数} \times \text{批处理效率}}{\text{每批固定开销}} \]

本系列 Trino 主线采用 Page 级向量化；DuckDB 采用 DataChunk + morsel；Spark 3.x 在 Whole-Stage Codegen 与 columnar batch 间折中（第 13 篇）。

flowchart LR
  VOL["Row Volcano<br/>next() x N"]
  VEC["Vectorized<br/>batch x N/k"]
  VOL -->|"OLAP"| VEC

二、列向量 batch 的结构

2.1 逻辑布局

一批数据在内存中按 列连续 存放（与 columnar-engine 列存基础 一致）：

Filter 理想路径：对条件列向量化求值 → 生成 selection → 后续算子只处理选中行（ClickHouse IColumn::filter；DuckDB Vector 文档）。

2.2 Flat vs Dictionary 编码

Dictionary 上 filter 可先 扫字典 再映射索引，跳过大量重复解码（与 lakehouse Parquet 篇 字典页呼应）。Connector 读湖时若保留 dictionary block，Filter 算子需识别编码（第 8 篇）。

2.3 与 columnar-engine/04 的对照

三者在 「列数组 + 批算子」 上同构，差异在 分布式 Page exchange（Trino）与 存储耦合深度（ClickHouse MergeTree）。

三、Trino：Page 与 Operator 链

Trino worker 上算子间传递 Page（A 级：Trino Documentation Developer guide；io.trino.spi.block.Block）。

Block 类型包括 LongArrayBlock、IntArrayBlock、VariableWidthBlock、DictionaryBlock 等；DictionaryBlock 即字典编码列在引擎内的表示。

flowchart TB
  PS["PageSource<br/>getOutput()"]
  FIL["FilterOperator"]
  PRJ["ProjectOperator"]
  AGG["AggregationOperator"]
  PS -->|"Page"| FIL
  FIL --> PRJ
  PRJ --> AGG

向量化点：FilterOperator 对整列生成 mask 或 compacted Page；ProjectOperator 调用 ExpressionCompiler 生成的 batch 函数。相对 第 6 篇 的行 Volcano，Trino 在 Operator 内部 完成批处理，Driver 仍按 Page 推拉。

MPP 边界：跨 Worker 时 Page 序列化经 Exchange（第 11 篇）；batch 越大，shuffle 压缩比与延迟权衡越敏感。

四、Morsel-Driven 并行（DuckDB 主线）

Morsel-driven（Leis et al., Morsel-Driven Parallelism，2014）把表划分成固定大小 morsel，线程动态领取 morsel 执行 pipeline，配合 work-stealing 平衡负载。

DuckDB 执行模型（A 级：DuckDB Documentation Internals / Execution）：

flowchart TB
  T1["Thread 1<br/>morsel 0-2047"]
  T2["Thread 2<br/>morsel 2048-4095"]
  T3["Thread 3<br/>steal morsel ..."]
  SCAN["TableScan"]
  SCAN --> T1
  SCAN --> T2
  SCAN --> T3

与 Trino 按 Split 开多 Driver 类似：二者都是 shared-nothing 线程 + 划分输入，区别是 DuckDB 在 单进程 内完成，Trino 在 多 Worker 上还要考虑 split locality（第 10 篇）。

NUMA 感知：DuckDB 文档描述 morsel 调度可结合 NUMA-local 分配；Trino 依赖 OS 与 connector 的 split host 提示，不保证 NUMA 最优。

五、SIMD 与引擎边界

向量化 不等于 手写 AVX intrinsic 遍布源码。常见分层：

1. 编译器 auto-vectorize tight loop（filter、算术）。
2. 运行时选 kernel（DuckDB UnifiedVectorFormat 路径）。
3. 专用库（如 xsimd、libdivide 用于 hash）。

引擎责任是保证 内存对齐、连续访问、少分支；是否生成 SIMD 指令取决于编译器与类型宽度。不在此篇给出未标注来源的「加速比」。

六、本机 DuckDB 1.5.4 实验

脚本：post/db/query-engine/reproduce/run_duckdb_experiments.py。数据：tpch.orders 100 000 行、tpch.customer 5 000 行（内存生成，非官方 TPCH 文件）。

环境：WSL2（Linux 6.6.87.2）、i9-12900K / 32 GiB、DuckDB 1.5.4（Python 绑定）。

查询 q_join（customer ⋈ orders，日期 filter + GROUP BY + LIMIT 10）：

cd post/db/query-engine/reproduce
python3 run_duckdb_experiments.py

解读（机制，非跨引擎排名）：

• Scan 以 vectorized batch 读内存表；Join 走 HASH_JOIN，与 第 9 篇 build/probe 模型一致。
• 100k 行规模下全内存，未观测 spill；更大 fact 或更低 memory_limit 才会触发外排 hash（DuckDB 文档 Optimization）。

同脚本的 q_filter（count(*) + 等值谓词）叶算子为 SEQ_SCAN + filter pushdown 到 scan，可用于对照 第 8 篇 的谓词位置。

七、三种执行语义并排

八、边界

• ClickHouse MergeTree 读路径与 PREWHERE → columnar-engine 第 5 篇。
• Hash join / agg 内存与 spill → 第 9 篇。
• Spark Tungsten 二进制 row → 第 13 篇。

参考资料

系列导航：系列索引 · 上一篇：Volcano 迭代器模型 · 下一篇：Scan、Filter 与 Project