本系列第 1/3 篇 · Join 执行 · 下一篇 · 相关：DOP 三次决策

先定义 Hash Join 需要保存的状态

引言：Hash Join 不只是一次哈希查找

Hash Join 用 Build 侧索引减少等值匹配的查找工作，但其资源边界由键表示、重复行、并发与落盘协议共同决定。StarRocks、Doris、ClickHouse 和 DuckDB 展示了不同的状态组织方式。

教科书对 Hash Join 的描述非常简洁：选择较小的一侧构建哈希表（Hash Table），再用另一侧逐行查表。在哈希函数近似均匀、Build 侧能够放入内存时，Build 与 Probe 主路径接近 O(N + M)；若需输出 Z 条结果，还要计入 O(Z)，多对多重复键可能使输出远大于输入。但真正的工程实现远比这个模型复杂：

• Join Key 是单个整数、定长组合键，还是需要序列化的变长键？
• 一个 Key 对应一行还是多行，重复链如何组织？
• Inner、Outer、Semi、Anti、Mark Join 如何共享同一条 Probe 主路径？
• 多线程如何并行 Build，又如何避免让 Probe 为并行付出额外代价？
• 哈希表占满内存后，哪些状态可以撤销，哪些状态不能直接落盘？
• Build 与 Probe 如何使用完全一致的分区规则，保证匹配行最终相遇？
• 单个热点 Key 大到任何分区都装不下时，递归分区是否仍然有效？

这些问题决定了 Hash Join 究竟是一个只在小表上很快的算法，还是一个能支撑复杂分析负载、在内存压力下仍然可靠的执行系统。

本文基于以下本地源码版本展开分析：

本文关注执行引擎的共性与差异。Spark 的 JoinSelection、HashedRelation、AQE 与「Hash Join 默认不可 Spill」边界见文末 附录：Spark Hash Join 特有路径。

比较方法

本文以 SQL 语义、状态所有权和 Spill 恢复为共同坐标。先建立键、Payload、重复链与匹配标记的内存模型，再核对各引擎何时撤销状态、如何分区和恢复。源码展示的是固定快照中的机制，性能结论仍需对应负载的测量。

从 SQL 到内存对象

Join 输出列与谓词决定 Build 必须保留哪些状态，因而相同连接键也可能得到不同内存布局。考虑以下查询：

1SELECT o.order_id, c.region
2FROM orders o
3JOIN customers c
4  ON o.customer_id = c.customer_id;

若优化器选择 customers 为 Build 侧，逻辑过程如下：

 1Build side: customers
 2  ├─ 计算 customer_id 的 Hash
 3  ├─ 保存 Join Key 与输出所需 Payload
 4  └─ 建立 Hash Bucket -> Build Row 的索引
 5
 6Probe side: orders
 7  ├─ 批量计算 customer_id 的 Hash
 8  ├─ 定位 Bucket
 9  ├─ 校验完整 Key，排除 Hash 冲突
10  └─ Gather Build Payload，拼接输出

内存不能只按原始数据量估算

Build 侧哈希表的近似内存可以写成：

1M_build ≈ M_key
2        + M_payload
3        + M_bucket/control
4        + M_duplicate_link
5        + M_null_and_match_flags
6        + M_runtime_filter
7        + M_allocator_fragmentation

因此，“Build 表原始数据只有 2 GiB”并不意味着 2 GiB 内存足够。低装载因子的桶数组、对齐、字符串 Arena、重复链和输出列物化都可能把实际占用放大。

另一个常被忽略的量是输出规模：

1JoinOutput = Σ count_probe(k) × count_build(k)

Hash Join 只能让查找接近常数时间，无法消除多对多 Join 的组合爆炸。一个高频 Key 同时出现在两侧时，即使哈希表能够放入内存，输出也可能成为真正瓶颈。

Build 与 Probe 之间是一份状态契约

Build 阶段至少要向 Probe 阶段交付：

• Join Key 的编码方式；
• 哈希函数与桶布局；
• Build 行或 Payload 的定位方式；
• NULL 是否参与匹配；
• 重复 Key 的遍历方式；
• Outer Join 所需的命中标记；
• 分区与 Spill 的元信息。

这份状态一旦建立，Probe 就不能随意更换 Hash 或 Key 编码。External Hash Join 只是把这份契约从“一张全局哈希表”扩展到“按分区反复建立的局部哈希表”。

内存快路径：键、语义与并行 Build/Probe

键编码决定哈希表形态

数据库很少把所有 Join Key 都转换成通用对象。通用表示便于编程，却会引入虚函数、分支、间接访问、重复 Hash 和额外内存。

更常见的选择是：

StarRocks：Key Constructor 与 Hash Method 解耦

StarRocks 在 join_hash_table.cpp 中把两个问题分开处理：

1. JoinKeyConstructorType 决定如何从列中构造 Key，例如 ONE_KEY、SERIALIZED_FIXED_SIZE 和通用序列化；
2. JoinHashMapMethodType 决定如何由 Key 定位行，例如 DIRECT_MAPPING、RANGE_DIRECT_MAPPING、DENSE_RANGE_DIRECT_MAPPING、LINEAR_CHAINED 和 BUCKET_CHAINED。

这意味着“Key 是什么”和“如何索引 Key”可以独立优化。对值域足够小的整数，直接映射可能比通用 Hash 更合适；对唯一 Build Key，*_SET 变体可以省掉重复链；无法命中特化条件时，再退回通用 Bucket Chaining。

ClickHouse：从数值键到加密哈希键

ClickHouse 的 chooseMethod() 会根据实际列类型选择 Map：

• 单个数值键使用 key8、key16、key32、key64 等专用类型；
• 多个定长键总宽度不超过 16 或 32 字节时，打包为 keys128 或 keys256；
• 单个 String/FixedString 使用专用路径；
• 其他情况使用无歧义序列化值的宽 Hash 表示。

对应的 Value 也并非一种：HashJoin.h 区分 RowRef、RowRefList 和 AsofRowRefs，分别服务于单行、重复行和 ASOF Join。

DuckDB：Pointer Table 与 TupleDataCollection

DuckDB 的 JoinHashTable 把 Build 行放在 TupleDataCollection 中，Pointer Table 的 Entry 保存行指针和部分 Hash Salt。

Probe 时先检查 Salt，只有 Salt 相符的候选才做完整 Key 比较。碰撞使用线性探测，而相同 Key 的多行通过 Tuple 尾部的 Next Pointer 串起来。这种设计有三个效果：

1. Bucket Entry 保持紧凑，先用 Salt 过滤掉大量不可能匹配的行；
2. Key 与 Payload 采用统一 Tuple 布局，匹配后可以向量化 Gather；
3. 数据本体与 Pointer Table 分离，External Join 可以分区保存 Tuple，再为当前分区重建 Pointer Table。

重复 Key、NULL 与 Join 语义

Join 类型决定需要保存重复链、匹配标记还是仅判断存在，NULL 语义又限制可提前丢弃的行，因此哈希表不能只实现键到单值的映射。

重复 Key 不是边缘情况

对 Inner Join，一个 Probe Key 可能命中多个 Build Row。执行器不能假设一批输入会在一次调用中完整输出，因为结果可能超过向量或 Chunk 的最大行数。

因此 Probe 状态通常要保存两个位置：

1probe_row_index  -> 当前处理到哪一条 Probe Row
2build_match_ptr  -> 该 Probe Row 的重复链处理到哪里

DuckDB 的 ScanStructure::AdvancePointers() 会沿 Next Pointer 继续遍历；Doris 的 Probe 模板保存 Probe/Build 索引，在输出 Block 已满时挂起，下次继续。这是向量化 Hash Join 能够处理一对多结果的关键。

Join Type 改变的是状态，不只是输出格式

不同 Join Type 对哈希表提出不同要求：

Doris 在 hashjoin_probe_operator.cpp 中通过 ProcessHashTableProbe<JoinOpType> 模板为多种 Join 语义生成专用路径。DuckDB 则在 Tuple 布局中为 Right/Full Join 追加 found 布尔位，并对 Correlated Mark Join 维护分组计数。

NULL 需要服从 SQL 三值逻辑

普通 = 下，NULL = NULL 不是 TRUE。Build 阶段通常可以过滤不可能匹配的 NULL Key，但 Right/Full Outer Join 仍然必须保留这些 Build 行，以便最后输出未匹配结果。

如果谓词是 IS NOT DISTINCT FROM，NULL 又可以彼此匹配。于是 NULL 处理不能只写在 Hash 函数里，而必须同时进入：

• Key 编码；
• Build 行保留策略；
• Probe 匹配判断；
• Mark/Anti Join 的三值逻辑（Three-Valued Logic）；
• Spill 分区后的恢复逻辑。

并行 Build 与 Pipeline 化 Probe

Hash Join 天然存在 Build Barrier：Probe 必须等到可用的 Build 状态建立后才能开始。现代执行引擎主要优化 Barrier 两侧的并行度（Degree of Parallelism），而不是假装它不存在。

两种常见并行 Build 方案

并行 Build 可以使用局部表合并或分区协作，前者增加合并与峰值内存，后者需要协调分区所有权。

1方案 A：线程本地 Hash Table
2  local build ─┐
3  local build ─┼─► merge/finalize ─► shared probe
4  local build ─┘
5
6方案 B：按 Hash Shard 并行 Build
7  input block ─► hash scatter ─► shard 0 / shard 1 / ...
8                                  └─ 各线程独占部分 Bucket

线程本地表写冲突少，但 Finalize/Merge 可能昂贵；共享表省掉合并，但要控制并发插入、扩容和内存分配竞争。

ClickHouse 的 ConcurrentHashJoin 给出了一个很有代表性的折中：Build 时按 Hash 把 Bucket 分给不同 Slot，每个线程只修改自己的 Two-Level Map Bucket；Build 完成后把 Bucket 移入公共 Map，Probe 线程读取同一张共享表，从而避免每批 Probe 数据再次 Scatter。

Probe 的向量化流水线

一批 Probe 数据通常经历：

1表达式计算
2  -> NULL 过滤/选择向量
3  -> 批量 Hash
4  -> Bucket 定位
5  -> Salt/Key 比较
6  -> 重复链展开
7  -> Build Payload Gather
8  -> Residual Predicate
9  -> 输出 Chunk

这里有两个性能重点：

1. 先使用 Selection Vector 压缩候选，再做昂贵比较；
2. 匹配阶段只传递 Build Row 的位置，最后按输出列 Gather，避免提前复制完整行。

外存协议：为什么落盘后必须重新建立关系

为什么 Hash Table 不能直接 Spill

当内存不足时，一个看似自然的想法是“把哈希表写入磁盘，稍后读回来”。但内存哈希表常包含：

• 指向 Arena 或 Block 的裸指针；
• 依赖当前地址空间的 Next Pointer；
• 根据 Capacity 构造的 Bucket 数组和 Bitmask；
• 与当前分配器绑定的字符串内存；
• Build Row 的并发命中标记。

这些对象即使序列化成功，恢复到不同地址后也未必有效。而且随机读取磁盘 Bucket 会把一次顺序扫描变成大量随机 I/O。

所以主流实现 Spill 的不是 Bucket，而是可重新构建哈希表的逻辑数据：

 1In-memory Hash Table
 2  └─ revoke
 3      └─ logical build rows / column blocks
 4          └─ hash partition
 5              └─ sequential spill files
 6
 7restore one partition
 8  └─ rebuild local Hash Table
 9      └─ read matching probe partition
10          └─ normal vectorized probe

这也解释了为什么可 Spill Hash Join 不是给普通 Hash Join 增加一个 write() 方法，而是额外建立一套分区、文件、异步 I/O、恢复和状态调度系统。

Grace/Hybrid Hash Join 的正确性基础

设分区函数为：

1partition_id = P(hash(join_key), level)

Build 行和 Probe 行只有在满足以下不变量时才会进入同一分区：

1Hash_build(key) == Hash_probe(key)
2P_build(hash, level) == P_probe(hash, level)

一个完整的外存 Join 通常分成五步：

1. Build 数据按 Hash 分区；
2. 内存允许时保留部分分区，其余顺序落盘；
3. Probe 数据使用相同规则分区；
4. 对每个分区恢复 Build 数据并重建局部哈希表；
5. 读取对应 Probe 分区执行普通 Probe，完成后释放该局部状态。

如果某个 Build 分区仍然太大，则增加分区位或更换层级，再对 Build/Probe 两侧同步重分区。这就是 Recursive Grace Hash Join。

Hybrid 的含义是：并非所有数据都必须落盘。能够留在内存的分区直接完成 Join，可以少写一次 Build、少写一次 Probe，也少读两次文件。

源码阅读坐标：四层架构不要混在一起

阅读四套代码时，最好始终区分以下四层，否则很容易把“算法选择”“Pipeline 调度”“哈希索引”和“Spill 文件”混成一个概念：

后续四章都按同一顺序展开：先给出入口和对象关系，再分析 Hash Table 布局，然后进入 Build/Probe 热路径，最后讨论 Spill 状态机。这样能够看清“相同问题如何被不同架构拆分”，而不仅是对照类名。

四套引擎如何撤销、分区与恢复

StarRocks：从可撤销哈希表到多分区恢复

StarRocks 的普通 Hash Join 由 JoinHashTable 承担数据结构选择，Pipeline Build/Probe Operator 负责调度。可 Spill 路径位于：

• spillable_hash_join_build_operator.cpp
• spillable_hash_join_probe_operator.cpp

架构路径：Operator、Controller 与数据结构分层

StarRocks 没有把所有逻辑都塞进 HashJoinNode。Pipeline 执行路径可以沿下面的对象关系阅读：

 1HashJoinBuildOperator                 HashJoinProbeOperator
 2  │ push_chunk(build)                   │ push_chunk(probe) / pull_chunk()
 3  ▼                                     ▼
 4HashJoiner  <────── build dependency ───────┐
 5  │                                        │
 6  ├─ HashJoinBuilder                       ├─ HashJoinProber
 7  │    ├─ SingleHashJoinBuilder            │    ├─ SingleHashJoinProberImpl
 8  │    └─ AdaptivePartitionHashJoinBuilder │    └─ PartitionedHashJoinProberImpl
 9  │                                        │
10  └────────────── JoinHashTable ───────────┘
11                    │
12                    └─ JoinHashMap<LogicalType,
13                                   KeyConstructor,
14                                   HashMapMethod>

各层职责非常明确：

• HashJoinBuildOperator 和 HashJoinProbeOperator 对接 Pipeline 的 push/pull/set_finishing 协议；
• HashJoiner 更像 Controller，维护依赖、阶段、Runtime Filter、残余谓词和 Lazy Materialization；
• HashJoinBuilder/HashJoinProber 决定单表还是 Cache-Aware 分区表；
• JoinHashTable 负责 Build 数据所有权、Key/Map 特化和 Join 语义；
• JoinHashMap<LT, CT, MT> 是最终进入热循环的模板实例，LT、Key 构造方式和 Map 实现都已静态确定。

HashJoiner 的阶段不是注释意义上的概念，而是控制 Pipeline 可执行操作的状态机：

 1// 简化自 hash_joiner.h
 2enum HashJoinPhase {
 3    BUILD,       // 只允许追加 Build Chunk
 4    PROBE,       // Hash Table 已只读，允许 Probe
 5    POST_PROBE,  // 扫描未匹配/已匹配的 Build Row
 6    EOS
 7};
 8
 9bool has_post_probe(join_type) {
10    return RIGHT_OUTER || RIGHT_ANTI || FULL_OUTER;
11}

Build Operator 在 set_finishing() 中完成 build_ht()、Runtime Filter 构建并唤醒 Probe Dependency；Probe Operator 第一次获得输入时，通过 _reference_builder_hash_table_once() 引用只读表。Right/Full 类 Join 在 Probe 输入结束后进入 POST_PROBE，调用 probe_remain() 输出 Build 侧剩余行。

因此 StarRocks 的 Barrier 是：

1build push
2  -> build_ht
3  -> create runtime filters
4  -> notify probe dependency
5  -> probe references read-only HT
6  -> probe push/pull
7  -> optional post-probe scan

JoinHashTable：列数据、行号链和哨兵行

JoinHashTable 的数据本体仍是列式 build_chunk，哈希索引主要由 first 和 next 两个 uint32_t 数组构成：

1build_chunk
2  row 0      : dummy/default row
3  row 1..N   : real build rows, columnar storage
4
5first[bucket] = bucket chain 的第一条 build row id
6next[row_id]  = 同一 bucket 的下一条 build row id
70             = 链表结束，同时可表示 outer join 的 NULL build row

第 0 行是一个关键约定。_init_build_column() 为每个 Build Column 先追加 Default 值；之后真实数据从 Row ID 1 开始。这样 Probe 输出 (probe_row, build_row=0) 时，可以直接从 Default/Nullable Column 取到 Outer Join 所需的 NULL，而不必在热循环中为“未匹配”维护完全不同的数据路径。

源码中的内存统计也揭示了真实 Hash Table 的组成：

1// 简化自 JoinHashTable::mem_usage()
2usage  = build_chunk->memory_usage();         // Build 数据本体
3usage += first.capacity() * sizeof(uint32_t); // Bucket head
4usage += next.capacity()  * sizeof(uint32_t); // Row chain
5usage += build_pool->reserved_bytes();        // 变长 Build Key
6usage += probe_pool->reserved_bytes();        // Probe 临时序列化 Key
7usage += build_key_column->memory_usage();
8usage += build_slice.size() * sizeof(Slice);

这说明 StarRocks 的 Map Cell 并不保存完整 Payload。哈希表给出 Build Row ID，输出阶段再从列式 build_chunk 按索引 Gather。其优点是索引紧凑、输出列可裁剪；代价是 Build Chunk 生命周期必须覆盖整个 Probe/Post-Probe，并且 Spill 时不能只写 first/next。

两级选择：先构造 Key，再选择索引结构

JoinHashMapSelector 的设计重点是把 Key 表示与 Hash Map 形态解耦。

第一步 _determine_key_constructor()：

 1// 伪代码，分支与源码一致
 2if (只有一个普通等值 Key) {
 3    if (String 实际最大宽度 <= 4/8/16 && 末尾不存在歧义 0 字节)
 4        return SERIALIZED_FIXED_SIZE_INT/BIGINT/LARGEINT;
 5    return ONE_KEY;
 6}
 7
 8for (每个组合 Key) {
 9    if (定长连续类型，或可证明为短 String)
10        total_bytes += field_bytes + null_safe_flag_byte;
11    else
12        return SERIALIZED_VARCHAR;
13}
14return pack_to_32/64/128_bits_when_possible;

这里 NULL-safe equal 要额外写入空值标记，否则 NULL 与底层默认值会被编码成同一个 Key。短 String 特化还要排除尾部 \0 可能造成的编码歧义。

第二步 _determine_hash_map_method()：

 1// 简化自 _determine_hash_map_method()
 2if (BOOLEAN/TINYINT/SMALLINT)
 3    return DIRECT_MAPPING;
 4
 5if (ONE_KEY && (INT || BIGINT))
 6    if (value range 足够小且内存模型更优)
 7        return RANGE_DIRECT_MAPPING / DENSE_RANGE_DIRECT_MAPPING;
 8
 9if (bucket size 在实现上限内)
10    return LINEAR_CHAINED[_SET];
11
12return BUCKET_CHAINED; // 通用回退；ASOF 使用专用 chained 版本

Range Direct Mapping 不是只比较 max-min 与行数。源码把 L2/L3 Cache 和 Join Type 一起放入成本判断：

• Semi/Anti 且没有 Other Conjunct 时，只需表达“Key 是否存在”，可以用 Bitset Set；
• 普通 Join 要保存重复关系，比较 Direct Map 的值域内存与 first/next 成本；
• Dense Range 使用约 2 bit 状态表示值域位置，再保留按行的 first，用于兼顾稠密度和重复行。

最终 Build 通过双重 Dispatch 生成具体模板：

 1// 简化自 JoinHashTable::build()
 2tie(key_constructor, map_method) = selector.construct(...);
 3
 4dispatch_join_hash_map(key_constructor, map_method, [&]<CUT, MUT>() {
 5    using LT = logical_type_of<CUT>;
 6    using CT = key_constructor_of<CUT>;
 7    using MT = map_method_of<MUT>;
 8    hash_map = make_unique<JoinHashMap<LT, CT, MT>>(table_items, probe_state);
 9});
10
11hash_map->build_prepare(state);
12hash_map->probe_prepare(state);
13hash_map->build(state);

外层运行时只做一次 Variant Dispatch，内层 Probe 热路径拿到的是确定的 LT/CT/MT，避免逐行判断 Key 类型和 Map 类型。

Probe：可暂停的重复链与延迟物化

SingleHashJoinProberImpl 对一批 Probe Chunk 保存三个核心对象：

• _probe_chunk：原始 Probe 列；
• _key_columns：表达式计算后的 Join Key；
• _current_probe_has_remain：当前批次是否还有重复匹配未输出。

 1// 简化自 hash_join_components.cpp
 2push_probe_chunk(chunk) {
 3    _probe_chunk = chunk;
 4    prepare_probe_key_columns(_key_columns, _probe_chunk);
 5}
 6
 7probe_chunk() {
 8    hash_table->probe(keys, probe_chunk, output, &has_remain);
 9    filter_probe_output_chunk(output);      // Other/Where Conjunct
10    lazy_output_chunk(output);              // 最后补齐延迟列
11    if (!has_remain) _probe_chunk = nullptr;
12}

has_remain 让哈希表能够在输出 Chunk 达到容量时暂停重复链，下次 pull_chunk() 继续，而不重新 Hash 已处理的 Probe Row。Residual Predicate 先在只包含必要列的中间 Chunk 上计算，只有通过过滤的行才物化 Lazy Column，这对宽表 Join 尤其重要。

Right Outer、Right Anti 和 Full Outer 的 Build 命中状态不能在普通 Probe 输出结束时丢弃。它们通过 POST_PROBE 再扫描哈希表，probe_remain() 分批输出命中或未命中的 Build Row。

Adaptive Partition Hash Join：为 Cache 分区，不是为 Spill 分区

StarRocks 还有一个容易与 External Spill 混淆的 AdaptivePartitionHashJoinBuilder。它的目标不是把数据写盘，而是让若干子 Hash Table 尽量落入 L2/L3 Cache。

源码成本模型估算一次 Probe 会触碰：

18 bytes(first + next)
2+ Join Key 比较字节
3+ Build 输出列字节

同时估算把 Probe Row Scatter 到 16 个分区的成本。只有“减少 Cache Miss 的收益”大于“Probe Shuffle 的代价”时，才保留分区模式：

1BUFFERING
2  ├─ 行数进入收益区间 ─► APPENDING，按 CRC32 + fmix32 分到 16 个 Builder
3  └─ 最终不划算 ──────► 合并成 SingleHashJoinBuilder
4
5APPENDING
6  ├─ 每 8 个 Chunk 重新估算每行 HT 字节
7  └─ 收益消失 ────────► _convert_to_single_partition()

Broadcast Join 还会根据最大 Probe DOP 下调 Cache Miss Factor，因为多个 Prober 共享同一张表，跨核 Cache 复用会改变分区收益。

因此 StarRocks 实际存在两套“Partition”：

Spill 不是默认路径，而是内存撤销后的模式切换

Build 初期仍走普通内存哈希表。收到内存撤销请求后，Operator 把策略切换为 SPILL_ALL。第一次切换时需要完成两个动作：

1. 根据当前 ht_mem_usage() 与 Spill MemTable 阈值估算分区数量；
2. 通过 _convert_hash_map_to_chunk() 把已建哈希表恢复为可分区的 Chunk，再与后续输入一起 Spill。

这一步很重要：系统没有在查询开始时就为最坏情况支付全部 Spill 成本，而是在真正发生内存压力时，把已经建立的不可撤销结构转换成可撤销数据。

Build Spill 采用按列格式，并给数据追加分区 Hash。列式写出更符合 StarRocks 的 Chunk 模型，也便于压缩和恢复。

Probe 侧复用 Build 分区布局

Probe Operator 从 Build Spiller 获取完整分区列表，并用它初始化 Probe Spiller。新到达的 Probe Chunk 计算相同 Hash 后有两条路径：

• 对应 Build 分区已经选入当前处理集合：直接送给相应 Prober；
• 对应 Build 分区尚未加载：写入相同 Partition ID 的 Probe Spill。

于是内存分区可以边接收 Probe 边执行，落盘分区等后续恢复。这正是 Hybrid Hash Join 的效果。

一次加载多少分区由可用内存决定

_acquire_next_partitions() 不只选择一个分区。它优先处理仍在内存中的 Build 分区，再按可用字节选择一个或多个落盘分区。对每个选中的分区分别创建 Builder/Prober，异步读取 Build 数据并重建局部哈希表。

1available memory
2  ├─ partition 1: 420 MiB ─► local HT 1
3  ├─ partition 7: 180 MiB ─► local HT 2
4  └─ partition 9: 90 MiB  ─► local HT 3
5
6probe rows ─► partition id ─► prober 1 / 2 / 3 / spill

这让小分区可以合批并行恢复，降低“一次只处理一个小分区”造成的 CPU 和 I/O 空闲。

Spill 与 Runtime Filter 的冲突

普通 Build 完成后，StarRocks 可以发布 IN Filter 或 Bloom Filter。但 Spill Build 当前会发布空 Runtime Filter。源码注释给出的原因很实际：若要构建全局 Filter，需要重新读取全部 Spill 数据，或者在切换前就知道完整 Build 规模。

这揭示了一个普遍矛盾：

Runtime Filter 希望尽早看到完整 Build Key 集合，Spill 希望尽早释放 Build 状态；两者对生命周期的要求相反。

可行方向包括分区级 Filter、可增量合并的 Bloom Filter，或在分区元数据中维护低成本摘要，但都要权衡误判率、发布时间和额外内存。

Apache Doris：把递归 Spill 做成显式状态机

Doris 的普通 Build/Probe 路径位于 hashjoin_build_sink.cpp 与 hashjoin_probe_operator.cpp。可 Spill 路径由 Partitioned Hash Join 包装普通 Hash Join：

• partitioned_hash_join_sink_operator.cpp
• partitioned_hash_join_probe_operator.cpp

架构路径：外层 Operator 与内层 Hash Join Kernel

Doris 的普通路径可以按下面的调用链阅读：

 1HashJoinBuildSinkOperatorX
 2  └─ HashJoinBuildSinkLocalState::process_build_block()
 3       ├─ 执行 Build Key 表达式
 4       ├─ init_hash_method<JoinDataVariants>()
 5       ├─ try_convert_to_direct_mapping()
 6       └─ ProcessHashTableBuild<HashTableCtx>::run<JoinOp>()
 7            └─ JoinHashTable<Key, Hash, DirectMapping>::build()
 8
 9HashJoinProbeOperatorX
10  └─ HashJoinProbeLocalState
11       └─ ProcessHashTableProbe<JoinOp>::process<HashTableCtx>()
12            ├─ init_serialized_keys()
13            ├─ JoinHashTable::find_batch<JoinOp>()
14            ├─ 生成 probe_indexs/build_indexs
15            ├─ Gather 两侧输出列
16            └─ Other Join Conjunct / Mark Conjunct

这里使用了两次正交的模板化：

• HashTableCtx 固化 Key 类型、Hash Method、是否 Nullable/Serialized/Direct Mapping；
• JoinOp 固化 Inner、Outer、Semi、Anti、Null-Aware 等语义。

std::visit 只发生在 Block 级入口，真正的逐行热循环已经是具体模板实例。这与 StarRocks 的 Key Constructor/Map Method Dispatch 目标相同，但 Doris 把 Join Type 也更直接地下沉到 ProcessHashTableBuild/Probe 模板。

可 Spill 路径不是替换这个 Kernel，而是在外面再包一层：

1PartitionedHashJoinSinkOperatorX
2  ├─ 未 Spill：转发给 inner HashJoinBuildSinkOperatorX
3  └─ 已 Spill：Build Block -> partition -> SpillFile[]
4
5PartitionedHashJoinProbeOperatorX
6  ├─ 未 Spill：转发给 inner HashJoinProbeOperatorX
7  └─ 已 Spill：Probe Block -> partition -> SpillFile[]
8                 -> recover one pair
9                 -> recreate inner build/probe operators

JoinHashTable：first/next/visited 三个数组

Doris 的 JoinHashTable 是理解普通 Join 的最短入口。它没有把 Payload 塞进 Cell，而是使用 Build Block 行号：

1// 源码数据成员的语义化摘录
2DorisVector<uint32_t> first;   // bucket -> first build row
3DorisVector<uint32_t> next;    // build row -> next row in bucket
4DorisVector<uint8_t>  visited; // Right/Full/Semi/Anti 的逐行命中状态
5const Key* build_keys;          // 完整 Key，碰撞后做相等比较

Build Block 同样保留第 0 行作为 Mock Row，真实行从 1 开始。prepare_build() 根据 Join Type 决定是否分配 visited：

 1// 简化自 prepare_build<JoinOpType>()
 2bucket_size = DirectMapping
 3    ? force_bucket_size
 4    : normalize((num_elem + 1) * 8 / 7); // 预留装载率空间
 5
 6first.resize(bucket_size + 1); // 额外一个位置保存 NULL Key 链头
 7next.resize(num_elem);
 8
 9if (FULL_OUTER || RIGHT_OUTER || RIGHT_ANTI || RIGHT_SEMI)
10    visited.resize(num_elem);

插入本身非常直接：

1for (uint32_t row = 1; row < num_elem; row++) {
2    uint32_t bucket = bucket_nums[row];
3    next[row] = first[bucket];
4    first[bucket] = row;
5}
6
7if (!keep_null_key)
8    first[bucket_size] = 0; // bucket_size 这个额外槽专门表示 NULL 链

因此 Doris 的冲突处理是 Separate Chaining，但链节点不是堆对象，而是稠密 uint32_t next[]。这比每行分配 Node 更节省内存，也让 Prefetch 和顺序访问更可控。

Direct Mapping 时，Bucket 已由 Key 唯一确定，_eq() 可以直接返回 true；普通 Hash 路径仍要用 build_keys[row] 做完整比较，避免 Hash 冲突产生错误匹配。

Build：Key 初始化、NULL 编码与批量插入

Build 的关键入口是 process_build_block()。可以把它拆成六步：

 1// 结构化伪代码，保留源码调用关系
 2convert_overflow_columns(block);                    // String/Variant 物理格式稳定化
 3extract_join_column(block, null_map, raw_ptrs);    // 计算并拆出 Build Key
 4hash_table_init(state, raw_ptrs);                  // 选择 HashMethod/DirectMapping
 5
 6visit(hash_table_variant, join_op, [&](ctx, op) {
 7    ProcessHashTableBuild<Ctx> process(rows, raw_ptrs, ...);
 8    process.run<op>(ctx, null_map,
 9                    &has_null_in_build_side,
10                    short_circuit_for_null,
11                    have_other_join_conjunct);
12});

ProcessHashTableBuild::run() 内部再做：

1prepare_build<JoinOp>()
2  -> 如有必要，把单列 NULL 替换为默认值
3  -> init_serialized_keys(..., is_build=true)
4  -> 计算 bucket_nums
5  -> hash_table.build(keys, bucket_nums, rows, keep_null_key)

NULL 有两种表示策略：

• Null-Safe Equal 需要把 NULL 编入 Key，使 NULL <=> NULL 可匹配；
• 普通等值 Join 用外部 Null Map 排除 NULL，并在 Null-Aware/Outer 等特定语义下保留 NULL 行。

short_circuit_for_null 是 Null-Aware Anti Join 的重要优化：当语义允许仅凭 Build 侧存在 NULL 就确定结果时，不再无意义地构建完整哈希表。

Probe：行号向量把“搜索”与“输出”分开

ProcessHashTableProbe<JoinOp> 不会在命中时立即复制整行，而是先填两个索引向量：

1probe_indexs = [0, 0, 2, 5, ...]
2build_indexs = [7, 9, 3, 0, ...]
3
4(0, 7), (0, 9) -> Probe Row 0 命中两个 Build Row
5(5, 0)         -> Probe Row 5 未命中，Outer Join 输出 NULL Build Row

随后 probe_side_output_column() 与 build_side_output_column() 才按索引批量 Gather。若 Probe 索引连续、整批已消费且没有未完成的 Build 重复链，源码甚至直接转移 Probe Column 所有权，避免一次复制。

Join Type 在 find_batch<JoinOp>() 中编译期分派：

1if constexpr (INNER || LEFT/RIGHT/FULL_OUTER)
2    find_batch_inner_outer_join<JoinOp>();
3else if constexpr (LEFT_SEMI || LEFT_ANTI || NULL_AWARE_LEFT_ANTI)
4    find_batch_left_semi_anti<JoinOp>();
5else if constexpr (RIGHT_SEMI || RIGHT_ANTI)
6    mark_build_rows_visited();

Inner/Outer 的主循环可以概括为：

 1while (build_idx && output_not_full) {
 2    if (probe_key == build_keys[build_idx]) {
 3        probe_indexs.push(probe_idx);
 4        build_indexs.push(build_idx);
 5        if constexpr (RIGHT/FULL) visited[build_idx] = 1;
 6    }
 7    build_idx = next[build_idx];
 8}
 9
10// Left/Full 未命中时写 build_idx=0，复用 Mock Row

HashJoinProbeLocalState::_probe_index 和 _build_index 保存跨 Block Pull 的断点。当一条 Probe Row 的重复链产生超过 Batch Size 的结果时，下一次调用从同一个 build_idx 继续，而不是重新查 Bucket。

Residual Conjunct 会使流程多一层：先产生候选 (probe_idx, build_idx)，再执行非等值条件，最后根据结果修正 Outer/Semi/Anti 语义。换言之，Hash 只负责 Equi-Key 候选集，SQL 的完整 ON 条件仍在其后。

正常路径与 Spill 路径复用同一个内核

Partitioned Operator 初始仍把输入交给内部普通 Hash Join。发生内存撤销后，Build/Probe 数据按配置的 Partition Count 分区并写入 Spill File。

恢复时，Doris 为当前 Build Partition 建立内部 Runtime State，再复用普通 Hash Join 完成：

1recover build partition
2  -> internal HashJoin build
3  -> stream corresponding probe partition
4  -> internal HashJoin probe
5  -> release

这种“外层负责 External 调度，内层保持 In-Memory 内核”的分层减少了两套 Join 语义不一致的风险。Inner、Outer、Semi、Anti 等复杂语义仍由同一个 Probe 模板完成。

工作队列显式保存递归层级

Build 和 Probe 初次 Spill 完成后，系统把对应文件加入处理队列。队列项可以抽象成：

1PartitionTask {
2  build_file,
3  probe_file,
4  level
5}

如果恢复后的 Build Partition 仍然无法在预算内建表，repartition_current_partition() 会创建下一层的 Build/Probe Spill File，使用新的分区层级对两侧同时重分区，再把子任务放回队列。

1level 0 partition too large
2  ├─ level 1 / subpartition 0
3  ├─ level 1 / subpartition 1
4  ├─ ...
5  └─ level 1 / subpartition N-1

递归深度由 spill_repartition_max_depth 限制。达到最大深度仍无法放入内存时，系统返回明确错误，而不是无限生成 Spill 文件。

恢复状态机与二次内存撤销

_pull_from_spill_queue() 将每个队列项处理成一个小型 Pipeline：

 1SETUP PARTITION
 2  -> 分批 read build_file 到 recovered_build_block
 3  -> 达到 spill_buffer_size 时主动 yield
 4  -> 文件读完后创建 inner RuntimeState/BuildSink/ProbeOperator
 5  -> 用 recovered_build_block 完成 inner build
 6
 7PROBE PARTITION
 8  -> 分批 read probe_file 到 queue_probe_blocks
 9  -> push 给 inner ProbeOperator
10  -> pull output
11  -> inner EOS 后释放当前分区，处理下一项

这里的 yield 是关键设计：恢复文件不能在一次调度中无限读取，否则既占用 Pipeline Worker，也可能在 Hash Table 建立前就把整个分区物化进内存。

更进一步，Doris 允许在恢复 Build Partition 的中途再次收到内存撤销。此时 repartition_current_partition() 必须同时处理两部分数据：

1already recovered rows in _recovered_build_block
2+ unread remainder behind _current_build_reader

源码先把已恢复 Block 路由到下一层子分区，再让 SpillRepartitioner 从 Reader 的当前位置继续，避免从文件头重读导致重复行。Probe 文件也使用同一新 Level 和 Fanout 重分区，最后把所有子文件重新放回工作队列。

恢复前的内存预留也显式包含：

1spill I/O baseline
2+ first[bucket_count] * 4
3+ next[build_rows] * 4
4+ optional visited[build_rows]
5+ serialized key estimate

这比只按 recovered_build_block.allocated_bytes() 申请内存更接近真正的 Build 峰值。

Spill 后为什么停用 Runtime Filter

从普通 Hash Join 切换到 Partitioned 模式时，_revoke_unpartitioned_block() 会调用 runtime_filter_producer_helper->skip_process()。原因与 StarRocks 类似：原先的 Build 状态被拆散到文件，继续发布一个未覆盖全部 Build Key 的 Filter 会产生假阴性，直接破坏正确性。

这也说明 Runtime Filter 不是 Hash Join 的附属统计，而是带正确性约束的 Build 输出。要在递归 Spill 中保留它，需要 Filter 本身能够跨 Partition/Level 无损合并，并且只有全局完成后才能作为排除型过滤器发布。

可观测性围绕退化路径设计

Doris 暴露 SpillMaxPartitionLevel、分区总数、恢复 Build/Probe 行数与耗时等指标。相比只记录 Spill Bytes，这些指标更接近问题本质：

• Level 持续升高，通常意味着分区估算不足或数据倾斜；
• Build 恢复时间高，可能是磁盘吞吐、解压或重建哈希表成为瓶颈；
• Probe 恢复行数远大于有效输出，说明过滤能力或 Build/Probe 选择可能不佳。

ClickHouse：Hash、Parallel Hash 与 Grace Hash

ClickHouse 将不同资源假设直接呈现为不同 Join Algorithm：

从 Planner 到 IJoin：算法是显式策略对象

ClickHouse 的选择入口位于 chooseJoinAlgorithm()。Planner 不只生成一个枚举，而是创建实现统一 IJoin 接口的策略对象：

 1// 简化自 tryCreateJoin()
 2switch (algorithm) {
 3case DIRECT:          return tryDirectJoin(dictionary_or_key_value_storage);
 4case HASH:            return make_shared<HashJoin>(...);
 5case PARALLEL_HASH:   return make_shared<ConcurrentHashJoin>(..., max_threads);
 6case GRACE_HASH:      return make_shared<GraceHashJoin>(initial_buckets,
 7                                                        max_buckets, ...);
 8case PARTIAL_MERGE:   return make_shared<MergeJoin>(...);
 9case FULL_SORTING_MERGE:
10                       return make_shared<FullSortingMergeJoin>(...);
11case AUTO:            return supported_merge
12                            ? make_shared<JoinSwitcher>(...)
13                            : make_shared<HashJoin>(...);
14}

parallel_hash 还会参考右侧大小估计和 parallel_hash_join_threshold；grace_hash 则显式携带初始/最大 Bucket 数和临时存储 Scope。因此 ClickHouse 的架构不是“一种 Hash Join 内部自动切换所有模式”，而是：

1Planner/TableJoin
2  -> chooseJoinAlgorithm
3      -> IJoin
4          ├─ HashJoin
5          ├─ ConcurrentHashJoin (内部包含多个 HashJoin)
6          └─ GraceHashJoin      (当前 Bucket 内部包含一个 HashJoin)

ConcurrentHashJoin 和 GraceHashJoin 都复用 HashJoin 作为内存内核：前者解决 Build 并行，后者解决外存分桶。

RightTableData：Map、Block 与 Arena 的所有权

ClickHouse 的 HashJoin::RightTableData 把索引与右表数据分开：

 1// 语义化摘录
 2struct RightTableData {
 3    Type type;                        // key32/key64/keys128/hashed/...
 4    vector<MapsVariant> maps;         // ON 中每个 OR disjunct 一张 Map
 5    ScatteredColumnsList columns;     // 右表 Block 的列所有权
 6    NullmapList nullmaps;             // Right/Full 的 NULL/ON 未命中行
 7    Arena pool;                       // String Key、RowRefList 扩展节点
 8    size_t rows_to_join;
 9    size_t keys_to_join;
10};

MapsVariant 的 Value 由 Join Strictness 决定：

1MapsOne  : Key -> RowRef       // ANY，只保留一个右表引用
2MapsAll  : Key -> RowRefList   // ALL，保留同 Key 的全部行
3MapsAsof : Key -> AsofRowRefs  // ASOF，Key 下再维护可检索结构

RowRef 不是复制 Payload，而是引用 ColumnsInfo + row_num。ScatteredColumns 还保留 Selector，允许 Concurrent Hash Join 的不同 Slot 共享原始 Block，只记录本 Slot 拥有哪些行。

这个所有权模型解释了三个源码现象：

1. 右表 Block 必须一直保存到 Probe/未匹配行输出结束；
2. String Key 和重复链扩展节点由 Arena 统一释放，避免大量小对象析构；
3. Right/Full Join 的 NULL Key 不能插入普通 Map，却要通过 nullmaps 留到最后输出。

Key Method、Build 与 Probe 的双重 Dispatch

chooseMethod() 先根据右侧实际列选择 Key Map：

1if (one_numeric_key)       return key8/key16/key32/key64;
2if (all_fixed && bytes<=16) return keys128;
3if (all_fixed && bytes<=32) return keys256;
4if (one_string)             return key_string;
5if (one_fixed_string)       return key_fixed_string;
6return hashed; // 对复杂组合键生成无歧义序列化后的 UInt128 Hash

Build 入口 addBlockToJoin() 的核心路径是：

1materialize right columns
2  -> 提取所有 ON disjunct 的 Key Column
3  -> 保存需要输出的右表列为 ScatteredColumns
4  -> 从 Key 中拆出 NullMap
5  -> joinDispatch(kind, strictness, map)
6  -> HashJoinMethods<kind, strictness, map>::insertFromBlockImpl()

这里 joinDispatch() 同时分派 JoinKind × JoinStrictness × MapsOne/All/Asof，所以具体插入函数知道是否只留一行、是否保留重复、是否需要 Used Flag。

Probe 入口 joinBlock() 以同样方式分派：

1joinDispatch(kind, strictness, maps, [&](auto kind, auto strictness, auto typed_maps) {
2    result = HashJoinMethods<kind, strictness, TypedMaps>::joinBlockImpl(
3        left_block, columns_to_add, typed_maps);
4});

这种模板爆炸换来的收益是热循环没有 switch(join_kind)。同时 JoinResultPtr 允许一次 Left Block 产生多次输出：next() 返回当前结果、未完成的 Block 和 is_last，用于控制 ALL Join 的重复展开。

Build 完成后还有一个数据相关优化：如果 ALL INNER/LEFT 的右侧 Key 实际唯一，all_values_unique 会保持为真，系统可以把 ALL 提升为 RightAny 路径，消除无意义的 RowRefList 遍历。

ConcurrentHashJoin：Build 分片，Probe 尽量共享

ConcurrentHashJoin 创建最多 256、且为 2 的幂的 Slot。Build Block 先计算与 Hash Map 一致的 Hash，再路由到不同 Slot：

1right block
2  -> calculate hash/bucket
3  -> selector = bucket & (slots - 1)
4  -> ScatteredBlock[slot]
5  -> slot-local HashJoin::addBlockToJoin()

Two-Level Map 把一张逻辑 Map 划成固定数量的 Submap Bucket。每个 Slot 只写 slot, slot+N, slot+2N... 这些 Bucket，因此 Build 阶段不需要让所有线程争用同一张普通 Hash Map。

onBuildPhaseFinish() 不是逐行重新插入，而是把其他 Slot 的 Submap Bucket 移到 Slot 0：

1for (slot = 1; slot < slots; ++slot) {
2    for (bucket = slot; bucket < NUM_BUCKETS; bucket += slots)
3        common_map.impls[bucket] = move(slot_map.impls[bucket]);
4
5    common_columns.splice(slot_columns); // Block 所有权也转移
6}
7
8for (slot = 1; slot < slots; ++slot)
9    slot.maps = common_maps;              // Probe 共享同一逻辑 Map

因此它试图实现“Build 时分治，Probe 时共享”。如果使用的是 Two-Level Map，Probe Block 无需再次按 Slot Scatter，所有线程可读公共 Map；无法使用 Two-Level 的特例仍需把 Probe Block 路由到各内部 HashJoin。

Right/Full Join 还要合并各 Slot 的 NullMap 和 Used Flag，否则同一右表未匹配行可能被重复输出。这部分状态合并说明 Parallel Hash Join 的难点不只在并发插入，还包括 Build 完成后的语义状态归并。

Grace Hash Join 保留当前 Bucket，其余落盘

GraceHashJoin.h 对三个阶段有非常清楚的说明：

1. Build Block 按 Key Hash 拆到多个 Bucket；当前 Bucket 加入内存 HashJoin，其余写临时文件；
2. Probe Block 同样拆分；当前 Bucket 直接 Probe，其余写对应左侧文件；
3. 逐个读取后续 Bucket 的右侧数据重建 Hash Join，再读取左侧数据完成 Probe。

1right/build block
2  ├─ bucket 0 ─► in-memory HashJoin
3  ├─ bucket 1 ─► right temp file 1
4  └─ bucket N ─► right temp file N
5
6left/probe block
7  ├─ bucket 0 ─► probe now
8  ├─ bucket 1 ─► left temp file 1
9  └─ bucket N ─► left temp file N

超限时动态倍增 Bucket

如果当前内存 Hash Join 超过限制，rehashBuckets() 将 Bucket 数量翻倍，并受 max_num_buckets 约束。因为 Bucket 数保持 2 的幂，增加一位 Hash 即可细分原分区。

动态倍增有两个优点：

• 初始数据不大时，不必创建大量小文件；
• 真正超限后再增加分区，能够根据运行时规模适配。

它也带来状态复杂度：并发线程可能持有旧 Bucket Snapshot，源码需要检测 Bucket 数变化并重新 Scatter 当前 Block；早先写入旧 Bucket 的行，也可能在后续阶段再次分发到新 Bucket。

源码中的溢出处理比“Bucket 数乘二”多两次关键重建：

1current in-memory HashJoin overflow
2  -> rehashBuckets(): bucket_count *= 2
3  -> releaseJoinedBlocks(): 从 HashJoin 取回逻辑右表 Block
4  -> 用新 bucket_count 重新 Scatter
5  -> 当前 Bucket 的行重建一个新 HashJoin
6  -> 其他 Bucket 写各自临时流

这再次印证：扩分区时迁移的是逻辑 Block，不是带地址的 Hash Map Bucket。

临时文件是 Join 状态的一部分

每个 FileBucket 同时拥有左、右临时流，并在 WRITING_BLOCKS、JOINING_BLOCKS、FINISHED 状态间转换。临时文件使用 Native Block 流和压缩配置，并记录压缩/未压缩字节以及 Join 临时文件数。

这不是外围 I/O 工具，而是 Join 状态机的一部分：只有在 Bucket 停止写入后，才能安全切换为读取并 Build；只有其非匹配 Build 行输出完成后，才能释放当前 Hash Join。

后续 Bucket 由 getDelayedBlocks() 串联处理：它找到下一个非空 Bucket，读取右侧文件并调用 addBlockToJoinImpl() 重建内存表，onBuildPhaseFinish() 后返回 DelayedBlocks；后者读取左侧文件、必要时再次按最新 Bucket 数 Scatter，再调用普通 HashJoin Probe。于是 Grace 层只负责文件和 Bucket 生命周期，Join Kind/Strictness 语义继续由内层 HashJoinMethods 保证。

DuckDB：让分区选择服从内存预留

DuckDB 的 physical_hash_join.cpp 与 join_hashtable.cpp 把内存 Hash Join 与 External Hash Join 组织在同一套 JoinHashTable 中。

架构路径：Sink、Finalize、Operator 与 Source

DuckDB 的 Hash Join 同时实现 Pipeline Sink、Operator 和 Source 三种角色：

 1Build child
 2  -> PhysicalHashJoin::Sink()          每线程 Local JoinHashTable
 3  -> PhysicalHashJoin::Combine()       收集 local_hash_tables
 4  -> PrepareFinalize()                 估算总工作集和最坏分区
 5  -> PhysicalHashJoin::Finalize()
 6       ├─ In-memory: merge -> unpartition -> build pointer table
 7       └─ External : repartition -> choose current partitions -> build pointer table
 8
 9Probe child
10  -> ExecuteInternal()
11       ├─ PerfectHash probe
12       ├─ normal JoinHashTable::Probe
13       └─ external ProbeAndSpill
14
15Source phase
16  -> 输出 Right/Full 未匹配 Build Row
17  -> External 模式继续 BUILD -> PROBE -> SCAN_HT 多轮循环

为什么 External Join 还需要 Source？第一轮 Probe 来自上游 Pipeline，可以直接走 ExecuteInternal()；没有进入当前 Build 分区的 Probe Row 已写入 ProbeSpill。之后上游输入已经结束，剩余分区需要由 Hash Join 自己驱动“重建表—读 Probe Spill—输出”，因此转为 Source 状态机。

Build 的线程本地路径也不是立刻并发写同一 Pointer Table：

 1// 简化自 PhysicalHashJoin::Sink/Combine
 2Sink(chunk) {
 3    join_key_executor.Execute(chunk, join_keys);
 4    payload_chunk.ReferenceColumns(chunk, payload_columns);
 5    local_hash_table.Build(append_state, join_keys, payload_chunk);
 6}
 7
 8Combine() {
 9    local_hash_table.GetSinkCollection().FlushAppendState();
10    global.local_hash_tables.push_back(move(local_hash_table));
11}

线程只向本地 RadixPartitionedTupleData 追加；到 Finalize 才决定合并、重分区和并行建立全局 Pointer Table。这把高并发的“数据物化”与更容易产生原子冲突的“索引构建”分开了。

Tuple Layout：Hash 在 Finalize 后变成 Next Pointer

JoinHashTable 构造时创建统一 TupleDataLayout：

 1Tuple row
 2┌──────────────────────┐
 3│ equality conditions  │  完整 Key，最终相等比较
 4├──────────────────────┤
 5│ residual columns     │  非等值 ON 条件需要的 Build 列
 6├──────────────────────┤
 7│ output payload       │  查询真正输出的 Build 列
 8├──────────────────────┤
 9│ optional found flag  │  Right/Full 类 Join 的逐行命中标记
10├──────────────────────┤
11│ hash / next pointer  │  Build 暂存 Hash；Finalize 后改作重复链 Next
12└──────────────────────┘

Build() 先计算 Key Hash，并把它连同 Key/Payload 一起追加到 sink_collection。此时 Tuple 最后一槽保存 Hash，方便 Radix Partition 和 Finalize，尚不存在地址相关状态。

Finalize 遍历已经 Pin 住的 Tuple Block，读取最后一槽的 Hash，把 Tuple 地址插入 Pointer Table；若同 Key 已存在，则最后一槽被覆盖为旧链头地址，成为 Next Pointer。pointer_offset 同时标识这两个阶段的复用位置。

这种布局有两个重要结果：

1. Spill/移动阶段保存的是 Hash，Tuple 仍可安全重定位；
2. 一旦 Finalize 建立指针链，当前 data_collection 必须保持 Pin 住，直到本轮 Probe 完成。

Pointer Table：开放寻址与重复链是两套机制

Pointer Table Entry ht_entry_t 保存 Tuple Pointer 和 Hash Salt。插入时先做：

1offset = hash & bitmask
2salt   = extract_high_hash_bits(hash)

随后进入开放寻址循环：

 1// 结构化自 InsertHashesLoop()
 2while (entry[offset] occupied && entry[offset].salt != salt)
 3    offset = (offset + 1) & bitmask;
 4
 5if (entry empty) {
 6    entry[offset] = {salt, tuple_ptr};
 7} else {
 8    candidate_ptr = entry[offset].pointer;
 9    gather candidate full key;
10
11    if (full_key_equal) {
12        tuple.next = candidate_ptr;       // 相同 Key：接入重复链
13        entry[offset].pointer = tuple_ptr;
14    } else {
15        offset = next_offset;             // 只是 Salt/Hash 碰撞：继续线性探测
16    }
17}

所以要区分：

• 不同 Key 的 Hash 冲突：Pointer Table 中线性探测；
• 相同 Key 的重复行：Entry 指向链头，Tuple 内 Next Pointer 串链。

Salt 是一次廉价预过滤。Salt 不同就不需要访问随机 Tuple；Salt 相同才 Gather 完整 Key 并调用 RowMatcher。表较小时 Salt 收益不够，UseSalt() 可以关闭它。

并行 Finalize 使用 Atomic Entry：抢占（Preemption）空槽时 CAS；把重复行接到已有链头时用 Compare-Exchange 更新 Entry，同时把旧指针写进新 Tuple 的 Next。chains_longer_than_one 记录是否真的出现重复，Probe 可以为 Unique Key 走无链 Fast Path。

Probe：先定位指针，再由 ScanStructure 实现语义

JoinHashTable::Probe() 只完成 Key 准备、Hash 和候选 Pointer 定位：

1PrepareKeys(NULL/filter)
2  -> Hash(keys)
3  -> GetRowPointers(open addressing + salt + full-key compare)
4  -> ScanStructure.pointers[]

真正的 Join Type 语义位于 ScanStructure::Next()：

1switch (join_type) {
2case INNER/RIGHT:       NextInnerJoin(); break;
3case LEFT/OUTER:        NextLeftJoin();  break;
4case SEMI:              NextSemiJoin();  break;
5case ANTI:              NextAntiJoin();  break;
6case MARK:              NextMarkJoin();  break;
7case RIGHT_SEMI/ANTI:   NextRightSemiOrAntiJoin(); break;
8case SINGLE:            NextSingleJoin(); break;
9}

对于 Inner Join，ScanInnerJoin() 对当前 Pointer 批量执行完整 Equality/Residual Predicate；产生结果后 GatherResult() 按 Pointer 从 TupleDataCollection 读取 Build 列。AdvancePointers() 再沿 Tuple Next Pointer 前进，直到所有重复链耗尽。

如果一个输入 Chunk 产生超过 STANDARD_VECTOR_SIZE 的结果，last_match_count/last_sel_vector 保存尚未输出的选择向量；下次 ExecuteInternal() 返回 HAVE_MORE_OUTPUT，继续消费同一输入 Chunk。这个设计把 Pipeline 的有限 Chunk 与 Join 的无限输出放大解耦。

Right/Full Join 的 found 位在匹配时被写为 true，上游 Probe 结束后再通过 Source 扫描 Build Tuple，输出未命中行。源码允许多个线程并发写同一个布尔位，因为唯一写入值都是 true，最终语义是幂等（Idempotency）的。

Build 从一开始就保留 Radix 分区能力

Build Chunk 被组织成：

1[equality keys]
2[non-equality condition columns]
3[payload]
4[optional found flag]
5[hash]
6[next pointer in tuple layout]

数据先进入 RadixPartitionedTupleData。Finalize 时，如果全部数据能够装入内存，就把分区合并并构建一张表；如果预算不足，则增加 Radix Bits，把数据保留为多个分区。

与“内存不足后再从哈希表反解行”相比，这种设计更早保存了可分区数据，但也要求 Build 数据布局从开始就兼顾内存与外存路径。

External 决策同时考虑 Tuple 与 Pointer Table

DuckDB 估算每个分区时，不只计算 Tuple 数据，还加上按行数估算的 Pointer Table：

1partition_ht_size = partition_data_size
2                  + PointerTableSize(partition_count)

如果最大分区超过 max_ht_size，SetRepartitionRadixBits() 会增加分区位，并把目标估算到预算的一部分，为 Hash Table 构建和 Probe 留出余量。

这一点非常重要：只让逻辑行“勉强塞满”内存，会导致 Pointer Table 一分配就 OOM。可靠的 External Join 必须对恢复阶段的完整工作集做预算。

Finalize：由内存 Reservation 决定执行形态

PrepareFinalize() 先计算：

1total_size             = 所有 Tuple + 全局 Pointer Table
2max_partition_ht_size  = 最大分区 Tuple + 该分区 Pointer Table
3probe_side_requirement = Probe Radix Partition 的线程 × 分区 Buffer

然后向 TemporaryMemoryManager 注册最小 Reservation：至少要能处理一个最大 Build Partition，并为 Probe 分区 Buffer 留空间。Finalize() 的核心判断是：

 1external = reservation < total_size;
 2
 3if (external) {
 4    load_factor = EXTERNAL_LOAD_FACTOR; // 更紧凑的 Pointer Table
 5    if (max_partition_ht_size + probe_requirement > reservation)
 6        SetRepartitionRadixBits(...);
 7    PrepareExternalFinalize(reservation - probe_requirement);
 8} else {
 9    merge_local_hts();
10    Unpartition();
11    try PerfectHash;
12    otherwise ScheduleFinalize();
13}

External 模式将 Load Factor 从默认值调整为更紧凑的配置，有时仅靠减少 Pointer Table Capacity 就能重新落回 In-Memory 路径。若仍超限，新增 Radix Bits 的目标不是刚好低于预算，而是把估算最大分区压到约 max_ht_size / 4，为偏差、并行和中间 Buffer 留余量。

如果单个分区已占总大小 80% 以上，源码判断为 very_very_skewed，不再假设增加 Radix Bits 能有效均分。这是把“相同热点 Key 无法被 Hash 位拆开”直接编码进执行决策。

每轮选择一组能装下的分区

PrepareExternalFinalize() 会把未完成分区按大小近似排序，再选择一组总计不超过本轮预算的分区，移动到当前 data_collection 并构建 Pointer Table。

Probe 时，ProbeAndSpill() 根据 current_partitions 把输入分成两类：

• 当前已 Build 的分区立即 Probe；
• 其他分区写入 ProbeSpill，等待后续轮次。

每轮结束后释放当前数据，再选择下一组 Build Partition。这个过程与 StarRocks 的“多分区装箱”相似，但 DuckDB 明确把 TemporaryMemoryState 的 Reservation、Probe Side Requirement 和 Pointer Table 一并纳入选择。

External Source 的全局状态机更完整地表示为：

1INIT
2  -> BUILD current_partitions
3  -> PROBE matching ProbeSpill partitions
4  -> SCAN_HT (仅 Right/Full 等需传播 Build 侧的 Join)
5  -> Reset current HT
6  -> BUILD next_partitions
7  -> ...
8  -> DONE

PrepareExternalFinalize() 会优先按近似大小从小到大选择分区，但对小差异做取整，尽量保留 Partition ID 顺序以降低 Eviction/I/O 抖动；并且即使单个分区超预算也至少选择一个，保证状态机能前进或暴露真实倾斜问题，而不是空转。

倾斜会反向限制并行度

DuckDB 在 Finalize/External Build 阶段检测 Key 是否倾斜。严重倾斜时，它可能改用单线程构建，避免多个线程同时争用同一冲突链或原子 Bucket。

这说明并行度不是越高越好：当 Key 分布高度集中时，增加线程既不能增加有效分区，也会放大同步和 Cache Coherence 成本。

特化与边界：直接映射、过滤及热点键

Direct Mapping 与 Perfect Hash Join

如果单整数 Key 的值域很小且足够稠密，通用哈希表并非最佳结构：

1slot = key - min_key

这可以消除 Hash、碰撞与大部分 Key 比较。StarRocks 的 Range/Dense Range Direct Mapping 和 DuckDB 的 Perfect Hash Join 都在利用这个事实。

DuckDB 的 PerfectHashJoinExecutor 会检查：

• 只有一个 Join 条件；
• Key 是整数类型且有 Min/Max 统计；
• 值域跨度低于阈值；
• Build Key 不存在无法表达的重复关系。

满足条件后，Build Payload 可以按值域直接放入数组，Probe 只需范围判断和偏移寻址。

这类优化的适用边界也很明确：

• 值域很大但数据稀疏时，数组空间会严重浪费；
• Key 重复时，一个 Slot 仍需表示多行；
• 统计错误可能导致原本节省内存的方案反而扩大内存。

因此直接映射不是“更快的通用 Hash Join”，而是基于值域证明成立后的专用物理算子。

Runtime Filter：在 Probe 之前减少工作

Hash Join 的 Build Key 集合天然可以生成过滤器：

• Min/Max Filter：适合有序或范围集中的数值；
• IN Filter：适合基数很小的 Build Key；
• Bloom Filter：适合较大集合，允许假阳性；
• Prefix/Range Filter：适合部分 String 或范围谓词。

Runtime Filter 的收益不只是减少 Probe 哈希查找。如果它能够下推到 Scan，还会减少解码、谓词计算、网络传输和上游 Exchange。

但 Filter 有三个约束：

1. 完整性。 过滤器不能产生假阴性，否则会丢失正确结果；
2. 时机。 等 Build 全部完成再发布，Probe 可能已经读完大量数据；
3. Join 语义。 Outer/Anti/Mark Join 并不都允许用同样方式过滤 Probe 行。

Spill 又增加了第四个约束：Build Key 分散在文件中，系统要么维护可增量合并的全局摘要，要么只做分区级 Filter，要么放弃提前发布。

四个引擎的实现对照

相同的分区 Spill 名称可以对应不同撤销、恢复与调度协议，比较时需要同时看数据布局和状态生命周期。

这张表不意味着某个实现绝对更好。不同设计反映的是不同运行环境：

• 分布式 MPP 引擎更关注 Workgroup、内存撤销、异步 I/O 与 Runtime Filter；
• ClickHouse 把算法选择显式暴露，更方便用户按负载取舍；
• 嵌入式 DuckDB 必须把单进程内的 Buffer Manager、临时内存与并行任务统一调度。

数据倾斜：递归分区的边界

假设一个 Key hot_key 占 Build 侧 60% 数据。无论使用 Hash 的第几位：

1hash(hot_key) = constant

所有 hot_key 行仍会进入同一个子分区。增加分区数量只能拆开不同 Hash 的 Key，不能拆开同一个 Key。

因此递归分区面对倾斜时可能出现：

1level 0: partition 3 too large
2  -> level 1: partition 3.7 still too large
3      -> level 2: partition 3.7.2 still too large
4          -> reach max depth / fail

可选的工程策略包括：

• 对热点 Key 使用单独路径；
• 在允许时选择另一侧 Build；
• 改用 Sort Merge Join；
• 对特定 Join 语义给数据加 Salt，并复制另一侧热点行；
• 对 Semi/Anti Join 只保留 Key 是否存在，消除重复 Payload；
• 通过统计与采样提前识别热点，而不是等 Spill 多层后才发现。

Salting 并不是免费方案。若给 Build 热点 Key 拆成多个 Salt，Probe 侧可能需要复制或使用相同拆分规则；对 Outer Join 还要避免重复输出未匹配行。它是一个新的 Join 算法设计，而不是简单修改 Hash Seed。

评估与调优：连接数据结构、内存和执行证据

如何评估一套 Hash Join 设计

数据结构、内存与 Spill 协议需要共同满足语义要求；只测内存快路径无法验证超限后的恢复与进展。

数据结构

键与负载表示决定查找和复制成本，评估时应区分整数特化、组合键和变长键。

• 是否为单整数、定长组合键和变长键选择不同表示？
• Bucket 中存完整行、Row ID，还是指针？
• 重复 Key 是否会产生大量小对象和随机分配？
• Outer Join 的命中标记按 Key 还是按 Row 保存？

内存模型

峰值内存包括索引、原始负载与临时状态；遗漏 Arena、重复链或恢复缓冲会低估资源需求。

• 内存估算是否包含 Bucket、Payload、Arena、重复链和碎片？
• 扩容峰值是否同时持有旧表与新表？
• 哪些内存可撤销，撤销需要多长时间？
• 当前分区恢复时是否为 Pointer Table 和 Probe Buffer 留出空间？

Spill 协议

分区一致性与恢复后的语义必须同时成立；Build 与 Probe 只要使用了不同映射，就可能漏掉匹配。

• Build/Probe 是否使用完全一致的 Hash 与 Partition Metadata？
• Spill 文件保存逻辑数据还是不可恢复的地址？
• 是否保留内存分区以减少 I/O？
• 超大分区如何递归处理，最大深度和失败信息是什么？
• 查询取消时，异步 I/O、临时文件和内存状态能否及时回收？

Pipeline 与可观测性

Build Barrier 与恢复轮次改变可运行任务数量，调度和指标需要显式区分快路径、等待和恢复。

• Build Barrier 前后如何保持线程忙碌？
• I/O、解压、重建 Hash Table 与 Probe 能否重叠？
• 是否能看到每层分区大小、倾斜度、Spill/Restore Bytes、重分区次数和 Build/Probe 耗时？
• Profile 能否区分“磁盘慢”“哈希表重建慢”和“输出爆炸”？

Benchmark 必须覆盖状态转换，而不只是内存快路径

只用均匀随机 Key、内存充足、Inner Join 和 count(*) 测试，会系统性高估一套 Hash Join 的成熟度。更有区分度的矩阵应覆盖：

每组实验都应同时报告吞吐、P50/P99、峰值内存、Hash Table Expansion、Spill/Restore Bytes、临时文件数、最大分区、递归深度、输出行数和取消清理时间。否则一个“更快”的实现可能只是多占内存、没有进入 Spill，或者把工作推迟到下游输出。

跨引擎比较还要固定语义与物理前提：相同 Join 顺序、Build Side、并行度、内存上限、输入编码、压缩方式和热/冷缓存状态。比较默认配置更接近产品体验；比较算法则必须控制这些变量，两者不能混为同一结论。

排障与调优：应该看什么

Hash Join 慢或 OOM 时，不应只看总 Spill Bytes。建议按以下顺序分析：

1. 确认 Join 选择与 Build 方向。 统计是否把大表误判成小表？Outer Join 是否限制了可选方向？
2. 看 Build Row、Distinct Key 与值域。 行数大但 Key 很少，重复链可能比 Bucket 更贵；小整数稠密时应关注直接映射机会。
3. 看最大分区而不是平均分区。 平均 200 MiB 不代表最大分区不会达到 8 GiB。
4. 看 Spill Level 与 Repartition 次数。 层级增长通常比总写盘量更能说明倾斜。
5. 看 Probe 输入与 Join 输出。 Probe 行多但输出少，应强化 Runtime Filter；输出远大于输入，则是多对多放大。
6. 看恢复阶段的内存峰值。 Tuple 数据装得下，不代表 Pointer Table、解压 Buffer 和输出 Chunk 同时装得下。
7. 看 CPU 与 I/O 是否重叠。 磁盘未打满且 CPU 也空闲，往往是状态机串行化或分区过碎。

调优动作应该对应根因：

进一步思考：好的 Hash Join 是资源自适应算子

从四套源码可以看到，Hash Join 的演进方向并不是不断优化某个 Hash 函数，而是把更多不确定性纳入运行时控制：

• Key 分布决定数据结构；
• Distinct Count 决定 Set、Map 与重复链；
• 值域决定 Direct/Perfect Hash；
• 实际内存决定是否 Spill、分多少区、一次恢复几个分区；
• 倾斜决定递归是否有效、并行度是否应该降低；
• Join 语义决定哪些行可过滤、哪些匹配状态必须跨分区保存。

这也给优化器与执行器的边界提出了新要求。优化器不应只下发一个静态的 HASH_JOIN 节点，还应传递：

• Build 侧大小、NDV、Min/Max 与热点 Key 估计；
• Join Key 是否唯一；
• Runtime Filter 的候选类型；
• Spill 是否允许，以及资源预算；
• 可接受的退化算法与失败边界。

执行器应反馈实际分区大小、重分区层级、峰值内存和命中率，以便区分估计偏差与算法边界。这是接口设计建议；能否改善后续选择，还需验证反馈的适用范围与失效条件。

验证与失败边界

验证应覆盖阈值两侧、重复键与 NULL、Spill 恢复和清理失败，并记录最大分区与并发内存。单个热键无法被同一哈希规则继续拆散；支持 Spill 不等于任意数据都能完成，也不能解决超大结果集本身的输出成本。

附录：Spark Hash Join 特有路径

本附录只保留 Spark 相对主文四引擎的差异：JoinSelection 决策树、HashedRelation、AQE 动态切换，以及内置 Hash Join 不可 Spill 的失败边界。通用哈希表/分区恢复协议见正文。

选择前提：语义、分发与资源估计

背景：为什么 Spark 没有默认选择 Hash Join

在 StarRocks、ClickHouse 等分析型数据库中，Hash Join 通常是等值连接的核心实现：选择较小一侧构建哈希表（Hash Table），另一侧流式 Probe，在哈希分布近似均匀且内存充足时，Build/Probe 主路径接近 O(N + M)；若显式输出 Z 条匹配结果，总工作量还要计入 O(Z)。

Spark SQL 的默认选择却不同。小表满足广播条件时，Spark 会优先使用 BroadcastHashJoin；一旦两侧都不能广播，常见执行计划往往变成 SortMergeJoin，而不是 ShuffledHashJoin。

差异需要从内存与恢复路径解释：本文快照中的 HashedRelation 缺少可恢复的 Join Spill，而 SMJ 的排序和同 Key 缓冲有落盘路径。AQE 可以修正统计与策略，不能改变算子的内存模型，因此比较还需计入数据倾斜（Data Skew）和失败边界。

本文基于 Apache Spark 提交 786bb3d 分析以下问题：

• Spark 原生支持哪些 Join 算子？
• JoinSelection 如何在 BHJ、SHJ、SMJ 之间做静态选择？
• Broadcast 与 Shuffle Hash Join 的物理执行路径有什么区别？
• HashedRelation 为什么不能像 Sort 一样稳定 Spill？
• AQE 如何在运行时升级或降级 Join？
• Native Engine 为什么更倾向实现可 Spill 的 Partitioned Hash Join？

这里的配置默认值和 Rule 顺序只对应这一源码快照。Spark 官方当前的配置文档仍列出 spark.sql.shuffledHashJoinFactor=3，但具体发行版、Vendor Runtime 和 Native Plugin 可能改写 JoinSelection 或替换物理算子。排查生产计划时，应同时记录 Spark Build Version、Session Conf、Initial/Final Plan 和插件版本，不能仅凭文章中的默认值判断。

选择顺序与估计边界

BHJ、SHJ 与 SMJ 的选择依赖语义、分发和内存条件。广播要求全局 Build 较小，SHJ 则要求每个 Task 的局部 Build 能驻留；不能广播且无法可靠估计内存时，可外排的 SMJ 提供另一条路径。这里的大小判断通常来自统计估计，并不是对实际峰值内存的证明。下面依次核对准入条件、执行结构与 AQE 的调整范围。

Spark Join 的算子坐标

Spark SQL 的常见物理 Join 可以分为以下几类：

对于等值 Join，BHJ、SHJ 和 SMJ 并不是三个完全独立的实现：

• BHJ 与 SHJ 共享 HashJoin 中的 Probe、Join Type 和 WholeStage Codegen 逻辑；
• BHJ 与 SHJ 都使用 HashedRelation；
• SHJ 与 SMJ 都继承 ShuffledJoin，要求两侧满足 Join Key 的聚簇分布。

这种分层很重要：选择 BHJ 还是 SHJ，首先改变的是数据如何到达 Build/Probe 算子；选择 Hash 还是 Merge，才真正改变分区内部的 Join 算法。

静态规划：JoinSelection 的决策树

物理选择入口位于 SparkStrategies.JoinSelection。公共判断逻辑位于 JoinSelectionHelper。

先区分等值与非等值 Join

ExtractEquiJoinKeys 把 Join 条件拆成：

• 左右等值 Key；
• 不能放入 Key 的剩余条件 nonEquiCond；
• Join Type、左右子计划与 Hint。

BHJ、SHJ 和 SMJ 都依赖等值 Key。剩余条件可以在匹配后继续过滤，但如果完全没有等值 Key，Hash Join 和 SortMergeJoin 都不能直接使用，规划器会转向 Cartesian Product 或 Broadcast Nested Loop Join。

1Join condition
2   │
3   ├─ 有 equi keys ──► BHJ / SHJ / SMJ
4   │                    └─ residual condition 在匹配后过滤
5   │
6   └─ 无 equi keys ──► Cartesian / Broadcast Nested Loop

Hint 的优先级

当用户提供 Hint 时，JoinSelection 按以下顺序尝试：

1. BROADCAST；
2. SHUFFLE_MERGE；
3. SHUFFLE_HASH；
4. SHUFFLE_REPLICATE_NL。

Hint 不是无条件命令。比如 Full Outer Join 不能规划为 BHJ，Hash Key 不满足二进制相等语义时也不能强制使用 Hash Join。无法应用的 Hint 会报告告警，并继续尝试其他策略或无 Hint 路径。

无 Hint 的默认顺序

createJoinWithoutHint() 的源码顺序非常直接：

1createBroadcastHashJoin(false)
2  .orElse(createShuffleHashJoin(false))
3  .orElse(createSortMergeJoin())
4  .orElse(createCartesianProduct())
5  .getOrElse(createBroadcastNestedLoopJoin())

对应准入条件如下：

所以，“Spark 默认选择 SMJ”并不完全准确。更精确的说法是：

Spark 首先尝试 BHJ；对不能广播的大表等值 Join，SHJ 的准入条件默认关闭，于是 SMJ 成为最常见落点。

Hash Join 的三个准入条件

哈希连接（Hash Join）必须同时满足键相等语义、合法 Build 方向和资源判断；Hint 不能让语义不支持的 Join 变得合法。

Key 必须支持二进制稳定相等

当前源码的 hashJoinSupported() 要求所有 Join Key 都满足：

1UnsafeRowUtils.isBinaryStable(e.dataType)

这意味着 UnsafeRow 的二进制相等必须与 SQL 语义相等一致。当前实现重点排除包含非 Binary Collation String 的类型：两个字符串可能在大小写不敏感或重音不敏感规则下语义相等，但底层字节不同。

这里不能简单理解为“所有可 Hash 的 JVM 类型都能 Join”。Spark 要求 Hash、Equality 和 SQL Collation 语义闭合；不满足时会记录 Warning，并放弃 BHJ/SHJ。

Join Type 决定哪一侧可以 Build

Build 侧不是永远选右表。外连接（Outer Join）必须保留特定一侧的未匹配行，因此不同 Join Type 的合法 Build 方向不同。

当两侧都合法时，getBuildSide() 根据 stats.sizeInBytes 选择更小的一侧。统计不准确不仅影响是否广播，也会影响哪一侧承担 Hash Table 内存。

大小判断不是一个阈值

BHJ 只需要满足 canBroadcastBySize()：

10 <= sizeInBytes <= autoBroadcastJoinThreshold

当前源码默认 spark.sql.autoBroadcastJoinThreshold=10MB。

SHJ 静态选择则需要同时满足：

1preferSortMergeJoin == false
2
3BuildSize < autoBroadcastJoinThreshold × numShufflePartitions
4
5BuildSize × shuffledHashJoinFactor <= StreamSize

当前基线的 spark.sql.shuffledHashJoinFactor 默认值是 3。第二个条件用全局大小近似“平均每个 Shuffle 分区能装入与广播阈值相近的 Hash Table”；第三个条件要求 Build 侧明显小于 Stream 侧，以抵消构建 Hash Table 的成本。

这个估算有明显局限：平均值无法发现倾斜。总数据量满足条件，不代表最大的 Build 分区一定能放进单个 Task 的内存。

Hash 执行：两种分发方式与同一个内存核心

BroadcastHashJoin：用全局小表换掉 Shuffle

BroadcastHashJoinExec 的核心价值不是 Hash Probe 本身，而是避免 Stream 侧 Shuffle。

 1                  Driver
 2                    │
 3        collect Build side rows
 4                    │
 5             build HashedRelation
 6                    │
 7              Torrent Broadcast
 8          ┌─────────┼─────────┐
 9          ▼         ▼         ▼
10      Executor 1 Executor 2 Executor N
11          │         │         │
12       probe local stream partitions

Build 侧在哪里构建

BroadcastExchangeExec.relationFuture 执行以下步骤：

1. child.executeCollectIterator() 把 Build 侧收集到 Driver；
2. HashedRelationBroadcastMode.transform() 构建 HashedRelation；
3. 检查行数上限和构建后字节数上限；
4. 通过 Spark Broadcast 发送序列化 Relation；
5. Executor Task 调用 asReadOnlyCopy()，以只读方式 Probe。

源码位于 BroadcastExchangeExec。

BHJ 的内存风险因此横跨多个层次：

• Driver 需要收集并构建整个 Relation；
• Broadcast Block 需要网络传输和缓存；
• Executor 需要反序列化并持有只读 Hash Table；
• 多个并发 Broadcast 会竞争 Driver 与 Executor 内存。

当前源码还设置了执行期保护：

• spark.sql.broadcastTimeout 默认 300 秒；
• 普通 UnsafeHashedRelation 受 BytesToBytesMap 最大 Key 数约束；
• spark.sql.maxBroadcastTableSize 当前默认 8 GiB。

自动广播阈值与执行期上限不是一回事。Hint 可以绕过 10 MiB 的自动选择门槛，但不能让 Driver 内存和 Hash Table 容量变成无限。

Probe 与 WholeStage Codegen

非 Null-Aware Anti Join 路径的核心非常短：

1val broadcastRelation = buildPlan.executeBroadcast[HashedRelation]()
2
3streamedPlan.execute().mapPartitions { streamedIter =>
4  val hashed = broadcastRelation.value.asReadOnlyCopy()
5  join(streamedIter, hashed, numOutputRows)
6}

HashJoin 根据 Join Type 生成不同路径：

• Inner Join：查到一个或多个 Build Row 后输出；
• Outer Join：没有命中时补 Null Row；
• Semi Join：首次满足条件后即可返回 Stream Row；
• Anti Join：确认无匹配后返回；
• Existence Join：输出是否存在匹配的标记。

若 keyIsUnique=true，代码生成可以调用 getValue()，避免为重复 Key 创建 Iterator；非唯一 Key 则调用 get() 遍历同 Key Value 链。唯一性因此不仅影响语义，还会改变生成代码的控制流和对象分配。

ShuffledHashJoin：每个分区独立 Build

ShuffledHashJoinExec 要求两侧按 Join Key 形成相同的 ClusteredDistribution。EnsureRequirements 会在需要时插入 Shuffle Exchange。

1Left  ──Hash Shuffle──┐
2                     ├─ Partition 0: build local hash + stream probe
3Right ──Hash Shuffle──┘
4
5                     ├─ Partition 1: build local hash + stream probe
6                     └─ Partition N: build local hash + stream probe

一次 Task 的执行模型

doExecute() 使用 zipPartitions 配对两侧相同分区：

1streamedPlan.execute().zipPartitions(buildPlan.execute()) {
2  (streamIter, buildIter) =>
3    val hashed = buildHashedRelation(buildIter)
4    join(streamIter, hashed, numOutputRows)
5}

这个流程存在明确的不对称：

• Build 分区必须被完整消费并形成 Hash Table；
• Stream 分区可以惰性读取，一行一行 Probe；
• Relation 在 Task Completion Listener 中关闭；
• 峰值内存由最大的 Build 分区决定，而不是平均分区大小决定。

所以 SHJ 的真正准入问题不是“总表是否足够小”，而是：

对每一个 Shuffle 分区，Build Rows、Hash Index、重复 Key 链和外连接匹配标记能否同时装入该 Task 可获得的 Execution Memory？

Outer Join 为什么更复杂

当 Full Outer Join，或者外侧恰好被选为 Build 侧时，只流式 Probe 无法输出未匹配的 Build Row。buildSideOrFullOuterJoin() 使用两阶段算法：

1. Probe Stream 侧，并记录哪些 Build Row 已匹配；
2. 遍历整个 HashedRelation，补出未匹配 Build Row。

Key 唯一时用 BitSet 记录 keyIndex；Key 不唯一时用 OpenHashSet[Long] 编码 (keyIndex, valueIndex)。这会增加额外内存，也会使这类 Join 无法保持普通 Stream 侧输出顺序。

Semi/Anti Join 的重复 Key 优化

对 Left Semi、Left Anti 等存在性语义，如果剩余条件不依赖 Build 侧非 Key 列，Build Hash Table 每个 Key 只需保留一行。

ignoreDuplicatedKey 会让 HashedRelation 在 Key 已存在时跳过后续 Value：

1普通 Inner Join：key -> row1 -> row2 -> row3
2存在性 Join：    key -> one row

这一优化可以显著减少高重复维表的内存，但不能解决高基数 Build 分区无法装入内存的问题。

HashedRelation：Hash Join 的内存核心

HashedRelation.apply() 根据 Key 形态选择两种实现：

UnsafeHashedRelation

UnsafeHashedRelation 使用 BytesToBytesMap：

• Append-only 开放寻址 Hash Table；
• Key/Value 以连续二进制记录写入 Memory Page；
• Hash Array 保存记录地址与 Hash Code；
• 同 Key 多 Value 通过地址链连接；
• 负载因子达到约 70% 时尝试扩容并 Rehash；
• 内存来自 TaskMemoryManager，可使用 On-Heap 或 Off-Heap Tungsten Memory。

逻辑布局可以简化为：

1Hash Array
2┌──────────┬──────────┬──────────┐
3│ addr/hash│ addr/hash│ addr/hash│
4└────┬─────┴──────────┴────┬─────┘
5     │                      │
6     ▼                      ▼
7Memory Pages
8[key length][value length][key bytes][value bytes][next value address]

UnsafeHashedRelation.apply() 持续读取 Build Iterator，生成 Join Key 并调用 loc.append()。一旦 append 返回 false，代码释放 Map 并抛出 cannotAcquireMemoryToBuildUnsafeHashedRelationError。

LongHashedRelation

单 Long Key 可以走更专门的 LongToUnsafeRowMap：

• 初始使用 Sparse Open Addressing；
• Build 完成后 optimize() 尝试切换 Dense Mode；
• Dense Mode 通过 key - minKey 直接定位地址；
• 所有 UnsafeRow 字节紧凑放在 Page 中。

当 Key 范围紧凑时，Dense Mode 能减少 Hash 计算与冲突探测；当 Key 分布稀疏时则保留 Sparse Mode，避免按巨大数值范围分配数组。

Spill 真相：Spark Hash Join 为什么会直接失败

底层容器具有 spill() 接口不等于 Join 能落盘恢复；本文版本的 BytesToBytesMap 只有进入特定破坏性迭代路径才执行相应 spill，常规 HashedRelation 查找不具备这套协议。

BytesToBytesMap 的 spill() 为什么帮不到 Join

BytesToBytesMap 是 MemoryConsumer，看上去实现了 spill()：

1public long spill(long size, MemoryConsumer trigger) throws IOException {
2  if (trigger != this && destructiveIterator != null) {
3    return destructiveIterator.spill(size);
4  }
5  return 0L;
6}

关键是 destructiveIterator。它用于 UnsafeKVExternalSorter 接管 Map 内容的聚合/排序路径；Hash Join Build 不会建立这个迭代器。因此 Join 使用的 Map 被要求释放内存时，通常只能返回 0L。

LongToUnsafeRowMap.spill() 更直接：

1def spill(size: Long, trigger: MemoryConsumer): Long = 0L

因此，Spark Hash Join 的内存行为是：

1申请 Hash Array / Memory Page
2          │
3          ├─ 成功：继续 Build
4          │
5          └─ 失败：无法把已有 Hash Partition 落盘
6                     │
7                     └─ append=false -> 释放 Relation -> Task 失败

统一内存管理不等于所有算子都可 Spill

TaskMemoryManager 可以在 Execution Memory 紧张时要求其他 MemoryConsumer 释放内存。Sort、Aggregation 或 Shuffle Buffer 可能把数据写盘，为 Hash Table 腾出空间；但 Hash Table 自己没有 Partitioned Spill 状态机。

这会形成一个不对称结果：

• Hash Join 可以从其他可 Spill 算子释放的内存中受益；
• 它自己无法有效响应内存回收；
• 如果同一 Task 中没有其他可驱逐对象，或者 Build 分区仍超过可用内存，Task 最终失败。

所以把 spark.memory.fraction 调大只能扩大失败边界，不能把 SHJ 变成 Grace Hash Join。

倾斜是 SHJ 最危险的输入

假设 Build 侧总大小 20 GiB、200 个分区，平均每个分区约 100 MiB，看起来可以接受；但如果一个热点 Key 形成 5 GiB 分区，该 Task 仍需一次性构建 5 GiB 以上的 Relation。

Hash Table 实际内存还高于原始输入：

1Build 原始 UnsafeRow
2  + Join Key 副本
3  + Hash Array 空槽
4  + 地址与长度元数据
5  + 重复 Key 链指针
6  + Rehash 峰值
7  + Outer Join 匹配标记

因此不能用 Shuffle 文件压缩后的字节数直接等价 Hash Table 峰值。

Sort-Merge 对照（不展开）

大表默认路径仍是 SortMergeJoin：外排可完成、内存上限更清晰，但 merge 并非 O(1) 内存，倾斜与重复键仍可变慢。通用 Hash vs Sort 决策见 Hash 还是 Sort；此处只强调 Spark 内置 Hash Join 缺省不可 Spill，大表不能假设“选了 Hash 就与主文四引擎一样可恢复”。

运行时调整：AQE、倾斜与 Native Engine

AQE：运行时重新选择 Join

AQE 不是在执行到一半时把 SortMergeJoinExec 对象直接改成 BroadcastHashJoinExec。其核心流程是：

 1Query Stage 物化
 2      │
 3      ▼
 4获得真实 size / mapStats
 5      │
 6      ▼
 7把 Stage 包装回 LogicalQueryStage
 8      │
 9      ▼
10invalidateStatsCache + AQEOptimizer
11      │
12      ▼
13更新 Runtime Stats 或注入 Join Hint
14      │
15      ▼
16重新执行 planner.plan()
17      │
18      ▼
19再次进入 JoinSelection

入口位于 AdaptiveSparkPlanExec.reOptimize()。

Runtime Stats：SMJ/SHJ 升级为 BHJ

物化后的 QueryStageExec 产生 isRuntime=true 的统计。canBroadcastBySize() 检测到运行时统计后，会使用：

1spark.sql.adaptive.autoBroadcastJoinThreshold

未配置时回退到普通自动广播阈值。若某侧真实大小远小于静态估算，重新规划就可能从 SMJ/SHJ 变为 BHJ。

需要注意：如果 Shuffle Stage 已经物化，之前的 Shuffle 成本不会神奇消失。AQE 能避免后续 Sort/Merge，并可能用 Local Shuffle Read 读取已有 Block，但无法回收已经发生的网络和写盘成本。

DynamicJoinSelection：基于分区统计注入 Hint

DynamicJoinSelection 读取 ShuffleQueryStageExec.mapStats.bytesByPartitionId，有两类核心判断。

避免不划算的 Broadcast

如果某侧包含大量空分区，Shuffle Join 的许多 Task 可以快速短路；把另一侧广播后，反而可能失去这个优势。规则会根据 Join Type 和哪一侧为空，注入 NO_BROADCAST_HASH。

Outer/Anti Join 需要保留未匹配行，不能简单在任意空侧短路，因此源码对 LeftOuter、RightOuter、LeftAnti 做了额外区分。

SMJ 升级为 SHJ

如果所有分区都满足：

1partitionSize <= maxShuffledHashJoinLocalMapThreshold

并且 advisoryPartitionSizeInBytes 也不大于该阈值，规则会注入 PREFER_SHUFFLE_HASH。若同时需要禁用 Broadcast，则注入更强的 SHUFFLE_HASH。

这个条件的本质是：AQE 已经看到每个 Build 分区的真实字节数，可以比静态“总大小除以分区数”更可靠地判断局部 Hash Table 风险。

但当前基线中：

1spark.sql.adaptive.maxShuffledHashJoinLocalMapThreshold = 0

因此，除非用户显式配置，AQE 通常不会主动把 SMJ 升级为 SHJ。这再次说明 Spark 对无 Spill Hash Join 的默认态度非常保守。

用户 Hint 优先于 AQE Hint

DynamicJoinSelection 只在对应一侧没有用户 Join Strategy Hint 时写入 AQE Hint。优先级由此保持一致：

1用户显式 Hint
2    > AQE 根据真实统计注入的 Hint
3    > 静态自动选择

已经广播的 Stage 不允许回退

LogicalQueryStageStrategy 排在普通策略之前。

如果 Join 某一侧已经是 BroadcastQueryStageExec，规则会强制保留 BHJ，防止因 Stage 完成顺序或统计波动，把已经构建和广播的数据重新改成 Shuffle Join。

这是一种 Sunk Cost 保护：物理 Stage 一旦完成，重新规划必须尊重已经付出的执行成本。

Local Shuffle Read 的准确位置

OptimizeShuffleWithLocalRead 常被误解为“BHJ 降级为 Shuffle Join 后使用 Local Read”。源码实际处理的是：

• AQE 重规划得到 BHJ；
• Probe 侧已经存在物化的 Shuffle Stage；
• 在不引入额外 Shuffle 的前提下，改用本地 Shuffle Block 读取。

它服务于 Shuffle Join → BHJ 的升级复用，而不是给 BHJ → Shuffle Join 的降级提供本地读。

倾斜优化能否解决 SHJ OOM

OptimizeSkewedJoin 可以识别显著大于中位数和绝对阈值的分区，将一侧切成多个 Mapper Range，并复制另一侧对应分区。

1原始：        (L-skew, R)
2
3拆分后：      (L-1, R-copy-1)
4              (L-2, R-copy-2)
5              (L-3, R-copy-3)

它能降低单 Task 输入并增加并行度（Degree of Parallelism），但不能从根本上赋予 HashedRelation Spill 能力：

• Build 分区拆分后仍需完整驻留；
• 若倾斜来自单个超高频 Key，另一侧复制可能增加总 I/O；
• 两侧同时倾斜时，Split 组合可能形成更多 Task；
• Hash Table 放大系数仍未被 Map Output Bytes 精确表达。

因此，Skew Join Optimization 是风险缓解，不是 SHJ 内存安全证明。

Native Engine：为什么 Hash Join 更容易成为默认方案

Velox 等执行引擎把向量化 Probe、显式内存管理与 Partitioned Spill 结合，改变了 Hash Join 的可用内存边界；“Native”这一实现语言标签本身不保证收益。Velox 的职责拆分为 HashBuild、HashProbe 与 HashJoinBridge：

 1Build vectors -> HashBuild -> in-memory HashTable
 2                      │
 3                      ├─ memory arbitration / reclaim
 4                      │
 5                      └─ spill selected hash partitions
 6                                  │
 7Probe vectors -> HashProbe --------┤
 8                      │            │
 9                      └─ spill matching probe partitions
10                                  │
11                          restore partition and re-join

公开源码中的 HashBuild 和 HashProbe 都包含 Spill、Memory Reclaim 和递归恢复状态。

这与 Spark JVM HashedRelation 的关键区别不是 C++ 一定比 JVM 快，而是算法状态机不同：

• Spark SHJ：一个 Build 分区对应一张必须完整驻留的 Hash Table；
• Partitioned Hash Join：先按 Hash Bit 切分，内存不足时只保留部分 Partition，Build/Probe 对称落盘，随后逐 Partition 恢复。

Native Engine 还可以利用批量 Probe、SIMD、Dictionary Vector 和更精细的 Memory Arbitration 降低 CPU 与对象开销。但 Spill 一样会引入磁盘 I/O、递归分区、文件数量和长尾问题，并不是免费的性能保险。

对于 Databricks Photon、Snowflake 等闭源引擎，公开资料可以说明它们采用 Native/Vectorized Execution，但无法像本文分析 Spark 和 Velox 一样验证具体 Hash Table、Partition 数、Spill 触发和恢复协议。因此不应把产品层性能描述当成可复现的源码结论。

验证时看什么（精炼）

• Explain / AQE：是否 BHJ↔SHJ/SMJ 切换、Broadcast 是否被拒绝回退、Local Shuffle Read 位置。
• Task 指标：peak execution memory、spill（Hash Join 上通常应接近 0；若出现 OOM 看分区倾斜而非“再加大内存就算 Spill”）。
• 适用边界：用户 Hint > AQE Hint；已广播 Stage 不回退；热键无法靠同一 hash 再拆。

Spark 源码阅读坐标

1. JoinSelection：等值/非等值、Hint、BHJ/SHJ/SMJ 准入。
2. HashedRelation / UnsafeHashedRelation / LongHashedRelation：内存核心。
3. BroadcastHashJoinExec / ShuffledHashJoinExec：Build/Probe 与 codegen。
4. AQE：DemoteBroadcastJoin、DynamicJoinSelection、runtime stats 升级路径。

源码阅读路线

如果希望继续深入，建议按以下顺序阅读。

StarRocks

1. join_hash_table.cpp：Key Constructor 与 Hash Map Method 选择；
2. hash_joiner.cpp：Build/Probe 状态封装；
3. spillable_hash_join_build_operator.cpp：内存撤销与 Build 分区；
4. spillable_hash_join_probe_operator.cpp：分区选择、恢复与 Probe Spill。

Apache Doris

1. join_hash_table.h：Join Hash Table 结构；
2. hashjoin_build_sink.cpp：普通 Build 与 Runtime Filter；
3. hashjoin_probe_operator.cpp：Join Type 模板与向量化 Probe；
4. partitioned_hash_join_probe_operator.cpp：恢复队列与递归重分区。

ClickHouse

1. HashJoin.cpp：Key Method、Build 与 Probe；
2. ConcurrentHashJoin.h：Two-Level Map 并行构建；
3. GraceHashJoin.h：外存 Join 的三阶段模型；
4. GraceHashJoin.cpp：Bucket 文件、动态 Rehash 与延迟 Probe。

DuckDB

1. join_hashtable.hpp：Tuple Layout、Probe State 与 Spill State；
2. join_hashtable.cpp：Pointer Table、Salt、重复链与外存分区；
3. physical_hash_join.cpp：Pipeline、Finalize Task 与临时内存预留；
4. perfect_hash_join_executor.cpp：稠密整数值域的直接寻址。