【分布式 OLAP 查询引擎】Coordinator 与 Worker：Query / Stage / Task / Driver

第 1 篇 把 OLAP 引擎分成「单进程嵌入式」与「需要 shuffle 的分布式 MPP」两类：数据量超过单机内存或需要跨节点并行 scan 时，必须引入 Coordinator（协调者） 与 Worker（执行节点） 的分工。PostgreSQL 与 DuckDB 的查询全在进程内完成；Trino 则把 计划编译、Stage 调度、Split 分配 放在 Coordinator，把 Task 执行、Page 流 shuffle 放在 Worker。

运维第一次打开 Trino Web UI，通常会看到 Query ID、Stage、Task、Split 计数；读 EXPLAIN 输出又会遇到 Fragment。这些名字各自对应什么对象？一条 SELECT 从客户端提交到 Worker 线程里跑 Operator，中间经过几层调度单元？

本文不教 Trino 安装参数大全，而是把 Trino 476+ 主线的 MPP 骨架讲清楚，回答四个会直接决定资源规划与故障排查的问题：

• Coordinator 与 Worker 各自管什么，和 Spark Driver/Executor 差在哪？
• Query → Stage → Task → Driver → Operator 五层各自输入输出是什么？
• Split 由谁产生、由谁调度、与 data locality 怎么配合？
• 读 EXPLAIN 里的 Fragment 时，应如何映射到 Stage？

后文默认读者已读过本系列 第 1 篇 的全景，以及 第 7 篇 的 Page 与向量化 batch 直觉。Shuffle 与 Exchange 的细节留到 第 11 篇；从 SqlQueryExecution 到 Iceberg Split 的完整时序见 第 12 篇。

环境说明：本机为 WSL2（Linux 6.6.87.2）、i9-12900K / 32 GiB，未安装 JVM，当前环境 Docker Compose 不可用。本文架构结论来自 Trino 476 官方文档（Overview、Administration）与 trinodb/trino 源码；不粘贴未在本机执行的 EXPLAIN ANALYZE 输出或 Web UI 截图。文末给出 reproduce/ 栈的可复现步骤，读者可在具备 Docker 的环境自行验证。

版本锚定：Trino 476+（trinodb/trino:476 镜像与文档主线一致）。Fault-tolerant execution（retry-policy=TASK）与 pipelined execution 在 Stage 调度实现上分支不同，本篇以默认 pipelined 模式为主，fault-tolerant 边界在 第 16 篇 标注。

一、两层进程模型：Client、Coordinator、Worker

Trino 集群在逻辑上分成三类角色（来源：Trino Documentation，Overview — Architecture）：

flowchart TB
  CL["Client<br/>SQL / result pages"]
  CO["Coordinator<br/>plan · schedule · split assign"]
  W1["Worker 1<br/>Task · Driver · Operator"]
  W2["Worker 2<br/>Task · Driver · Operator"]
  CL -->|"HTTP / JDBC"| CO
  CO -->|"create Task · assign Split"| W1
  CO -->|"create Task · assign Split"| W2
  W1 <-->|"exchange pages"| W2

与 流式处理 Flink 运行时 对照：Flink 的 JobManager 近似 Trino Coordinator 的调度面；Flink TaskManager 近似 Trino Worker。差异在于 Trino 把 Split（connector 提供的数据切分单元）作为 scan 并行度的第一公民，且 查询级资源组与内存账户 由 Coordinator 集中裁决（第 16 篇）。

Spark 对照（点到为止，Catalyst 全面对照见 第 13 篇）

二、五层执行模型：Query → Stage → Task → Driver → Operator

Trino 把一次 SQL 执行切成五层对象（来源：trinodb/trino 源码包 io.trino.execution、io.trino.operator；Trino Documentation Developer guide — Internal architecture）：

flowchart TB
  Q["Query<br/>一次 SQL 提交"]
  S["Stage<br/>shuffle 边界切分"]
  T["Task<br/>PlanFragment 的一个并行实例"]
  D["Driver<br/>Task 内执行线程"]
  O["Operator<br/>物理算子 pipeline"]
  Q --> S --> T --> D --> O

2.1 Query：生命周期状态机

客户端提交 SQL 后，Coordinator 创建 Query，由 QueryStateMachine 跟踪状态（源码：core/trino-main/src/main/java/io/trino/execution/QueryStateMachine.java）。典型状态迁移包括：

编排入口是 SqlQueryExecution（core/trino-main/src/main/java/io/trino/execution/SqlQueryExecution.java）：它持有 QueryStateMachine，在 start() 里完成 analyze → plan → distribute → schedule。这与 Spark Driver 里 SparkSession.sql(...).collect() 触发 QueryExecution 的路径类似，但 Trino 把 调度与 exchange 固化在 coordinator 侧服务里，不依赖外部 YARN/K8s 时再嵌一层「application master」语义。

2.2 Stage：shuffle 边界上的调度单元

Stage 是 Query 内按 exchange（数据重分布） 切出来的子计划。每个 Stage 对应一个 PlanFragment（分布式物理计划片段）。Stage 之间构成 DAG：下游 Stage 通过 RemoteSource 消费上游 Stage 的输出 buffer。

Pipelined 模式下，PipelinedStageExecution（core/trino-main/src/main/java/io/trino/execution/scheduler/PipelinedStageExecution.java）管理 Stage 内 Task 的创建、Split 分配与完成检测。类注释写得很直白：pipeline 模式里 Task all-or-nothing 地推进，中间数据以 streaming 方式在 Stage 间管道传输。

读 EXPLAIN 时看到的 Fragment N 就是 PlanFragment 的编号；Fragment 之间的箭头 即 Stage 依赖。Fragment 0 通常是输出 Stage（向客户端吐结果），最大编号往往是 leaf scan Stage。

2.3 Task：PlanFragment 的一个并行实例

Task 是某个 PlanFragment 在 一个 Worker 上的一次部署。并行度为 \(p\) 时，该 Fragment 对应 \(p\) 个 Task，每个 Task 处理不同的 partition 或不同的 Split 子集。

Coordinator 通过 RemoteTaskFactory 向 Worker 发送 CreateTaskRequest（REST：/v1/task/{taskId}）。Worker 上的 SqlTaskExecution（core/trino-main/src/main/java/io/trino/sql/planner/SqlTaskExecution.java）在收到 Task 后构建本地 Operator pipeline，并向 Coordinator 周期性汇报 TaskStatus（CPU、内存、completed splits、buffer 状态）。

2.4 Driver：Task 内的执行线程

一个 Task 可包含多个 Driver——每个 Driver 是一条 无 pipeline breaker 的 Operator 链上的执行线程。Driver 数受 task.concurrency（session 级，默认与 Worker 核数相关）约束。

Driver 的工作循环（简化）：从 Source Operator pull 一个 Page → 依次传给下游 Operator → 直到 Sink。多个 Driver 并行处理同一 Task 内不同的 Split 或 pipeline 分区，类似 第 7 篇 里 morsel-driven 的「多线程各拉一批」。

2.5 Operator：物理算子

Operator 是 Volcano 模型在 Trino 里的实现单元（接口 io.trino.operator.Operator）：getOutput() 拉取、needsInput() / addInput() 推送（部分算子）。Scan、Filter、Project、HashAggregation、HashProbe、Exchange 各自是一个 Operator 或 OperatorFactory。

Pipeline breaker（Sort、Hash Aggregation build 阶段等）会截断 Operator chain，因此一个 Task 里常有 多条 Driver，每条 Driver 覆盖 breaker 之间的一段 pipeline。

三、Split：connector 切分与 Coordinator 调度

3.1 Split 是什么

Split 是 connector 暴露给引擎的 只读数据切分单元，实现 ConnectorSplit SPI（io.trino.spi.connector.ConnectorSplit）。对 Iceberg 表，一个 Split 通常对应 一个 data file（或大文件的一个 byte range）；对 JDBC，可能是主键范围；对 Kafka，可能是一个 topic partition 的一段 offset 范围。

Split 由 SplitSource 异步产生；Coordinator 侧的 SplitScheduler（core/trino-main/src/main/java/io/trino/execution/SplitScheduler.java）批量拉取 Split，交给 NodeScheduler 分配到 Worker Task。Leaf Stage（含 TableScan）的并行度上界，往往由 可用 Split 数 与 集群 Worker 数 共同决定。

3.2 Locality 与 host affinity

Connector 可在 Split 上标注 preferred hosts（例如 HDFS block location、对象存储同 AZ hint）。NodeScheduler 在分配时优先把 Split 调度到亲和节点，减少跨机读放大。Iceberg on S3/MinIO 场景 locality 弱于 HDFS，更多依赖 并发 Split 数 而非磁盘亲和；manifest 过滤后 Split 变少会直接降低 scan 并行度——这是 第 12 篇 与 lakehouse 第 8 篇 的衔接点。

3.3 与 Spark Input Split 的对照

Spark 的 RDD/DataFrame partition 由 TextInputFormat.getSplits() 或数据源 V2 的 planInputPartitions() 决定；Trino 的 Split 在 planning 完成后、Task 启动前 由 Coordinator 统一调度。两者共同点是：scan 并行度由「切分单元数」封顶；差异是 Trino 把 Split 调度集中放在 Coordinator，Worker 只执行已分配的 Task。

四、一次查询的时序骨架

下面时序省略 planning 细节（见 第 12 篇），只保留 MPP 调度主路径：

sequenceDiagram
  participant C as Client
  participant CO as Coordinator
  participant SS as SplitScheduler
  participant W as Worker
  C->>CO: submit SQL
  CO->>CO: analyze · optimize · fragment
  CO->>CO: create Stages (DAG)
  loop leaf Stage
    CO->>SS: open SplitSource
    SS->>CO: batches of Splits
    CO->>W: createTask + addSplits
  end
  W->>W: Driver runs Operators
  W->>CO: output buffers / status
  CO->>C: result pages

关键观察：

1. Planning 在 Coordinator 完成，Worker 收到的是已编译的 PlanFragment + Split 列表。
2. 非 leaf Stage 的 Task 可能 不绑定 Split（输入来自上游 exchange buffer）。
3. Stage 按依赖顺序启动：pipelined 模式下，下游 Stage 可在上游开始输出后逐步消费（与 Flink pipelined region 类似，见 Flink 运行时）。

五、从 Cluster 视角读配置

与运行时直接相关的配置项（config.properties / session）：

reproduce/trino/config.properties 与 node.properties 已按单 coordinator + 单 worker 开发栈预置；生产上 Coordinator 不应 与 heavy Worker 混部在同一 JVM 堆上过小，否则 planning 与 exchange 协调会与 Task 抢内存（第 16 篇）。

六、常见问题与排查入口

七、可复现实验（需 Docker 环境）

本机未跑通以下步骤；脚本与 compose 文件位于 post/db/query-engine/reproduce/。

# 启动 Trino 476 + MinIO + Iceberg REST（需 Docker Compose）
docker compose -f post/db/query-engine/reproduce/docker-compose.yml up -d

# 健康检查
curl -s http://localhost:8080/v1/info | head

# 查看 Query / Stage / Task（CLI 内）
# EXPLAIN SELECT count(*) FROM iceberg.tpch.orders;
# 对照 Web UI：Query Details → Stage → Tasks
bash post/db/query-engine/reproduce/run_trino_experiments.sh

run_trino_experiments.sh 会建 iceberg.tpch.orders 并执行 EXPLAIN / EXPLAIN ANALYZE。输出以读者环境为准；字段含义见 第 12 篇 与 Trino 476 文档 EXPLAIN ANALYZE。

八、小结

Trino MPP 把一次 SQL 切成 Query → Stage → Task → Driver → Operator 五层：Coordinator 负责 plan 与 schedule，Worker 负责 Task 内 Page 流执行；Split 决定 leaf scan 并行度，Exchange 决定 Stage 边界（第 11 篇）。理解 Fragment 与 Stage 的对应关系，是读 EXPLAIN 与 Web UI 的前提。

下一篇进入 Shuffle 与 Exchange：hash / broadcast / replicate 分区函数，LOCAL 与 REMOTE exchange，以及倾斜在 EXPLAIN ANALYZE 上的判读方法。

参考资料

1. Trino 476 Documentation, Overview — Architecture（Coordinator / Worker 角色）。
2. Trino 476 Documentation, Developer guide — Internal architecture（Query / Stage / Task 术语）。
3. trinodb/trino release 476, core/trino-main/src/main/java/io/trino/execution/SqlQueryExecution.java（查询编排入口）。
4. trinodb/trino release 476, core/trino-main/src/main/java/io/trino/execution/QueryStateMachine.java（Query 状态机）。
5. trinodb/trino release 476, core/trino-main/src/main/java/io/trino/execution/scheduler/PipelinedStageExecution.java（pipelined Stage 调度）。
6. trinodb/trino release 476, core/trino-main/src/main/java/io/trino/execution/SplitScheduler.java（Split 批次调度）。
7. Trino SPI, io.trino.spi.connector.ConnectorSplit（Split 抽象）。
8. 本系列 第 12 篇（Coordinator 全路径时序）。

返回 系列目录 ｜ 上一篇：Hash Join 与 Hash Aggregation ｜ 下一篇：Shuffle 与 Exchange