【分布式 OLAP 查询引擎】Trino 查询路径：从 SqlQueryExecution 到 Page 流

lakehouse 第 18 篇 从「引擎如何读湖」讲了 partition/file/row-group 四层下推漏斗；第 10 篇 讲了 Query/Stage/Task/Driver 五层 MPP 对象。中间缺的一环 是：Coordinator 上的 Java 代码 按什么顺序 调用 Analyzer、Planner、SplitScheduler，Iceberg 的 manifest 过滤 在哪一步 变成 Worker 上的 Scan Task？

本篇是本系列 Trino 主线，把一条 SELECT 从提交到 Worker Page 流串成可对照源码的时序。读完应能回答：

• SqlQueryExecution 与 QueryStateMachine 各管哪段生命周期？
• Distributed plan 如何切成 PlanFragment 并变成 Stage DAG？
• Iceberg 表从 snapshot → manifest filtering → ConnectorSplit 的路径，与 lakehouse 第 8 篇 的元数据树如何一一对应？
• Worker 上 Operator pipeline 如何 pull Page？EXPLAIN / EXPLAIN ANALYZE 哪些字段对应哪一层？

Shuffle 细节见 第 11 篇；Iceberg planner 侧下推全链路见 第 15 篇（与 lakehouse/18 对读）。

环境说明：本机 WSL2、无 JVM、Docker Compose 不可用。机制与类名锚定 Trino 476 文档与 trinodb/trino 源码；不粘贴本机未执行的 CLI 输出。Trino 476 文档 EXPLAIN ANALYZE 的官方示例用于说明字段含义（标注为文档引用）。可复现栈见 post/db/query-engine/reproduce/。

版本锚定：Trino 476+；Iceberg table spec V2；Iceberg REST catalog + MinIO 与 lakehouse reproduce 栈同构。

一、全景：三阶段与两类线程

flowchart TB
  subgraph CO["Coordinator JVM"]
    A["1. Analyze<br/>Analyzer"]
    P["2. Plan · Optimize<br/>Logical → Distributed"]
    F["3. Fragment · Schedule<br/>SqlQueryExecution · Stages"]
    A --> P --> F
  end
  subgraph WK["Worker JVM"]
    T["SqlTaskExecution<br/>Drivers · Operators"]
    PG["Page stream"]
    T --> PG
  end
  F -->|"CreateTask · Splits"| T

Coordinator 上的 dispatcher 线程池 接收 HTTP 查询；query 线程池 跑 SqlQueryExecution；Worker 上 TaskExecutor 跑 Driver。线程池边界可在 QueryManager（core/trino-main/src/main/java/io/trino/execution/QueryManager.java）找到。

二、SqlQueryExecution 生命周期

2.1 创建与 start()

客户端提交 SQL 后，QueryCreation 构造 SqlQueryExecution 并注册 QueryStateMachine（core/trino-main/src/main/java/io/trino/execution/SqlQueryExecution.java）。start() 路径（简化）：

1. Analyze：Analyzer 解析 AST，绑定 Catalog/Schema/Table handle（第 2 篇）。
2. Plan：LogicalPlanner 生成逻辑计划，PlanOptimizers 应用规则与 CBO（第 3–5 篇）。
3. Fragment：PlanFragmenter 在 Exchange 处切分 SubPlan（与 第 11 篇 的 Stage 边界一致）。
4. Schedule：ExecutionPolicy（默认 phased / pipelined）选择 PipelinedQueryScheduler，创建 PipelinedStageExecution。
5. Split 调度：对 leaf fragment 打开 SplitSource，SplitScheduler 批量 addSplits 到 Remote Task。

QueryStateMachine 状态从 PLANNING → STARTING → RUNNING（日志里可见 io.trino.execution.QueryStateMachine DEBUG 行，见 GitHub issue 讨论中的 production 样例格式）。

2.2 与 Spark 的对照点

Spark 在 Driver 内 SparkPlanner → SparkPlan → execute() 触发 whole-stage codegen；Trino 始终 把 distributed plan 序列化 发给 Worker，Worker 解释执行 OperatorFactory（无 JVM 字节码 codegen 主路径）。对照见 第 13 篇。

三、Plan → Fragment → Stage DAG

3.1 PlanFragmenter 切分规则

PlanFragmenter（core/trino-main/src/main/java/io/trino/sql/planner/PlanFragmenter.java）从根节点 DFS：遇到 ExchangeNode 就切新 Fragment。每个 Fragment 携带：

• Root PlanNode（本 Stage 内算子树）
• Output partitioning scheme（HASH / BROADCAST / SINGLE …）
• Symbol 布局（列名与类型）

SubPlan 形成树；scheduler 按 依赖顺序（子 fragment 先于父）或 pipelined 策略启动 Stage。

3.2 RemoteTask 创建

对每个 (PlanFragment, partitionId)，PipelinedStageExecution 调用 RemoteTaskFactory 在目标 Worker 创建 Task。Task 初始可能 无 Split（纯 exchange 消费端）；leaf Task 等待 SplitScheduler 喂 Split。

四、Iceberg：从 manifest 过滤到 Split

这是本篇与 lakehouse 第 8 篇 的 引擎侧接点：lakehouse/08 讲元数据树各层字段；这里讲 Trino Iceberg connector 谁读、谁过滤、谁产出 Split。

4.1 元数据树回顾（只列引擎需要的层）

Planning 漏斗与 lakehouse 第 18 篇 一致；本篇强调 Trino 代码路径。

4.2 Coordinator 上的 TableHandle 与约束

IcebergMetadata.getTableHandle（plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergMetadata.java）根据 session 与 snapshot 选择构造 IcebergTableHandle，包含 schema、partition spec、enforced predicate（connector 可保证的约束）。

Analyzer/Planner 把 WHERE 中能下推的谓词转为 TupleDomain<IcebergColumnHandle>，附在 table layout 上（第 8 篇 Scan 下推）。Partition 列上的谓词 在此阶段就参与 layout constraint，对应 manifest 上的 partition field 过滤（隐藏分区 transform 的折叠见 Trino Iceberg 文档 Partitioning）。

4.3 IcebergSplitSource 与 planFiles()

Leaf Stage 调度时，IcebergSplitManager 打开 IcebergSplitSource（plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergSplitSource.java）。核心逻辑：

1. 用 table.newScan() 构建 Iceberg Scan。
2. 合并 layout predicate 与 dynamic filter（DPP/runtime filter，若有）。
3. toIcebergExpression(...) 转成 Iceberg Expression。
4. 若存在谓词列，调用 scan.includeColumnStats(...) 以启用 file-level stats pruning。
5. scan.planFiles() 返回 FileScanTask 迭代器——此处完成 manifest list → manifest entry → data file 过滤（Apache Iceberg TableScan#planFiles，与 pyiceberg scan.plan_files() 同构，lakehouse/18 第八节用过纯 Python 计数）。

每个 FileScanTask 可进一步按 targetSplitSize 切成多个 Trino Split（大文件 byte range 切分）。Split 对象实现 ConnectorSplit，携带 path、start、length、partition keys、file format 等。

flowchart TD
  SQL["WHERE 谓词"] --> TD["TupleDomain · IcebergExpression"]
  TD --> SCAN["Iceberg TableScan"]
  SCAN --> ML["读 manifest list"]
  ML --> MF["过滤 manifest entries<br/>partition + column stats"]
  MF --> FS["FileScanTask 列表"]
  FS --> SP["ConnectorSplit 批次"]
  SP --> SS["SplitScheduler → Worker Task"]

4.4 与 CBO 的交叉（planFiles 可能被多次调用）

GitHub issue #11708（Trino 维护者讨论）指出：CBO 的 getTableStatistics 也可能触发 planFiles()，且无谓词时会扫描全表 metadata，规划阶段代价高。476+ 有 metadata cache（iceberg.metadata-cache.enabled）等优化。生产上若 planning 慢，应区分 「metadata 读太多」 与 「Split 太多」——前者看 coordinator CPU 与 Iceberg manifest 体积，后者看 leaf Task 数。

4.5 Worker 上：PageSource

Task 内 TableScanOperator 调用 ConnectorPageSourceProvider.createPageSource，Iceberg 侧读 Parquet footer、做 row-group 级裁剪（漏斗第三层，见 lakehouse/18），解码为 Trino Page（Column / Block，第 7 篇）。

五、Worker 上 Page 流与 Driver

5.1 SqlTaskExecution

Worker 收到 CreateTask 后，SqlTaskExecution（core/trino-main/src/main/java/io/trino/sql/planner/SqlTaskExecution.java）：

1. 实例化 OperatorFactory 链（Scan → Filter → Project → … → Output）。
2. 为每条 pipeline 创建 Driver（DriverFactory）。
3. TaskExecutor 调度 Driver：process() 循环直到无输入或无输出。

5.2 Pull 语义

Trino Operator 以 pull 为主（第 6 篇）：Sink 侧 getOutput() 触发上游 getOutput()，数据以 Page（一批行，列式 Block）流动。Hash join build 阶段是 pipeline breaker：build Driver 先吃光 build 侧输入，probe Driver 再启动。

5.3 结果回传 Coordinator

根 Stage OutputBuffer 的 partition 0 由 Coordinator 上的 client buffer 消费，HTTP 分页返回 CLI/JDBC。476+ 可选 query spooling 把大结果 spill 到外部存储（与 Worker OOM 相关的 issue 讨论见 GitHub #25999，边界问题不在此展开）。

六、EXPLAIN 与 EXPLAIN ANALYZE 怎么读

6.1 EXPLAIN（规划期）

EXPLAIN 只展示 计划，不执行 scan。关注：

Iceberg 表名在 plan 里通常显示为 iceberg:<schema>:<table>；filterPredicate 展示 已下推 到 connector 的谓词子集。

6.2 EXPLAIN ANALYZE（执行期）

Trino 476 文档 EXPLAIN ANALYZE 定义：执行语句并展示 分布式计划 + 各 operator 实际代价。

文档引用示例（Trino 476 Documentation, EXPLAIN ANALYZE — Examples，以下为官方文档摘录，非本机输出）：

Queued: 374.17us, Analysis: 190.96ms, Planning: 179.03ms, Execution: 3.06s
Fragment 2 [SOURCE]
    CPU: 210.60ms, ... Input: 1500000 rows (18.17MB); per task: avg.: 1500000.00 std.dev.: 0.00
    ScanFilterProject[table = hive:sf1:orders, filterPredicate = (...)]
           Input: 1500000 rows (18.17MB), Filtered: 45.46%, Physical Input: 4.51MB

字段读法：

Iceberg 裁剪验证思路（需在 reproduce 栈执行，本文不给 fabricated 数字）：

1. 建 分区 Iceberg 表，写入多天数据。
2. EXPLAIN ANALYZE SELECT ... WHERE <partition predicate>。
3. 对比 ScanFilterProject 的 Input rows 与表总行数；检查 plan 中 filterPredicate 是否含分区列。
4. 用 $files 系统表（Trino Iceberg 文档）交叉验证选中文件数。

6.3 VERBOSE

EXPLAIN ANALYZE VERBOSE 增加 operator 级 histogram metrics（文档 Window 示例含 p50/p99 分布），适合定位 单 operator 长尾，不要求读者通读 Trino 全部内部字段。

七、session 与并行度旋钮（与本路径相关）

八、可复现实验

post/db/query-engine/reproduce/ 提供：

docker compose -f post/db/query-engine/reproduce/docker-compose.yml up -d
bash post/db/query-engine/reproduce/run_trino_experiments.sh

本机未执行上述命令；读者输出为准。建议追加：对 iceberg.tpch.orders 按 o_orderdate 范围查询，记录 Physical Input 与 $files 行数，验证 manifest 裁剪是否生效。

九、小结

Trino 查询路径在 Coordinator 上由 SqlQueryExecution 串联 analyze → plan → fragment → schedule；Iceberg 表的 Split 来自 IcebergSplitSource 对 TableScan.planFiles() 的调用，manifest 过滤发生在 planning/schedule 交界，物理扫描发生在 Worker PageSource。读 EXPLAIN ANALYZE 时，先看 Queued/Planning 是否异常（metadata 过重），再看 Scan 的 Input/Filtered/Physical Input（下推是否生效），最后看 Exchange 的 Blocked 与 std.dev.（shuffle 与倾斜）。

下一篇对照 Spark Catalyst / AQE 与 Iceberg reader：第 13 篇。

参考资料

1. Trino 476 Documentation, Overview — Architecture；Developer guide — Internal architecture。
2. Trino 476 Documentation, EXPLAIN ANALYZE（字段定义与官方示例）。
3. Trino 476 Documentation, Iceberg connector — Metadata tables, Partitioning。
4. trinodb/trino release 476, core/trino-main/src/main/java/io/trino/execution/SqlQueryExecution.java。
5. trinodb/trino release 476, core/trino-main/src/main/java/io/trino/execution/QueryStateMachine.java。
6. trinodb/trino release 476, core/trino-main/src/main/java/io/trino/sql/planner/PlanFragmenter.java。
7. trinodb/trino release 476, plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergSplitSource.java（planFiles() 调用链）。
8. trinodb/trino release 476, plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergMetadata.java。
9. Apache Iceberg Spec, Scan planning；org.apache.iceberg.TableScan#planFiles。
10. lakehouse 第 8 篇（manifest 树）；lakehouse 第 18 篇（读湖漏斗）。

返回 系列目录 ｜ 上一篇：Shuffle 与 Exchange ｜ 下一篇：Spark SQL 与 Catalyst