【分布式 OLAP 查询引擎】Shuffle 与 Exchange：分区、广播与倾斜

第 10 篇 把 Trino 的 Stage 边界钉在 exchange 上：Stage 之间不共享内存，中间结果必须 重分区（shuffle） 或 广播（broadcast） 才能继续算 join、aggregation final 阶段。运维看到查询慢、网络打满、个别 Task 的 Input rows 远高于均值——这些现象大多出在 Exchange 层，而不是单个 HashProbe 算子本身。

本文回答四个生产里高频的问题：

• Trino 有哪些 partitioning scheme，各自在什么物理计划节点出现？
• LocalExchange 与 Remote Exchange（RemoteSource / PartitionedOutput）差在哪？
• Broadcast join 与 partitioned shuffle join 在 plan 上长什么样？
• 数据倾斜 如何在 EXPLAIN ANALYZE 的 Input avg. / Input std.dev. 上暴露？Trino 与 Spark AQE 的缓解边界各在哪？

后文默认读者已读过 第 5 篇 的 join 算法选择与 broadcast 阈值，以及 第 10 篇 的 Stage/Task 模型。Trino 全路径时序见 第 12 篇；Spark Catalyst 与 AQE 细节见 第 13 篇。

环境说明：本机 WSL2、无 JVM、Docker Compose 不可用。机制描述来自 Trino 476 文档（Develop — Exchange、SQL — EXPLAIN ANALYZE）与 trinodb/trino 源码；不粘贴本机未执行的 EXPLAIN 输出。倾斜判读方法引用 Trino 官方文档 EXPLAIN ANALYZE 示例中的 Input std.dev. 字段说明。实验步骤见 reproduce/run_trino_experiments.sh。

版本锚定：Trino 476+；Spark 3.5+ 仅作 shuffle/AQE 对照引用（Apache Spark SQL Guide）。

一、为什么需要 Exchange

分布式 OLAP 里，leaf Stage 的 scan Task 各读各的 Split；一旦遇到 需要相同 key 落在同一并行实例 的算子（Hash Join build/probe 对齐、Hash Aggregation partial→final、Order By merge），就必须 按 key 重分布 行。

flowchart LR
  S1["Stage 1<br/>partial agg / scan"]
  EX["Exchange<br/>HASH / BROADCAST"]
  S2["Stage 2<br/>final agg / join probe"]
  S1 --> EX --> S2

Exchange 在 Trino 物理计划里是显式节点（LocalExchange、RemoteSource，以及输出侧的 PartitionedOutput）。这与 流式处理 Flink 运行时 里 JobVertex 之间的 IntermediateDataSet shuffle 同构：都是 pipeline 断点 + 网络（或内存）重分区。

二、Partitioning scheme 一览

Trino 在 PlanFragment 上标注 output partitioning（源码：io.trino.sql.planner.plan.ExchangeNode、PartitioningScheme）。常见 scheme：

EXPLAIN 输出里常见行：

Output partitioning: HASH [clerk][$hashvalue]

表示该 Fragment 输出按 clerk 列（及 hash 列）做 hash 分区；下游 Stage 的 RemoteSource 会从对应数量的 output buffer 拉取。

2.1 HASH 分区函数

Hash 分区通常使用 combine_hash 与 $operator$hash_code 组合（Trino 476 文档 EXPLAIN ANALYZE 示例）。对 join key \((k_1,\ldots,k_n)\)，同一组 key 值必落同一 partition index \(p \in [0, N)\)，其中 \(N\) 为下游 Task 数（parallelism）。

Partition 数由谁定？ 对 shuffle join，probe 端与 build 端 final Stage 的 partition 数一致，由 planner 在 determinePartitionCount 阶段结合 统计信息与 session（join_distribution_type、query.max-hash-partition-count 等）决定。统计偏差会导致 partition 数过小 → 单 Task 内存膨胀（第 5 篇）。

2.2 BROADCAST 与 REPLICATE

Broadcast 把 build 端小表 完整复制 到 probe 端每个 Task 的内存里。Plan 上表现为 build Stage 的 Output partitioning: BROADCAST，probe Stage 含 RemoteSource 且 join 类型为 InnerJoin with DistributionType::REPLICATED（逻辑名随版本略有差异，以 EXPLAIN 为准）。

Trino 476 文档 Cost-based optimizations 描述：当 build 表 estimated size 低于 threshold（默认约 100 MB 量级，受 join-max-broadcast-table-size 等影响）时 CBO 倾向 broadcast，避免 shuffle 网络代价。

REPLICATE 与 BROADCAST 在实现上都走「多副本」路径；差异多在 planner 对 语义合法性（outer join 哪端可广播）与 memory accounting 的标注。

三、LocalExchange 与 Remote Exchange

3.1 同一 Task 内：LocalExchange

LocalExchange 在 同一 Task 或同一 Worker 的多个 Driver 之间重分区，不走网络。典型场景：

• Partial aggregation 后，同一 Task 内多 Driver 的结果合并到 final agg 输入 layout；
• 调整 column layout 以匹配下游 Operator。

EXPLAIN ANALYZE 里 LocalExchange[partitioning = HASH, ...] 的 Blocked 时间往往反映 Driver 间 backpressure，而非跨机网络。

3.2 跨 Stage：PartitionedOutput + RemoteSource

跨 Worker 的 shuffle 由两段配合完成（源码：io.trino.operator.ExchangeOperator、PartitionedOutputOperator）：

flowchart TB
  subgraph W1["Worker A — upstream Task"]
    PO["PartitionedOutputOperator<br/>serialize · compress · partition"]
    BUF["OutputBuffer<br/>per partition queue"]
    PO --> BUF
  end
  subgraph W2["Worker B — downstream Task"]
    EC["ExchangeClient"]
    EX["ExchangeOperator<br/>RemoteSource 消费端"]
    EC --> EX
  end
  BUF -->|"HTTP / exchange manager"| EC

1. 上游 PartitionedOutputOperator 把 Page 按 hash 写入 OutputBuffer 的多个 partition queue。
2. 下游 ExchangeOperator（计划里显示为 RemoteSource[sourceFragmentIds = [k]]）通过 ExchangeClient 拉取对应 partition 的数据。
3. 数据路径可经 exchange-manager（本地文件、S3 等） spill 大 shuffle，边界见 Trino 476 文档 Exchange manager。

Trino 476 文档 EXPLAIN ANALYZE 官方示例中，Fragment 1 含 RemoteSource[sourceFragmentIds = [2]]，且 Blocked 时间占比较高——这通常表示下游在等上游数据或 network buffer 背压，而非 CPU 算子慢。

3.3 压缩与 serde

Shuffle Page 序列化经 PagesSerde；exchange.compression-codec 控制 LZ4/ZSTD 等。压缩降低网络带宽，增加 CPU；交互式集群常在 万兆网络 + 中等压缩 之间折中（具体默认值以 release 文档为准）。

四、Broadcast join 与 Shuffle join 的 plan 对照

第 5 篇 讲过 dynamic partition pruning（DPP） 与 runtime filter：probe 端 scan 可能带 dynamicFilter 约束，这与 Exchange 正交——DPP 减少 scan 量，Exchange 决定 join 数据如何对齐。

五、数据倾斜：成因与 plan 上怎么看

5.1 倾斜从哪来

• 业务 key 分布不均：少数 user_id / merchant_id 占绝大多数行。
• Null key 或默认键：大量行落入同一 hash bucket。
• Partition 数过少：统计低估 intermediate rows，planner 选了过小的 hash partition count。
• Broadcast 误判（另一类问题）：本不该 broadcast 的表被广播，表现为 内存 而非 hash skew。

5.2 EXPLAIN ANALYZE 判读（引用官方字段，非本机输出）

Trino 476 文档 EXPLAIN ANALYZE 说明：

For each plan node you can see … Input avg. … Input std.dev. … Such statistics are useful when one wants to detect data anomalies for a query (e.g.: skewness).

读法：

操作步骤（需在 Trino 环境执行）：

1. 对 suspected SQL 跑 EXPLAIN ANALYZE（必要时 VERBOSE）。
2. 定位 HashProbe / Aggregate FINAL / RemoteSource 节点。
3. 比较同 Stage 各 Task 的 Input rows（Web UI Task 视图更直观）。
4. 若 leaf scan 已均匀但 join 后 skew，问题在 join key；若 leaf 即 skew，可能是 Split 与文件大小 不均（Iceberg 小文件问题，见 lakehouse 第 17 篇）。

本文 不构造虚构的倾斜 benchmark 数字；PLAN.md 实验台账要求 skew 表 join 时记录各 task raw input rows，读者可在 reproduce/ 栈上自建 skew 键表验证。

5.3 Trino 与 Spark 缓解边界对照

Trino 生产上倾斜缓解仍偏 SQL 改写（salting）、join key 预处理、提高 hash partition 数、避免 broadcast 误判；Spark AQE 细节见 第 13 篇。

六、MergeExchange 与 ORDER BY

全局 ORDER BY 或 merge join 需要 有序 输入时，plan 可能出现 MergeExchange：各 partition 内有序，跨 partition multi-way merge。这是比 hash shuffle 更贵的 pipeline breaker（第 6 篇），大结果集排序应尽量避免无 LIMIT 的全局 sort。

七、与 lakehouse 读路径的交叉

Shuffle 解决 算子间 数据对齐；scan 量 由 Iceberg manifest 过滤决定（lakehouse 第 18 篇）。若 partition pruning 失效导致 leaf Stage 输入行数极大，后续 hash shuffle 的网络与 Blocked 时间必然上升——排查时应 先看 ScanFilterProject 的 Input / Filtered 比例，再查 Exchange（第 12 篇）。

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

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

# 观察 join 计划中的 Output partitioning 与 RemoteSource
# 建议在 reproduce 栈上追加 skew 键表：某 custkey 占 90% 行，对比 EXPLAIN ANALYZE 的 Input std.dev.
bash post/db/query-engine/reproduce/run_trino_experiments.sh

run_trino_experiments.sh 含 EXPLAIN join 与 filter 样例；skew 实验需读者自行 INSERT 倾斜数据。输出字段解释以 Trino 476 文档 EXPLAIN ANALYZE 为准。

九、小结

Exchange 是 MPP OLAP 的 网络与内存枢纽：HASH shuffle 对齐 join/agg key，BROADCAST 复制小表，LocalExchange 与 RemoteSource 区分进程内与跨 Worker 路径。倾斜诊断优先看 EXPLAIN ANALYZE 的 Input avg. / std.dev. 与 Web UI 各 Task 行数分布；Trino 与 Spark 在 运行时 skew 缓解 上能力不对称，选型与改写策略需分开评估。

下一篇进入本系列 Trino 主线：第 12 篇 从 SqlQueryExecution 到 Worker Page 流，并接上 Iceberg manifest → Split 的来源链。

参考资料

1. Trino 476 Documentation, EXPLAIN ANALYZE（Input avg. / Input std.dev. 与 skew 说明）。
2. Trino 476 Documentation, Cost-based optimizations — Join distribution（broadcast vs partitioned）。
3. Trino 476 Documentation, Exchange manager（shuffle 存储后端）。
4. trinodb/trino release 476, core/trino-main/src/main/java/io/trino/operator/ExchangeOperator.java。
5. trinodb/trino release 476, core/trino-main/src/main/java/io/trino/operator/PartitionedOutputOperator.java。
6. Apache Spark 3.5 Documentation, Adaptive Query Execution（skew join、coalesce 对照）。
7. 本系列 第 10 篇（Stage/Task 模型）。
8. 本系列 第 5 篇（join 物理选择与 DPP）。

返回 系列目录 ｜ 上一篇：Coordinator 与 Worker ｜ 下一篇：Trino 查询路径全拆解