【分布式 OLAP 查询引擎】Hash Join 与 Hash Aggregation：Build/Probe 与 Spill

多表 join 与 GROUP BY 是 OLAP 查询里最常触发 内存峰值 的两类算子。第 5 篇 在 规划层 选 hash vs broadcast、build/probe 侧；第 6 篇 把它们标为 pipeline breaker。本篇落在 执行层：hash 表长什么样、MPP 下 partial / final agg 如何合并、内存触顶时 spill 如何保住查询完成。

一、Hash Join：Build 与 Probe

Trino 采用 in-memory hash join（Trino Documentation Cost-based optimizations — Join distribution selection）。对流式一侧 probe，另一侧 build 进 hash 表。

1.1 阶段划分

flowchart TB
  subgraph build ["Build phase (breaker)"]
    BI["Build input<br/>consume all pages"]
    HT["Hash table"]
    BI --> HT
  end
  subgraph probe ["Probe phase (streaming)"]
    PO["Probe input"]
    OUT["Join output"]
    PO --> OUT
    HT --> OUT
  end

Trino 实现上常拆为 HashBuilderOperator + LookupJoinOperator（或 probe 侧 HashProbeOperator），由 Driver 特殊调度（Driver.java 注释：HashBuild 打断自然 pipeline）。

1.2 Build 侧选择与多 join

CBO 选择 较小 输入作 build（在 broadcast / partitioned 决策之后）。Trino 文档描述 多 join 链：默认将 最右表 作第一个 build，结果驻留内存再与下一表 probe——因此 SQL 手写 join 顺序在 极端内存 场景仍可能影响峰值（文档建议大表尽量靠左，属工程启发而非硬规则）。

Outer join 标记位 使 hash 表条目大于 inner join，内存估算需在 CBO 中体现（第 4 篇）。

1.3 分布式：Broadcast vs Partitioned

Partitioned join 的 build 表 分片 在各节点，单节点峰值约为 总 build 大小 / parallelism；仍可能 spill（第 11 篇）。

二、Hash 表布局（概念）

引擎实现各异，逻辑结构一致：

Probe 对 batch 内每行：h = hash(key) → 查 chain → 比较相等。向量化实现常 先算 hash 数组，再 gather 探测（第 7 篇）。Skew 导致单 bucket 过长时，即使总内存够也会极慢（第 17 篇）。

三、Hash Aggregation

GROUP BY + 聚合函数在单节点上构建 grouping hash 表：key 为分组列，value 为 partial accumulator（COUNT 计数、SUM 和、AVG 的 sum+count 等）。

3.1 Partial 与 Final 两阶段

MPP 查询常在每个 Worker 做 Partial，再 Exchange 按 group key 重分区，最后 Final 合并：

flowchart LR
  S1["Worker A<br/>Partial Agg"]
  S2["Worker B<br/>Partial Agg"]
  EX["Exchange<br/>hash(group key)"]
  FA["Final Agg"]
  S1 --> EX
  S2 --> EX
  EX --> FA

COUNT(DISTINCT x) 一般 不能 简单 partial-final（需 distinct 状态或两阶段 distinct agg），内存与 shuffle 代价显著高于普通 COUNT(*)——规划器可能插入 MarkDistinct 或 expanded plan（Trino 实现细节见源码 AggregationNode step 枚举）。

3.2 DISTINCT 与 GROUP BY 内存

统计信息 NDV 估错会导致 under-estimate → OOM 或 spill（第 4 篇）。

四、Spill：Revocable Memory 与磁盘

当 query.max-memory / query.max-memory-per-node 触顶，Trino 可 revoke 算子持有的 revocable memory，触发 spill to disk（Trino Documentation Spill to disk，476+）。

4.1 机制概要

配置入口：spill-enabled、spiller-spill-path 等（Spilling properties）。

4.2 Join Spill

文档描述：build 表分区；内存不足时 部分 partition 与对应 probe 行落盘；逐 partition 完成 join，降低峰值内存至约 最大 partition 规模。代价是 磁盘 IO 与更长 wall time——属正确性优先，非加速手段。

4.3 Aggregation Spill

中间 partial 状态 写盘 → 后续 load + merge，降低常驻 hash 表 footprint。属性 aggregation-operator-unspill-memory-limit 限制单次 unspill 内存（默认 4MB，版本以文档为准）。

4.4 支持范围与边界

Trino 文档列明 spill 支持：aggregation、inner/outer join、sort、window；并注明 不减少其他 join 类型 的内存需求。Sort spill 与 join spill 的调优见 第 16 篇。

本机说明：未在本环境构造超 query.max-memory 的 Trino 集群实验；不粘贴未观测的 spill 日志。验证步骤：Docker Trino 缩小内存上限 + 大 join，EXPLAIN ANALYZE 查看 spilledDataSize 等算子统计（476+ 字段以文档为准）。

五、DuckDB 对照：本机 HASH_JOIN 实验

脚本 reproduce/run_duckdb_experiments.py；DuckDB 1.5.4，WSL2，orders 100k × customer 5k join + agg：

DuckDB 同样在 build 完成后 probe，并行由 morsel-driven pipeline 负责（第 7 篇）。嵌入式场景无 partial/final exchange，单进程内完成 agg；联邦或 UNION ALL 分源时才出现类似两阶段合并。

六、与 Scan 的衔接

Hash join probe 侧理想 直连 Filter 后的 Scan Page 流（第 8 篇）。Dynamic filter 缩小 probe scan，降低 probe 行数 × build 不变 场景下的 CPU；build 侧仍取决于维表大小与 broadcast 决策。

七、算子对照表

八、边界

• Join 规划（broadcast 阈值、DPP）→ 第 5 篇。
• Resource group 与集群级内存 → 第 16 篇。
• Merge join / nested loop → 第 5 篇；本篇不展开。

参考资料

系列导航：系列索引 · 上一篇：Scan、Filter 与 Project · 下一篇：Coordinator 与 Worker 架构