【分布式 OLAP 查询引擎】Join 重排与物理算子选择

多表 join 的搜索空间随表数 指数增长。第 4 篇 的基数估计决定 哪种顺序的中间结果更小；本章在顺序确定后（或同时）选择 物理 join 算法 与 分布式数据分布：本地 HashJoin、跨节点 Partitioned shuffle，或小表 Broadcast。

本文以 Trino 476+ 文档 Join planning 为主线，用 DuckDB 1.5.4 本机 HASH_JOIN 计划 说明物理层形态；Spark 3.5+ AQE 作 B 级动态对照。不粘贴未实测的 Trino EXPLAIN ANALYZE broadcast/shuffle 耗时对比。

环境：WSL2、DuckDB 1.5.4（q_join median wall 0.0028 s，100k orders + 5k customer）；Trino/Spark 集群未在本机运行。

一、Join 顺序：left-deep 与 bushy

对 \(n\) 个关系，bushy tree 允许任意子树组合；left-deep 链只允许 右深递归（\((((R \bowtie S) \bowtie T) \bowtie U)\)）。

flowchart BT
  subgraph leftdeep ["Left-deep"]
    R1["R"] --- J1["⋈"]
    S1["S"] --- J1
    J1 --- J2["⋈"]
    T1["T"] --- J2
  end
  subgraph bushy ["Bushy"]
    R2["R"] --- JB["⋈"]
    S2["S"] --- JB
    T2["T"] --- JC["⋈"]
    U2["U"] --- JC
    JB --- JOUT["⋈"]
    JC --- JOUT
  end

Trino JoinOrderOptimizer 在 统计可用 时用 动态规划或贪心（具体启发式以 476+ 源码/文档为准）；表数超过阈值时 fallback 到固定顺序。

原则（与 PG 第 11 篇 同类）：尽早过滤、尽早缩小行数 的 join 优先（选择性高的谓词所涉及的表应早参与 filter 或早 join）。

二、物理 join 算子

OLAP 事实表 join 维表 默认 Hash Join；Merge Join 常见于 sort key 与 join key 对齐 的列存布局（ClickHouse 局部有序，columnar-engine）。

2.1 Hash Join 两阶段

flowchart LR
  BUILD["Build phase<br/>hash table on build side"]
  PROBE["Probe phase<br/>scan probe, lookup"]
  BUILD --> PROBE

Outer join 需 null 标记位 或 二次扫描 build 侧未匹配行（第 9 篇 展开）。

DuckDB 本机 q_join 计划根为 HASH_JOIN + HASH_GROUP_BY，build/probe 侧均为 SEQ_SCAN（第 2 篇 删减输出）。

三、分布式：Broadcast vs Partitioned Join

单节点 DuckDB 无 Exchange；Trino/Spark 必须决定 小表是否广播：

flowchart TB
  subgraph bcast ["Broadcast join"]
    DIM["Small dim<br/>replicate to all nodes"]
    FACT["Fact scan"]
    DIM --> H1["Local HashJoin"]
    FACT --> H1
  end
  subgraph part ["Partitioned shuffle join"]
    F2["Fact HASH(k)"]
    D2["Dim HASH(k)"]
    F2 --> EX["Exchange"]
    D2 --> EX
    EX --> H2["Co-located HashJoin"]
  end

Trino session/config 提供 broadcast join threshold（字节或 row，476+ Join planning）；低估 dim 大小 → broadcast 巨型表 → worker OOM（第 16 篇）。

本机 Trino 实验：reproduce/run_trino_experiments.sh 设计为对比 broadcast vs shuffle plan 形态；未执行时不写 wall time 数字。

四、DuckDB 实测：join 物理计划摘要

数据与 第 2、4 篇 相同。

无 broadcast 概念：多线程在同一进程内 partition morsel（第 7 篇）。对比 Trino 时，0.0028 s 仅作嵌入式 baseline，不是「Trino 应更快/更慢」的依据。

脚本另含 无 filter 的 customer-orders join 的 EXPLAIN，用于观察 纯 join 基数 ~500k（100k orders 平均一 cust 多行）时 HASH_JOIN 估算变化——读者可本地运行 python3 run_duckdb_experiments.py 查看完整树。

五、Dynamic Partition Pruning（DPP）

静态 partition pruning：plan 时从谓词 + partition spec 推出 只读哪些分区（lakehouse/18）。

Dynamic partition pruning：join 运行时，build 侧（如维表 date_dim）的实际 key 集合形成 runtime filter，probe 侧（fact）跳过无匹配分区/文件。

sequenceDiagram
  participant Build as Build scan dim
  participant RF as Runtime filter
  participant Probe as Probe scan fact
  Build->>RF: distinct join keys
  RF->>Probe: constraint on partition col
  Probe->>Probe: skip partitions

Trino Iceberg connector 与 Hive connector 在不同版本支持 dynamic filtering（476+ 文档 Dynamic filtering）；与 Bloom filter、min/max 互补。

Spark AQE（3.5+）在 shuffle 后 动态合并 partition、skew join（B 级对照，第 13 篇）：解决的是 执行期 数据分布与计划期估计不一致，与 DPP 互补。

六、Skew 与 join 计划

数据倾斜：单个 join key 对应极大 fact 行数 → 某 task 承担绝大部分 probe → straggler（第 11 篇）。

Trino 对 skew 的 内置自适应 弱于 Spark AQE；生产常 SQL 层加盐 或 bucket 表（第 17 篇）。

七、与 stream-processing 的对比

stream-processing 的 流式 join 维护 状态 buffer（有界 TTL），不是 batch hash table 一次性 build。交互式 OLAP join 假设 输入有界、一次性 spill 可接受；延迟目标 秒级 而非 毫秒级 event-time 窗口（stream-processing 第 1 篇）。

八、计划判读速查

Trino 字段命名以 476+ 为准；DuckDB 看 HASH_JOIN 与 Cardinality。

九、小结

Join 规划 = 顺序（CBO） + 算法（Hash/Merge/NL） + 分布（broadcast/shuffle/colocated） + 运行期裁剪（DPP）。统计 第 4 篇 错一步，broadcast 与 order 会连锁失败。DuckDB 本机 HASH_JOIN + 0.0028s 展示 嵌入式物理层；分布式代价在 Exchange 网络（第 10–12 篇）。

下一篇进入 Volcano 迭代器模型（执行语义，与优化器解耦）。

参考资料

1. Trino Documentation, Join planning / Dynamic filtering（476+）。
2. Apache Spark Documentation, Adaptive Query Execution（3.5+，B 级对照）。
3. DuckDB 1.5.4，reproduce/run_duckdb_experiments.py。
4. postgresql-kernel 第 11 篇（join paths）。
5. 第 4 篇（基数估计）。
6. lakehouse 第 18 章（partition pruning）。

返回 系列目录 ｜ 上一篇：统计与代价 ｜ 下一篇：Volcano 模型