【分布式 OLAP 查询引擎】Volcano 迭代器模型：Pull 语义与 Pipeline Breaker

物理计划生成之后，执行层如何把算子树跑起来？第 1 篇 把执行模型列为 Volcano pull、向量化 batch、morsel-driven 三条路径；本篇钉住 Volcano（迭代器）模型——几乎所有关系型执行器的共同祖先，也是理解 Trino Operator 与 PostgreSQL ExecProcNode 的对照基准。

核心问题有三：

1. Pull 语义 下 parent 如何驱动 child 产出行（或 Page）？
2. 哪些算子会 打断 pipeline，必须先吃完全部输入才能输出？
3. 分布式 OLAP（Trino）在 Volcano 骨架上做了哪些 批量化改造？

环境说明：Trino 476+ 机制来自官方文档与 trinodb/trino 源码（Operator.java、Driver.java）；PostgreSQL 对照来自 PG 执行器篇 与 execMain.c。本机未部署 Trino 集群，不粘贴未实测的 EXPLAIN ANALYZE。

一、Volcano 接口：Open / Next / Close

Graefe 等在 Volcano — An Extensible and Parallel Query Evaluation System（1990）提出：每个算子实现统一迭代器接口，parent 拉取（pull） child 的数据，child 对 parent 隐藏内部实现。经典三元组：

flowchart TB
  ROOT["Root (e.g. Output)"]
  JOIN["Join"]
  SCAN_O["Outer Scan"]
  SCAN_I["Inner Scan"]
  ROOT -->|"next()"| JOIN
  JOIN -->|"next()"| SCAN_O
  JOIN -->|"next()"| SCAN_I

Pull 的含义：Join.next() 内部决定何时调用 OuterScan.next() 与 InnerScan.next()；Scan 不主动推送。Volcano 论文还引入 Exchange 算子在多进程间传递迭代器结果，为 MPP 扩展留接口（Trino 的 ExchangeOperator 属同一谱系，第 11 篇 展开）。

1.1 与 Push 的对比

OLAP 向量化引擎常在算子 内部 批处理，对外仍保留「拉一批」的控制流（第 7 篇）。

二、Pipeline 与 Pipeline Breaker

Pipeline 指一组可 同步、流式 连接的算子：上游产出一行（或一批），下游立刻消费，无需缓冲全表。

Pipeline breaker（管道阻断算子） 必须先 完整消费 一侧或全部输入，才能向 parent 输出第一行。常见 breaker：

flowchart LR
  subgraph pipe ["Streaming pipeline"]
    S1["Scan"] --> F1["Filter"]
    F1 --> P1["Project"]
  end
  subgraph break ["Pipeline breaker"]
    P1 --> H["Hash Build"]
    H --> J["Hash Probe"]
  end
  J --> OUT["Output"]

Sort 之上、Sort 之下的 Scan–Filter–Project 仍可流水线并行；Sort 本身把 pipeline 切成两段。EXPLAIN 里 Stage 边界常与 breaker + Exchange 重合（第 10 篇）。

2.1 表达式求值在 Volcano 中的位置

每行（或每批）经过 Filter / Project 时要求值表达式。PostgreSQL 用 EEO opcode 数组 解释执行（见 PG 执行器篇）；Trino 在 Page 上对 Block 做批量求值。共同点：

• Constant folding 在 planner 完成，执行期不再重复算常量。
• Short-circuit：AND/OR 在逐行模型下可跳过；向量化下用 selection vector 压缩活跃行（第 7 篇）。
• NULL 语义：三值逻辑；SQL 标准 UNKNOWN 在 filter 中等价于丢弃行。

三、PostgreSQL：ExecutorRun 对照

PostgreSQL 是 Volcano 的行存教科书实现。入口在 src/backend/executor/execMain.c：

ExecProcNode 是 函数指针——每个 plan node 注册自己的实现（nodeSeqscan.c、nodeHashjoin.c 等）。Parent 不感知 child 类型，符合 Volcano 的 封装 原则。

Hash Join 在 PG 中的两阶段（A 级：nodeHashjoin.c）：

1. Build：对 inner 子计划循环 ExecProcNode，插入 hash 表。
2. Probe：对 outer 每行 probe；outer join 用标记位处理未匹配行。

这与 Trino 的 HashBuilderOperator + LookupJoinOperator 分工同构（第 9 篇）。

sequenceDiagram
  participant ER as ExecutorRun
  participant HJ as HashJoin
  participant Inner as Inner Scan
  participant Outer as Outer Scan
  ER->>HJ: ExecProcNode
  Note over HJ,Inner: Build phase
  loop until NULL
    HJ->>Inner: ExecProcNode
  end
  Note over HJ,Outer: Probe phase
  loop until NULL
    HJ->>Outer: ExecProcNode
    HJ->>HJ: probe hash table
  end

OLTP 约束：PG 默认 逐行 Volcano，兼容 MVCC 与复杂 trigger；OLAP 引擎为吞吐牺牲部分逐行灵活性，转向 batch（columnar-engine 向量化篇）。

四、Trino 476+：Operator 与 Driver

Trino worker 上，逻辑 plan fragment 经 LocalExecutionPlanner 变为 Operator 链，由 Driver 调度（A 级：core/trino-main/.../operator/Driver.java，Trino Documentation Developer guide）。

4.1 Operator 接口

Operator（io.trino.operator.Operator）核心方法：

Source 算子（如 TableScanOperator、PageSourceOperator）只实现 getOutput()；addInput 抛异常（A 级：PageSourceOperator.java）。

4.2 Driver：算子间的 Page 搬运

Driver.process() 从 pipeline 底端 向顶端迭代：若 current.getOutput() 非空且 next.needsInput()，则 next.addInput(page)。注释中明确 HashBuild / LookupJoin 等算子可能 break pipeline，需特殊处理（Driver.java）。

这与纯 Volcano 的「只调 next」不同：中间算子之间是 push Page，但 整体仍由 Driver 循环驱动，Source 仍是被动的。

flowchart TB
  D["Driver.process() loop"]
  TS["TableScanOperator<br/>getOutput()"]
  FIL["FilterOperator<br/>addInput / getOutput"]
  PROJ["ProjectOperator"]
  OUT["OutputOperator"]
  D --> TS
  TS -->|"Page"| FIL
  FIL --> PROJ
  PROJ --> OUT

Trino 的数据单元是 Page（若干列的 Block，默认最多 4096 行，版本相关），不是单行——这是相对 Graefe 原模型的 批量化 Volcano（第 7 篇）。

4.3 Task、Driver 与并行度

一个 Task 含多条 Driver（由 LocalExecutionPlan 的 DriverFactory 创建），每条 Driver 跑一条 operator 链；task.concurrency 控制并行 Driver 数。Scan 侧按 Split 分配不同 Driver（第 8 篇）。

五、Volcano 在 OLAP 中的局限与演进

Graefe 论文的价值在于 统一接口 + 可组合算子；现代引擎保留组合性，把 next() 里的「一行」换成「一批」。

六、阅读与排查时的用法

1. EXPLAIN 自底向上读：Leaf 多为 Scan；向上遇到 Hash、Sort、Exchange 标记 breaker 或 stage 切分。
2. 慢在 Scan 还是 breaker：EXPLAIN ANALYZE 比较 Scan/Filter CPU 与 HashBuild/OrderBy wall time（Trino 476+ 文档 EXPLAIN ANALYZE）。
3. 与 PG 对照：PG Seq Scan 行式 pull；Trino TableScan Page pull + connector 异步 isBlocked()。

七、边界与分工

• 不写 SIMD 内核与 Block 编码细节 → 第 7 篇。
• 不写 Split 与下推 → 第 8 篇。
• 不写 MPP shuffle 协议 → 第 11 篇。

参考资料

系列导航：系列索引 · 上一篇：Join 与物理算子选择 · 下一篇：向量化与 morsel-driven 执行