本文作者：徐飞
导读
在数据库管理系统（DBMS）中，连接操作（Join）是查询处理的核心环节之一，其性能直接影响到整个系统的响应速度和效率。随着数据量的不断增长和复杂查询需求的增加，优化连接操作的执行效率成为提升数据库性能的关键。Hash Join 作为一种高效的连接算法，因其在处理大规模数据集时的卓越性能而被广泛应用于现代数据库系统中。
本文深入介绍了 TiDB 中新 Hash Join 的设计与实现，对比了新旧 Hash Join 的性能表现。通过引入并发构建机制、优化的哈希表设计以及针对溢出（Spill）操作的全新算法，新 Hash Join 在多个方面实现了显著的性能提升。文章详细探讨了 Build 阶段 和 Probe 阶段 的优化策略，并通过实际测试数据展示了新 Hash Join 在处理大规模数据集时的高效表现。
1. 背景
JOIN 是数据库中极为重要的操作之一，其核心功能是将两个表的数据按照指定条件进行合并。例如，以下 SQL 语句：
SELECT * FROM a JOIN b ON a.id = b.id AND a.value >= b.value;
展示了如何通过 JOIN 将表 a 和表 b 的数据进行连接。接下来，我们先简要介绍 JOIN 的一些基本概念。
在上述 SQL 语句中，表 a 和表 b 是参与 JOIN 操作的表，而 ON 后面的条件（ a.id = b.id AND a.value >= b.value ）则是 JOIN 条件。下图展示了表 a 和表 b 的数据，以及通过 JOIN 操作后的结果：
这种仅将满足 JOIN 条件的 {a_row, b_row} 组合作为结果的连接方式被称为 内连接（Inner Join） 。除了 Inner Join 之外，还有其他几种常见的连接类型，例如 外连接（Outer Join） 、**半连接（Semi Join）**和 反半连接（Anti Semi Join） 等。
最基础的 JOIN 实现方式是 嵌套循环连接（Nested Loop Join ）。这种实现方式严格遵循 JOIN 的语义定义，没有引入额外的优化手段。以 Inner Join 为例，Nested Loop Join 的具体实现逻辑如下：
for a_row in a
for b_row in b
if condition(row_a, row_b) then
emit_result(a_row, b_row)
end if
假设表 a 的数据量为 m ，表 b 的数据量为 n ，则 Nested Loop Join 的算法复杂度为 O(m*n) 。这种复杂度显然较高，当表 a 和表 b 的数据量较大时，Nested Loop Join 的执行速度会显著变慢。
由于 Nested Loop Join 的性能较差，大多数数据库管理系统都提供了其他更高效的连接算法。其中， Hash Join 是最常见的一种高效连接算法。
在 Hash Join 中， JOIN 条件通常被分为两大类：
- 等值连接条件 ：连接条件的左右两边分别来自两个表，且为等值比较。例如，在上述例子中，
a.id = b.id即为等值连接条件。 - 非等值连接条件 ：除了等值连接条件之外的其他条件均属于非等值连接条件。例如，在上述例子中，
a.value >= b.value即为非等值连接条件。
Hash Join 的核心思想是利用哈希表来高效判断等值连接条件。因此，若要使用 Hash Join，连接条件中至少需要包含一个等值连接条件。
继续以上述 SQL 示例和数据为例，Hash Join 的执行过程示意如下：
在 Hash Join 中，表 b 的数据被构建为一个 Hash Table，其中 键（Key） 是等值连接条件中表 b 的数据（即 b.id ，以下称为 连接键（Join Key） ），而 值（Value） 是表 b 的原始数据。
当表 a 进行 Hash Join 时，会使用表 a 的 Join Key（即 a.id ）在 Hash Table 中查找相关的表 b 数据。Hash Table 查找的结果已经满足了等值连接条件 a.id = b.id 。对于没有非等值连接条件的连接操作，Hash Table 查找的结果即为连接的最终结果。而对于包含非等值连接条件的连接操作，可以在 Hash Table 查找结果的基础上，进一步对非等值连接条件进行求值。
在具体实现中，Hash Join 会从参与连接的两个表中选择一个表作为 构建端（Build 端） ，另一个表作为 探测端（Probe 端） 。Hash Join 的执行分为两个阶段：**Build 阶段（构建阶段）**和 Probe 阶段（探测阶段） 。
- Build 阶段 ：Build 阶段的主要任务是接收 Build 端的数据，并构造相应的 Hash Table。由于 Hash Table 需要保存在内存中，因此 Build 阶段的内存使用量与 Build 端的数据量成正比。通常情况下，选择数据量较小的表作为 Build 端是一个更优的选择。一方面，较小的数据量可以减少内存占用；另一方面，在 Probe 阶段，较小的 Hash Table 能够提供更高的查询性能。
- Probe 阶段 ：Probe 阶段的主要任务是使用 Probe 端的数据查询 Hash Table，并根据查询结果以及非等值连接条件的结果构造最终的连接结果。与 Build 阶段不同，Probe 阶段不会产生内存积累问题。Probe 端的每一行数据都可以独立地进行 Hash Table 查询，并生成连接结果。
从时间复杂度的角度来看，Hash Join 的性能表现如下：
Build 阶段 ：
Build 阶段的时间复杂度为 ***O(n)***，其中 n 是 Build 端表的大小。这一阶段主要涉及对 Build 端数据的扫描和 Hash Table 的构建。
**Probe 阶段： **
Probe 阶段的复杂度与数据分布密切相关。假设 Build 端每个 Join Key 平均有 k 行数据，Probe 端的 Join Key 在 Hash Table 中命中的概率为 x ，则 Probe 阶段的复杂度可以表示为 ***O(m*x*k + m*(1-x))***，其中 m 是 Probe 端表的大小。
因此，总体而言，Hash Join 的时间复杂度可以表示 ***O(m*x*k+m*(1−x)+n)***。
在大多数实际场景中， k 通常是一个较小的常数（例如 k≤10 ），因此 Hash Join 的复杂度大致可以简化为 O(m+n) 的量级。这意味着在常见情况下，Hash Join 的性能表现非常高效。
然而，在极端情况下，如果 Build 端和 Probe 端的所有 Join Key 都完全相同（即每个 Join Key 对应大量重复行），Hash Join 的时间复杂度可能会退化到 ***O(m*n)***。在这种情况下，Hash Join 不仅需要执行 Nested Loop Join 类似的操作，还需要额外进行 Hash Table 的构建和探测过程。因此，在这种极端情况下，Hash Join 的性能可能不如 Nested Loop Join。
2. TiDB 中的 hash join
TiDB 自首个版本（v1.0.0）起便支持 Hash Join。从功能角度来看，TiDB 的 Hash Join 已经具备了较为完善的支持：一方面，它支持 TiDB 中几乎所有的连接类型；另一方面，当内存使用量过高时，它还能自动触发 spill 操作。然而，从性能角度来看，TiDB 当前的 Hash Join 实现仍存在诸多已知问题，这也是我们需要在 TiDB 中引入新版本 Hash Join 的原因。
已知的性能问题
Build 阶段
目前，TiDB 的 Hash Join 在 Build 阶段存在以下两个主要问题：
1. 单线程构建问题 ：Hash Join 的 Build 阶段仅支持单线程操作。尽管 Hash Join 使用的 Hash Table 是并发安全的（concurrent map ( https://github.com/pingcap/tidb/blob/v8.1.0/pkg/executor/concurrent_map.go#L37 )），但在构建哈希表时，只使用了一个 Goroutine 来进行 hash table 的 build ( https://github.com/pingcap/tidb/blob/v8.1.0/pkg/executor/join.go#L1238 )。当构建表的规模较大时，单线程构建会显著拖慢整个连接操作的性能。
2. 哈希表性能瓶颈 ：Hash Join 中使用的哈希表是 Go 语言自带的 map 类型 ( https://github.com/pingcap/tidb/blob/v8.1.0/pkg/executor/concurrent_map.go#L31 )。虽然这种类型能够保证基本的性能下限，但对于 Hash Join 这种以哈希表为核心数据结构的算法来说，如果能够针对 Hash Join 的特点设计一个专用的哈希表，将能够实现比自带 map 类型更高的性能表现。
Probe 阶段
TiDB 的 Hash Join 在 Probe 阶段虽然支持多线程执行，但仍存在一些性能问题，主要体现在以下几个方面：
- Probe 接口设计不合理 ：
<!---->
- 哈希表查找与 Probe 分离，且不感知连接类型 ：在 TiDB 的 Hash Join 实现中哈希表以 Join Key 的哈希值为键，所以查找时需要对所有匹配的哈希表条目再次进行 Join Key 的比较。对于某些特殊连接类型（例如仅包含等值连接条件的 Semi Join），实际上并不需要查找哈希表中所有匹配的 Build 端数据，只需判断哈希表中是否存在匹配的 Build 端数据即可。然而，由于哈希表查找不感知连接类型，这种情况下仍需执行完整的哈希表查找。当 Build 端存在大量重复数据时，连接性能会受到显著影响 ( https://github.com/pingcap/tidb/issues/47424 )。
- Probe 接口以行为单位 ：目前 TiDB 的 Hash Join 以行为单位 ( https://github.com/pingcap/tidb/blob/v8.1.0/pkg/executor/joiner.go#L62 )进行 Probe 操作。如果连接操作中包含非等值连接条件，这些条件的求值很难实现完全的向量化执行，从而导致连接性能大幅下降。
<!---->
- 哈希表查找中 Join Key 比较存在冗余计算 ：TiDB 的哈希表以 Join Key 的哈希值作为键进行索引。在哈希表查找过程中，需要对所有匹配的哈希值再次进行 Join Key 的精确匹配。对于复杂的 Join Key（例如带有排序规则（collation）的字符串类型），比较操作需要重新计算字符串列的排序键。在当前的 TiDB Hash Join 实现中，每次 Join Key 比较都需要重新执行这些计算，从而导致大量冗余计算。
- Hash Join 生成结果的效率较低 ：Hash Join 生成结果的过程本质上是将
{build_row, probe_row}组合成一个新行，并将其插入到 Hash Join 的结果集中。目前，TiDB 的 Hash Join 在这一环节缺乏针对性的优化，导致生成结果的效率较低。
Spill
虽然 TiDB 的 Hash Join 支持在内存使用超出限制时进行 Spill 操作，但当前的 Spill 算法在功能和性能方面仍存在一些问题。
目前，Hash Join 的 Spill 操作完全封装在 hashRowContainer 中。如图所示， hashRowContainer 包含两部分内容：hash Table 以及 Build 端的数据。
当 Hash Join 需要触发 Spill 操作时， hashRowContainer 会将 Build 端的部分或全部数据（chunk 1 - chunk N）Spill 到磁盘，而 hash table 本身不会被 Spill。一个已经触发 Spill 操作的 hashRowContainer 如下图所示。
其中， chunk 2 至 chunk N 已被 Spill 到磁盘，而 chunk 1 仍然保留在内存中。
从功能角度来看，当前的 Spill 算法存在一个显著的缺点：hash table 本身无法被 Spill。然而，哈希表的规模会随着 Build 端数据量的增加而增大。这意味着即使已经触发了 Spill 操作，随着 Build 端数据量的进一步增加，Hash Join 的内存占用量依然会持续上升。
从性能角度来看，当前的 Spill 逻辑完全封装在 hashRowContainer 中，Probe 端无法感知是否发生了 Spill。例如，假设某一行探测数据的 Join Key 为 key1 ，在探测时，系统会将 {chunk1-row1} 、 {chunk2-row2} 、 {chunk2-rowN} 等所有相关数据读取到内存中。这表明，每探测一行数据，都可能引发一次或多次对磁盘的随机访问。众所周知，磁盘的随机访问速度较慢，因此一旦发生 Spill，Hash Join 的性能将显著下降。
其他问题
除了上述问题外，TiDB 的 Hash Join 还有一些其他功能上的缺失：
1. Semi Join 使用左表做 build ：对于 a SEMI JOIN b，TiDB 目前仅支持使用表 b 作为 Build 端。当表 a 较小而表 b 较大时，这种限制会导致连接操作的整体性能下降。
2. Null-Aware 左外半连接（Null-Aware Left Outer Semi Join）缺失：目前 TiDB 的 Hash Join 尚未支持 Null-Aware 左外半连接 。这使得此类连接操作只能退化为使用 Cross Join 来实现，而与 Hash Join 相比，Cross Join 的性能通常会下降数倍甚至数十倍。
性能问题的具体体现
为了直观地说明上述性能问题在实际查询中的具体体现，我们基于 TPC-H 50 数据集构造了一些简单的连接查询（join query）。
简单 join query 测试
以下是测试使用的 join query：
SELECT /*+ HASH_JOIN(lineitem) */ COUNT(*)
FROM orders
JOIN lineitem
ON l_orderkey = o_orderkey;
在测试中，orders 表包含 7500 万条数据 ，而 lineitem 表包含 3 亿条数据 。在 TiDB 中， orders 表被选为 Build 端，这已经是当前场景下的最优选择。
然而，即使在这种情况下，上述查询在 TiDB 中的执行时间仍然需要大约 65 秒 。
mysql> explain analyze select /*+ hash_join(lineitem) */ count(*) from orders join lineitem on l_orderkey = o_orderkey;
+-------------------------------+--------------+-----------+-----------+----------------+--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+--------------------------------------------------------------------------+----------+---------+
| id | estRows | actRows | task | access object | execution info | operator info | memory | disk |
+-------------------------------+--------------+-----------+-----------+----------------+--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+--------------------------------------------------------------------------+----------+---------+
| HashAgg_9 | 1.00 | 1 | root | | time:1m4.1s, loops:2, RU:299485.366752, partial_worker:{wall_time:1m4.057340034s, concurrency:5, task_num:292977, tot_wait:5m1.108226962s, tot_exec:19.104382394s, tot_time:5m20.286397525s, max:1m4.057285862s, p95:1m4.057285862s}, final_worker:{wall_time:1m4.057353412s, concurrency:5, task_num:9, tot_wait:10.089µs, tot_exec:1.818µs, tot_time:5m20.286492673s, max:1m4.057305079s, p95:1m4.057305079s} | funcs:count(1)->Column#26 | 196.6 KB | 0 Bytes |
| └─HashJoin_31 | 304437511.92 | 300005811 | root | | time:1m3.7s, loops:292978, build_hash_table:{total:29.5s, fetch:226.1ms, build:29.3s}, probe:{concurrency:5, total:5m20.3s, max:1m4.1s, probe:2m45.5s, fetch and wait:2m34.8s} | inner join, equal:[eq(test.orders.o_orderkey, test.lineitem.l_orderkey)] | 6.15 GB | 0 Bytes |
| ├─TableReader_33(Build) | 75000000.00 | 75000000 | root | | time:144.5ms, loops:73324, cop_task: {num: 1847, max: 0s, min: 0s, avg: 28.9ms, p95: 39.5ms, tot_proc: 51.3s, tot_wait: 67.9ms, copr_cache_hit_ratio: 0.01, build_task_duration: 43.2µs, max_distsql_concurrency: 15}, rpc_info:{Cop:{num_rpc:1847, total_time:53.3s}} | data:TableFullScan_32 | 6.14 MB | N/A |
| │ └─TableFullScan_32 | 75000000.00 | 75000000 | cop[tikv] | table:orders | tikv_task:{proc max:55ms, min:0s, avg: 27.3ms, p80:34ms, p95:36ms, iters:80639, tasks:1847}, scan_detail: {total_process_keys: 74946240, total_process_keys_size: 2023548480, total_keys: 74948071, get_snapshot_time: 37.3ms, rocksdb: {key_skipped_count: 74946240, block: {cache_hit_count: 26051, read_count: 356875, read_byte: 3.06 GB, read_time: 883.8ms}}}, time_detail: {total_process_time: 51.3s, total_suspend_time: 84.2ms, total_wait_time: 67.9ms, total_kv_read_wall_time: 50.4s, tikv_wall_time: 51.6s} | keep order:false | N/A | N/A |
| └─TableReader_35(Probe) | 300005811.00 | 300005811 | root | | time:830.3ms, loops:293432, cop_task: {num: 7790, max: 0s, min: 0s, avg: 32.6ms, p95: 58.2ms, tot_proc: 4m6.4s, tot_wait: 272.9ms, copr_cache_hit_ratio: 0.00, build_task_duration: 176.5µs, max_distsql_concurrency: 15}, rpc_info:{Cop:{num_rpc:7790, total_time:4m13.5s}} | data:TableFullScan_34 | 4.60 MB | N/A |
| └─TableFullScan_34 | 300005811.00 | 300005811 | cop[tikv] | table:lineitem | tikv_task:{proc max:0s, min:0s, avg: 30.9ms, p80:39.9ms, p95:56ms, iters:324135, tasks:7790}, scan_detail: {total_process_keys: 299971475, total_process_keys_size: 10798973100, total_keys: 299979250, get_snapshot_time: 154.2ms, rocksdb: {key_skipped_count: 299971475, block: {cache_hit_count: 167239, read_count: 1709356, read_byte: 13.4 GB, read_time: 4.13s}}}, time_detail: {total_process_time: 4m6.4s, total_suspend_time: 440.6ms, total_wait_time: 272.9ms, total_kv_read_wall_time: 4m0.8s, tikv_wall_time: 4m7.7s} | keep order:false | N/A | N/A |
+-------------------------------+--------------+-----------+-----------+----------------+--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+--------------------------------------------------------------------------+----------+---------+
在该查询中，Hash Join 的 Build 阶段 耗时约 30 秒 ，而 Probe 阶段 总耗时约 2 分 45 秒 。Probe 阶段有 5 个并发线程 ，平均每个线程耗时约 33 秒 。尽管 Build 端处理的数据量仅为 Probe 端的四分之一，且其处理逻辑也比 Probe 端更简单，但 Build 阶段的耗时却与 Probe 阶段相近。这直观地反映了 Build 阶段的 单线程执行 对整个查询性能的显著影响。
简单 join 的 spill
从上述的 EXPLAIN ANALYZE 结果可以看出，该 Hash Join 操作使用了 6.15 GB 的内存。我们通过设置 tidb_mem_quota_query 参数来控制单条查询的内存使用量，以触发 Hash Join 的 Spill 机制。以下是不同内存限制下的查询表现：
可以看到，尽管 Hash Join 支持 Spill 操作，但在将内存限制设置为 1 GB、2 GB 或 4 GB 时，查询均因内存超限而失败。原因在于，当前的 Hash Join Spill 机制无法 Spill 哈希表本身。该查询本身非常简单，Build 端的数据仅包含一个 int 类型的列，因此 6.15 GB 的内存消耗中，大部分是哈希表及其附属数据结构。由于哈希表无法 Spill，即使触发了 Spill 机制，也无法将内存使用量控制在设定的目标范围内。
当内存限制设置为 6 GB 时，查询终于能够顺利执行，但查询时间却增加了近 6 倍 。考虑到在不触发 Spill 的情况下，查询的内存消耗仅为 6.15 GB ，这意味着 Spill 操作仅节省了不到 *3%* 的内存使用量，但却导致查询时间增加了近 6 倍 。这一结果显然表明，当前 TiDB 的 Hash Join Spill 实现效率较低。
带不等值 join 条件的 join
测试 query 如下：
select /*+ hash_join(lineitem) */ count(*) from orders join lineitem on l_orderkey = o_orderkey and l_partkey > o_custkey
与之前的简单 join query 相比，该查询仅增加了一个非等值连接条件： l_partkey > o_custkey 。该查询的执行时间约为 143 秒。
以下是该查询的 EXPLAIN ANALYZE 结果：
mysql> explain analyze select /*+ hash_join(lineitem) */ count(*) from orders join lineitem on l_orderkey = o_orderkey and l_partkey > o_custkey;
+-------------------------------+--------------+-----------+-----------+----------------+---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+-----------------------------------------------------------------------------------------------------------------------------------------+----------+---------+
| id | estRows | actRows | task | access object | execution info | operator info | memory | disk |
+-------------------------------+--------------+-----------+-----------+----------------+---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+-----------------------------------------------------------------------------------------------------------------------------------------+----------+---------+
| HashAgg_9 | 1.00 | 1 | root | | time:2m23.2s, loops:2, RU:1187613.528820, partial_worker:{wall_time:2m23.188890662s, concurrency:5, task_num:183098, tot_wait:11m43.993928194s, tot_exec:11.904001006s, tot_time:11m55.944319636s, max:2m23.188869865s, p95:2m23.188869865s}, final_worker:{wall_time:2m23.188897834s, concurrency:5, task_num:9, tot_wait:44.454µs, tot_exec:2.241µs, tot_time:11m55.944420492s, max:2m23.188886895s, p95:2m23.188886895s} | funcs:count(1)->Column#26 | 196.6 KB | 0 Bytes |
| └─HashJoin_31 | 304437511.92 | 187489245 | root | | time:2m22.9s, loops:183099, build_hash_table:{total:30.3s, fetch:248.9ms, build:30s}, probe:{concurrency:5, total:11m55.9s, max:2m23.2s, probe:9m18.8s, fetch and wait:2m37.2s} | inner join, equal:[eq(test.orders.o_orderkey, test.lineitem.l_orderkey)], other cond:gt(test.lineitem.l_partkey, test.orders.o_custkey) | 6.72 GB | 0 Bytes |
| ├─TableReader_33(Build) | 75000000.00 | 75000000 | root | | time:160.2ms, loops:73324, cop_task: {num: 1847, max: 0s, min: 0s, avg: 33.9ms, p95: 46.6ms, tot_proc: 1m0.2s, tot_wait: 69.6ms, copr_cache_hit_ratio: 0.01, build_task_duration: 63.5µs, max_distsql_concurrency: 15}, rpc_info:{Cop:{num_rpc:1847, total_time:1m2.6s}} | data:TableFullScan_32 | 12.3 MB | N/A |
| │ └─TableFullScan_32 | 75000000.00 | 75000000 | cop[tikv] | table:orders | tikv_task:{proc max:66ms, min:0s, avg: 31.4ms, p80:39ms, p95:43ms, iters:80639, tasks:1847}, scan_detail: {total_process_keys: 74961632, total_process_keys_size: 11386071873, total_keys: 74963462, get_snapshot_time: 38.4ms, rocksdb: {key_skipped_count: 74961632, block: {cache_hit_count: 26917, read_count: 356085, read_byte: 3.05 GB, read_time: 905ms}}}, time_detail: {total_process_time: 1m0.2s, total_suspend_time: 103.3ms, total_wait_time: 69.6ms, total_kv_read_wall_time: 58s, tikv_wall_time: 1m0.5s} | keep order:false | N/A | N/A |
| └─TableReader_35(Probe) | 300005811.00 | 300005811 | root | | time:865.5ms, loops:293337, cop_task: {num: 7790, max: 0s, min: 0s, avg: 35.5ms, p95: 49.8ms, tot_proc: 4m27.2s, tot_wait: 299.2ms, copr_cache_hit_ratio: 0.00, build_task_duration: 234.5µs, max_distsql_concurrency: 15}, rpc_info:{Cop:{num_rpc:7790, total_time:4m36.7s}} | data:TableFullScan_34 | 11.7 MB | N/A |
| └─TableFullScan_34 | 300005811.00 | 300005811 | cop[tikv] | table:lineitem | tikv_task:{proc max:0s, min:0s, avg: 32.9ms, p80:44ms, p95:46ms, iters:324135, tasks:7790}, scan_detail: {total_process_keys: 299995731, total_process_keys_size: 58993503059, total_keys: 300003516, get_snapshot_time: 168.9ms, rocksdb: {key_skipped_count: 299995731, block: {cache_hit_count: 176686, read_count: 1700079, read_byte: 13.4 GB, read_time: 4.05s}}}, time_detail: {total_process_time: 4m27.2s, total_suspend_time: 463.8ms, total_wait_time: 299.2ms, total_kv_read_wall_time: 4m16.1s, tikv_wall_time: 4m28.6s} | keep order:false | N/A | N/A |
+-------------------------------+--------------+-----------+-----------+----------------+---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+-----------------------------------------------------------------------------------------------------------------------------------------+----------+---------+
在该查询中，Hash Join 的 Build 阶段 耗时约 30 秒 ，而 Probe 阶段 总耗时约 9 分 18 秒 。Probe 阶段有 5 个并发线程 ，平均每个线程耗时约 111 秒 。可以看出，增加了一个非等值连接条件后，Hash Join 的 Probe 阶段显著变慢。这是因为当前的 Hash JoinProbe 阶段以行为单位进行处理，非等值条件的执行难以完全向量化，从而导致性能下降。
3. TiDB 中 hash jion 的设计与实现
由于 TiDB 的 Hash Join 无法下推到 TiKV 执行，只能在 TiDB 单点进行计算，而上述性能问题使得大表连接操作很容易成为 TiDB 查询的瓶颈。因此，我们决定重新设计和实现 TiDB 的 Hash Join，以解决现有的已知问题。
Build 阶段的设计与实现
在设计 Hash Join 的 Build 端时，需要考虑以下关键点：
1. 哈希表的类型 ：主要有两种类型：
- 基于开放寻址法的哈希表
- 基于链式的哈希表
2. 哈希表的扩容策略 ：主要有两种策略：
- 按需扩容 ：在构建过程中根据需要动态调整哈希表的大小。
- 固定大小 ：在构建哈希表之前预先分配好大小，避免在构建过程中进行扩容。这种策略需要在接收完所有 Build 端数据后才能开始构建哈希表。
3. 哈希表并发构建方法 ：基本思路是将构建数据分区，具体实现可以选择：
- 物理分区 ：将构建数据重新组织为多个分区。
- 逻辑分区 ：不重新组织数据，仅在逻辑上进行分区。对于同一分区内的数据，如果存在多线程构建问题，可以通过锁或原子操作（CAS）来解决冲突。
4. Build 端数据的组织形式 ：主要有两种选择：
- 列式存储 ：构建数据按列存储。
- 行式存储 ：构建数据按行存储。对于 TiDB，其内部数据是按列式组织的。如果选择按行式存储构建数据，则需要对数据进行一次列转行的处理。
主流计算引擎的 Build 端设计调研
我们对一些主流计算引擎的 Build 端设计进行了调研，结果如下：
在新的 Hash Join 实现中，TiDB 的选择如下：
- 哈希表 ：采用固定大小（fixed size）的链式哈希表。
- 并发构建方法 ：采用物理分区 + 原子操作（CAS）避免冲突。
- Build 端数据组织格式 ：采用行式存储。
这一选择的核心思想是行存储、固定大小的哈希表和物理分区，主要考虑以下几点：
1. 数据局部性（Data Locality） ：在 Probe 阶段 ，采用行存储的 Build 端数据能够提供更好的数据局部性，从而提高缓存利用率和访问效率。
2. 数据布局优化 ：采用行存储为 Hash Join 专门设计数据布局提供了机会，使得某些功能和优化的实现更加简单。具体来说：
- 通过设计行存储的具体布局，可以更方便地在构建数据上进行标记，在左外连接（Left Outer Join）或 Semi Join 中使用左表作为 Build 端时都需要在 Build 数据上做标记。
- 通过设计行存储的具体布局，可以进一步优化哈希表的设计。
- 通过设计行存储的具体布局，可以优化 Probe 阶段 Join Key 比较的过程。
3. 固定大小哈希表的优势 ：通过采用固定大小的哈希表，完全消除了哈希表构建过程中的动态扩容（resize），从而显著提升 Build 阶段 的速度。
4. 分区与 Spill 的一致性 ：对构建数据进行物理分区，使得新的 Hash Join 的 Spill 算法能够基于分区实现。这种设计使得 Hash Join 与 Spill 逻辑保持一致，实现起来更加简单明了。
接下来，我们将详细介绍 Build 阶段 的设计。
行式存储的设计
在 TiDB 的新 Hash Join 实现中，行式存储的具体布局如下：
每行数据包含以下部分：
- All Column Data ：即 Build 端的数据。
- 额外信息 ：为了优化性能和功能实现，每行数据还附加了以下字段：
- Next Row Ptr ：指向另一行数据的指针。通过这个字段，可以实现链表功能，使得哈希表的每个条目可以直接存储每行数据的地址。
- Null and Used Flag Map:
- Null Map ：与 All Column Data 对应，每 1 位表示某一列是否为
NULL。 - Used Flag ：对于需要在构建数据上做标记的连接类型（如左外连接或 Semi Join），可以在此增加一个标志位，表示该行是否已被连接过。
- Serialized Key :
- 保存 Join Key 序列化后的数据。在 Probe 阶段 进行 Join Key 比较时，可以直接使用字节数组进行比较，而无需像当前哈希连接那样每次比较时都重新序列化 Build 端和 Probe 端的 Join Key。
- 对于序列化代价较高的 Join Key（如
DECIMAL类型或带有排序规则的字符串），这可以节省大量序列化时间。 - 该字段是可选的。对于一些简单的 Join Key（如
INT类型），可以直接使用 All Column Data 中的数据进行比较，这种情况下称为 Join Key 可以 内嵌 在 All Column Data 中。对于 Join Key 可以内嵌的场景，无需额外的 Serialized Key 字段。
All Columns Data 存储了在 Probe 阶段中需要使用的所有列的数据。对于固定长度的数据，我们直接存储数据本身；而对于可变长度的数据，我们采用先存储数据大小，再存储数据内容的策略。由于可变长度数据是直接存储在 All Columns Data 中的，这意味着我们无法在 O（1） 时间内随机访问 All Columns Data 中的每一列数据。因此，列的存储顺序需要结合 Probe 阶段的逻辑进行设计。
在 Probe 阶段，首先需要进行 Join Key 的比较。如果 Join Key 内嵌在 All Columns Data 中，那么它必须被放置在 All Columns Data 的最前面。对于其他非 Join Key 的列，通常情况下只有在将 Build 端（Build）行和 Probe 端（Probe）行拼接在一起时才会用到，因此对列的顺序没有特殊要求。然而，如果连接操作包含非等值连接条件，在 Join Key 匹配之后，还需要对非等值条件进行求值。非等值条件的求值需要将 Build 端和 Probe 端的行拼接在一起才能完成。
在 TiDB 当前的实现中，会将所有 Join Key 匹配的 Build 端和 Probe 端列合并到一个数据块（chunk）中，然后再对非等值条件进行求值。如果非等值条件的过滤率很高，这种将所有匹配的 Build 端和 Probe 端列拼接成一个数据块的操作可能会显得有些冗余，尤其是在 Build 端和 Probe 端列较多时。
在新的 Hash Join 设计中，当连接操作包含非等值条件时，Probe 阶段会在对非等值条件求值之前，尽量只将非等值条件所需的列拼接成数据块。根据非等值条件的求值结果，再将真正需要连接的 Build 端列和 Probe 端列拼接成数据块。由于非等值条件所需的列需要优先使用，因此 All Columns Data 中的列顺序设计如下：
Hash 表的设计
在新的 Hash Join 中，我们采用了固定大小（fixed size）的链式哈希表。哈希表的键通过 hash_value % hash_table_size 计算得出，每个哈希表条目存储构建行（build row）的地址。正如前面提到的行布局，构建行本身带有 Next Row Ptr （指向下一行为的指针）。因此，在哈希键冲突的情况下，构建行可以自然地形成一个链表。
Build 阶段整体流程的设计
Build 阶段 的主要任务是接收 Build 端数据，并基于这些数据构建哈希表。由于我们采用了固定大小（fixed size）的哈希表，因此必须在接收完所有 Build 端数据后，才能开始构建哈希表。
在 TiDB 中，数据的内存表示是列式的，而 Hash Join 中构建的数据是行式的。因此，需要在接收 Build 端数据的同时，进行列转行的操作。
Build 阶段的整体流程 如下图所示：
在 Build 阶段 ，我们将其细分为两个子阶段： Prebuild 阶段 和 Build 阶段 。
- Prebuild 阶段 ：对 Build 端数据进行分区，并将列式存储的数据转换为行式存储。
**Build 阶段：**基于转换后的行式数据构建哈希表。
这两个阶段均采用多线程并行执行。而且， Build 阶段 需要在所有 Prebuild 阶段 任务完成后才能开始。
Probe 阶段的设计与实现
在 Probe 阶段 ，由于 TiDB 已经支持多线程，因此其设计不会像 Build 阶段 那样涉及整体流程的改动。 Probe 阶段 的改进主要集中在 Probe 接口 的设计以及诸多细节优化上。具体改进如下：
Probe 接口改进
与当前 Hash Join Probe 接口相比，新的 Probe 接口 在两个主要方面进行了改进：
1. 集成哈希表查找（Hash Table Lookup） ： 将哈希表查找直接集成到 Probe 操作中，使得 Hash Join 能够根据具体的连接类型优化哈希表查找过程，尽可能减少冗余计算。
2. 以数据块（Chunk）为单位处理 ： 新的 Probe 接口 以数据块（Chunk）为单位进行处理，而不是逐行处理。这使得在连接过程中涉及的表达式计算可以充分利用向量化优化，具体受益部分包括：
- 非等值连接条件的计算 ：通过向量化执行，可以显著提升非等值条件的求值效率。
- Join Key 的序列化及哈希值计算 ：将 Join Key 的序列化和哈希值计算向量化，能够显著减少计算开销，尤其是在处理复杂 Join Key（如
DECIMAL或带排序规则的字符串）时。
Hash join lookup
在新的 Hash Join 实现中，哈希表的键通过 hash_value % hash_table_size 计算得出，而 TiDB 当前的 Hash Join 使用 hash_value 本身作为哈希表的键。显然，新的实现方式会导致更多的哈希冲突。
假设哈希表的大小为 65536 ，Build 端有以下 4 行数据：
其中， Row 1 和 Row 3 的哈希值相同，而 Row 2 和 Row 4 的哈希值与 Row 1 不同。但由于哈希表大小为 65536 ，这 4 行数据的哈希键（hash_value % hash_table_size）实际上都是 0x1234 ，因此它们在哈希表中位于同一个槽（slot）中。
假设 probe 时有两行数据：
这两行数据的哈希键（hash key）也都是 0x1234 。因此，在 Probe 阶段 ，这两行数据都需要与 Build 端的 4 行数据进行 Join Key 的比较。然而，由于它们的实际哈希值与 Build 端的 4 行数据均不匹配，这 8 次比较最终都以失败告终。我们将这种需要进行 Join Key 比较但结果失败的情况定义为 Probe Collision 。在这个例子中， Probe Collision 的次数为 8。相比之下，TiDB 当前的 Hash Join 使用哈希值作为哈希键，因此在这个例子中，当前 Hash Join 的 Probe Collision 次数为 0。
尽管我们在新的 Hash Join 中对 Join Key 比较进行了优化，但过多的 Probe Collision 仍可能引入性能问题。因此，我们在哈希表的实现中引入了 Tagged Pointer 来减少 Probe Collision 的发生。 Tagged Pointer 的原理是：在当前的 64 位操作系统中，指针使用 64 位表示，但实际地址的有效位通常少于 64 位，通常前 n 位为 0。我们可以利用这些前 n 位的 0 来存储其他信息（即 Tagged Info ）。
在 TiDB 的新 Hash Join 中， Tagged Info 使用了哈希值的前 n 位。假设 n = 24 ，以上述 Build 端数据为例， Tagged Info 如下所示（标红部分表示）：
在 Build 端进行哈希表构建时，具有相同哈希键（hash key）的多行数据会形成一个链表。链表中每个节点的 Tagged Field 用于存储该节点之后所有节点的 Tagged Info 的按位或值（bitwise OR）。以示例数据为例：
- Row 1 是链表的尾节点，因此其 Tagged Field 的值为 0x000000 。
- Row 2 的 Tagged Field 存储 Row 1 的 Tagged Info：0x123456 。
- Row 3 的 Tagged Field 存储： 0x123456 | 0x1234ac = 0x1234fe。
- Row 4 的 Tagged Field 存储： 0x1234fe | 0x123456 = 0x1234fe 。
- 哈希表对应槽位（slot）中的 Tagged Info 存储： 0x1234fe | 0x234512 = 0x3375fe 。
引入 Tagged Pointer 后，整个哈希表的结构如下图所示：
引入 Tagged Info 后，在 Probe 阶段 进行 Join Key 比较之前，会先对 Tagged Info 进行比对。以示例中的探测数据为例：
- Row 1 的 Tagged Info 为 0x423456 。由于 0x423456 | 0x3375fe ！= 0x423456 ，可以确定 Row 1 的哈希值肯定不等于 Build 端任何一行的哈希值，因此无需再进行 Join Key 的比对。
- Row 2 的 Tagged Info 为 0x123456 。由于 0x123456 | 0x3375fe == 0x123456 ，无法排除 Row 2 的哈希值可能等于 Build 端某一行的哈希值，因此需要进一步进行 Join Key 的比对。
由此可见，引入 Tagged Pointer 后， Probe Collision 的次数从 8 减少到了 4。在基于 TPC-H 50 的基准测试中，引入 Tagged Pointer 使得 Probe Collision 的次数减少了 80% 以上， Probe 阶段的整体时间也减少了约 15%。
在具体实现中，TiDB 会根据构建阶段所有行的地址信息，动态决定 n 的取值。这确保了 Tagged Pointer 的实现不依赖于具体的体系结构或操作系统。
Join key 比较
在新的 Hash Join 中，哈希表的键并不是直接使用 Join Key，而是基于 Join Key 的哈希值。因此，在 Probe 阶段 ，即使两行数据的哈希键相同，仍需进一步比较 Join Key 本身，以确保它们真正匹配。
由于在 Build 端进行列转行操作时，我们已经将 Build 端的 Join Key 序列化并存储为 Serialized Key ，因此在 Probe 阶段 进行 Join Key 比较时，可以直接使用这些 Serialized Key 进行比较。这种方法显著简化了 Join Key 比较的过程，尤其是对于那些在比较之前需要进行预处理的 Join Key 类型（例如带有排序规则的字符串类型）。通过直接使用 Serialized Key 进行比较，可以消除 Join Key 比较时的冗余计算。
构造 join 结果 chunk
除了查询哈希表和进行 Join Key 比较之外， Probe 端 还有一个重要任务是构造连接结果。构造连接结果是指将匹配成功的 {build_row, probe_row} 组合成一个新的 result_row 。由于 TiDB 中的数据以列存储格式保存在内存中，因此连接结果也需要按照列存储格式进行组织。
假设一行 probe_row 与两行 build_row 匹配成功，在 TiDB 当前的 Hash Join 实现中，连接结果的构造采用了最基础的方案，即依次将 build_row 和 probe_row 插入到结果块（result chunk）中，如下图所示：
在 TiDB 当前的 Hash Join 实现中，Build 端数据以列存储格式组织，这意味着 Build 端的两行数据在行与行之间以及列与列之间都缺乏数据局部性。虽然 Probe 端的行与行之间存在一定数据局部性，但当前的 Hash Join 并未充分利用这种局部性。
在新的 Hash Join 中，我们在构造连接结果块（chunk）时引入了两方面的优化：
- 基于数据局部性的优化 ：根据 Build 端和 Probe 端数据局部性的特点，设计了更高效的构造连接结果块的方法，以充分利用数据局部性，减少缓存未命中和数据访问延迟。
从上述介绍可以看出， Build Row 和 Probe Row 在数据局部性上具有不同特点。对于 Build Row ，它们通常随机分布在构建数据集中，彼此之间通常没有关联。然而，由于新的 Hash Join 中构建数据是按行存储的，因此 Build Row 在列与列之间具有一定的局部性。而 Probe Row 在行与行之间具有一定的局部性。根据这些特点，在新的 Hash Join 中，构造连接结果时对 Build Row 和 Probe Row 采用了不同的方法： Build Row 按行单位构造，而 Probe Row 按列单位构造。具体的构造方法如下图所示。
在构造 Build Row 的结果时，可以充分利用行存储带来的数据局部性；而在构造 Probe Row 的结果时，也可以利用 Probe Row 本身的局部性。在具体实现中，为了避免过多的额外开销， Build Row 的构造采用了 Micro Batch 方法，即每 32 行Build Row 批量处理，一次性构造 Result Chunk 。
- 针对非等值连接条件的优化 ：针对包含非等值连接条件的情况，进行了针对性的优化，以避免不必要的冗余操作，提高连接效率。
当连接操作包含非等值连接条件时，哈希表查找的结果需要进一步对非等值条件进行求值，才能最终确定连接结果。非等值条件的求值基于连接结果（join result），这意味着我们需要先根据哈希表查找的结果构造一个临时的连接结果块（join result chunk），然后在这个临时块上进行非等值条件的求值。
理论上，构造连接结果块需要将 Build 端（build row）和 Probe 端（probe row）的所有列拼接在一起。然而，非等值条件通常只涉及 Build 端和 Probe 端中少数几列，而 Build 端和 Probe 端本身可能包含许多列。如果将所有列都拼接起来，尤其是在非等值条件过滤率很高的情况下，会产生大量不必要的开销。
在新的 Hash Join 中，如果连接操作包含非等值条件，在哈希表查找之后，只会将必要的列拼接成连接结果块。在非等值条件过滤之后，再将连接结果块中剩余的列补全。
Spill 的设计与实现
在新的 Hash Join 中，引入了基于 Partition 的 Spill 方法。其核心思想是在 Build 阶段 对数据进行分区（Partition），当触发 Spill 时，根据需要释放的内存量，选择一个或多个分区进行 Spill。在 Spill 过程中，Build 端（Build）的数据将被完整地写入磁盘。在极端情况下，所有分区都可以被 Spill 到磁盘。
在 Probe 阶段 ，如果某行探测数据（Probe Row）对应的分区已经被 Spill，则该探测阶段对应的分区数据也会被 Spill 到磁盘。
Probe 阶段 结束后，如果发生了 Spill， Hash Join 将进入 Restore 阶段 。在 Restore 阶段 ，依次读取被 Spill 的 Build 端和 Probe 端分区，并重新进行连接操作。值得注意的是，这种 Spill 算法是递归的。在 Restore 阶段 的 Hash Join 仍然会在构建时进行分区，如果内存超限，可以继续触发 Spill。在具体实现中，我们限制了一个 Hash Join 因 Spill 产生的分区数量不能超过 1024 ，以避免递归无法结束的极端情况。
Spill 的整体流程如下图所示：
4. TiDB 新老 hash join 的切换与性能对比
在 TiDB v8.4.0 中，我们引入了新的 Hash Join 作为一项实验性功能。用户可以通过设置 TiDB 系统变量 tidb_hash_join_version 为 optimized 或 legacy 来启用或禁用新的 Hash Join 。在 v8.4.0 中，新的 Hash Join 支持了 Inner Join 和 Outer Join ，而在 v8.5.0 中，进一步支持了 Spill 功能。
我们基于 TPC-H 50 数据集对新旧 Hash Join 的性能进行了对比测试。以下是简单 Join query 的性能对比结果：
简单 join query
对于之前的简单 join query，启用新的 Hash Join 后，性能表现如下：
mysql> explain analyze select /*+ hash_join(lineitem) */ count(*) from orders join lineitem on l_orderkey = o_orderkey;
+-------------------------------+--------------+-----------+-----------+----------------+-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+--------------------------------------------------------------------------+----------+---------+
| id | estRows | actRows | task | access object | execution info | operator info | memory | disk |
+-------------------------------+--------------+-----------+-----------+----------------+-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+--------------------------------------------------------------------------+----------+---------+
| HashAgg_9 | 1.00 | 1 | root | | time:25.1s, loops:2, RU:310277.429136, partial_worker:{wall_time:25.07950759s, concurrency:5, task_num:292976, tot_wait:1m44.3707804s, tot_exec:20.948357444s, tot_time:2m5.397167833s, max:25.07944457s, p95:25.07944457s}, final_worker:{wall_time:25.079546556s, concurrency:5, task_num:9, tot_wait:97.733µs, tot_exec:1.409µs, tot_time:2m5.397547492s, max:25.079514741s, p95:25.079514741s} | funcs:count(1)->Column#26 | 277.8 KB | 0 Bytes |
| └─HashJoin_31 | 304437511.92 | 300005811 | root | | time:24.8s, loops:292977, build_hash_table:{total:5.62s, fetch:2.12s, build:2.9s}, probe:{concurrency:5, total:2m5.4s, max:25.1s, probe:38.9s, fetch and wait:1m26.5s, probe_collision:62264245} | inner join, equal:[eq(test.orders.o_orderkey, test.lineitem.l_orderkey)] | 4.63 GB | 0 Bytes |
| ├─TableReader_33(Build) | 75000000.00 | 75000000 | root | | time:1.52s, loops:73342, cop_task: {num: 1847, max: 0s, min: 0s, avg: 32.4ms, p95: 56.6ms, tot_proc: 58.1s, tot_wait: 57.1ms, copr_cache_hit_ratio: 0.01, build_task_duration: 42.1µs, max_distsql_concurrency: 15}, rpc_info:{Cop:{num_rpc:1847, total_time:59.8s}} | data:TableFullScan_32 | 2.68 MB | N/A |
| │ └─TableFullScan_32 | 75000000.00 | 75000000 | cop[tikv] | table:orders | tikv_task:{proc max:63ms, min:0s, avg: 31ms, p80:40ms, p95:55ms, iters:80639, tasks:1847}, scan_detail: {total_process_keys: 74946240, total_process_keys_size: 2023548480, total_keys: 74948071, get_snapshot_time: 31.2ms, rocksdb: {key_skipped_count: 74946240, block: {cache_hit_count: 27002, read_count: 355924, read_byte: 3.05 GB, read_time: 925.7ms}}}, time_detail: {total_process_time: 58.1s, total_suspend_time: 99.4ms, total_wait_time: 57.1ms, total_kv_read_wall_time: 57.2s, tikv_wall_time: 58.4s} | keep order:false | N/A | N/A |
| └─TableReader_35(Probe) | 300005811.00 | 300005811 | root | | time:10.9s, loops:293404, cop_task: {num: 7790, max: 0s, min: 0s, avg: 35.8ms, p95: 67.4ms, tot_proc: 4m32s, tot_wait: 243ms, copr_cache_hit_ratio: 0.00, build_task_duration: 177.8µs, max_distsql_concurrency: 15}, rpc_info:{Cop:{num_rpc:7790, total_time:4m39.1s}} | data:TableFullScan_34 | 2.30 MB | N/A |
| └─TableFullScan_34 | 300005811.00 | 300005811 | cop[tikv] | table:lineitem | tikv_task:{proc max:0s, min:0s, avg: 34.2ms, p80:49.1ms, p95:65ms, iters:324135, tasks:7790}, scan_detail: {total_process_keys: 299968435, total_process_keys_size: 10798863660, total_keys: 299976209, get_snapshot_time: 136.2ms, rocksdb: {key_skipped_count: 299968435, block: {cache_hit_count: 183434, read_count: 1693141, read_byte: 13.3 GB, read_time: 4.27s}}}, time_detail: {total_process_time: 4m32s, total_suspend_time: 507.9ms, total_wait_time: 243ms, total_kv_read_wall_time: 4m26s, tikv_wall_time: 4m33.3s} | keep order:false | N/A | N/A |
+-------------------------------+--------------+-----------+-----------+----------------+-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+--------------------------------------------------------------------------+----------+---------+
具体分析各个算子的时间：
可以看到， Hash Join Build 时间提升了约 10 倍 ，而 Hash Join Probe 时间提升了约 4 倍 。整体查询时间提升了约 2.5 倍 。之所以整体查询时间的提升没有达到 Build 和 Probe 时间提升的水平，主要是因为在新的 Hash Join 中， Table Scan 成为了新的瓶颈。在旧的 Hash Join 中， Table Scan 时间仅为 0.98 秒 ，而在新的 Hash Join 中， Table Scan 时间增加到了 12.4 秒 。
需要指出的是，旧的 Hash Join 中 Table Scan 时间短并不是因为其扫描速度特别快。实际上，在 TiDB 的执行模型中，各个算子在查询开始执行时会通过各自的 Goroutine 独立运行，算子之间通过类似 Channel 的数据结构进行通信。在 TiDB 的 Explain Analyze 中，算子的时间是指其父算子在 Channel 上等待的时间。
在旧的 Hash Join 中，由于 Hash Join 本身执行缓慢，其底层的 Table Scan 有足够的时间从 TiKV 读取数据并放入相应的 Channel 中。当 Hash Join 需要读取数据时，可以直接从 Channel 中获取，因此在旧的 Hash Join 中， Table Scan 的时间显得很短。而在新的 Hash Join 中，由于 Hash Join 执行速度更快，当其尝试从 Table Scan 算子读取数据时，TiKV 的数据可能尚未完全读取完成，导致读取数据时会阻塞在 Table Scan 上。因此，在 Explain Analyze 中， Table Scan 的时间会显著增加。
简单 join 的 spill
从上述 Explain Analyze 结果可以看出，使用新的 Hash Join 后，其内存占用量从之前的 6.15 GB 降低到了 4.63 GB 。这一改进主要得益于新的 Hash Join 中哈希表设计的优化，去除了某些不必要的中间数据结构，从而减少了内存使用。
同样地，我们通过设置 tidb_mem_quota_query 来触发 Hash Join 的 Spill 操作。
可以看到，即使将内存配额（memory quota）设置为 1 GB ，新的 Hash Join 仍然能够通过 Spill 顺利执行。与不发生 Spill 的情况相比，其性能下降不超过 100% 。相比之下，旧的 Hash Join 在 Spill 机制上存在明显不足。新的 Hash Join 在控制内存使用的效果以及查询的整体效率方面，相较于旧版本都有显著提升。
带不等值 join 条件的 join
对于带非等值 Join 条件的连接操作，其在新 Hash Join 中的性能表现如下：
mysql> explain analyze select /*+ hash_join(lineitem) */ count(*) from orders join lineitem on l_orderkey = o_orderkey and l_partkey > o_custkey;
+-------------------------------+--------------+-----------+-----------+----------------+------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+-----------------------------------------------------------------------------------------------------------------------------------------+----------+---------+
| id | estRows | actRows | task | access object | execution info | operator info | memory | disk |
+-------------------------------+--------------+-----------+-----------+----------------+------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+-----------------------------------------------------------------------------------------------------------------------------------------+----------+---------+
| HashAgg_9 | 1.00 | 1 | root | | time:29.3s, loops:2, RU:1205997.866144, partial_worker:{wall_time:29.28368721s, concurrency:5, task_num:183098, tot_wait:2m14.285571269s, tot_exec:12.083749663s, tot_time:2m26.418187777s, max:29.283640334s, p95:29.283640334s}, final_worker:{wall_time:29.283735588s, concurrency:5, task_num:9, tot_wait:5.908µs, tot_exec:1.683µs, tot_time:2m26.418341222s, max:29.2836757s, p95:29.2836757s} | funcs:count(1)->Column#26 | 277.8 KB | 0 Bytes |
| └─HashJoin_31 | 304437511.92 | 187489245 | root | | time:29.1s, loops:183099, build_hash_table:{total:6.38s, fetch:2.4s, build:3.09s}, probe:{concurrency:5, total:2m26.4s, max:29.3s, probe:1m3.5s, fetch and wait:1m22.9s, probe_collision:62253564} | inner join, equal:[eq(test.orders.o_orderkey, test.lineitem.l_orderkey)], other cond:gt(test.lineitem.l_partkey, test.orders.o_custkey) | 5.47 GB | 0 Bytes |
| ├─TableReader_33(Build) | 75000000.00 | 75000000 | root | | time:1.8s, loops:73346, cop_task: {num: 1847, max: 0s, min: 0s, avg: 37.6ms, p95: 68.1ms, tot_proc: 1m7.3s, tot_wait: 59.2ms, copr_cache_hit_ratio: 0.01, build_task_duration: 49.5µs, max_distsql_concurrency: 15}, rpc_info:{Cop:{num_rpc:1847, total_time:1m9.4s}} | data:TableFullScan_32 | 4.60 MB | N/A |
| │ └─TableFullScan_32 | 75000000.00 | 75000000 | cop[tikv] | table:orders | tikv_task:{proc max:77ms, min:0s, avg: 35.2ms, p80:45ms, p95:64.1ms, iters:80639, tasks:1847}, scan_detail: {total_process_keys: 74953536, total_process_keys_size: 11384841043, total_keys: 74955363, get_snapshot_time: 32.5ms, rocksdb: {key_skipped_count: 74953536, block: {cache_hit_count: 27064, read_count: 355890, read_byte: 3.05 GB, read_time: 942.7ms}}}, time_detail: {total_process_time: 1m7.3s, total_suspend_time: 119ms, total_wait_time: 59.2ms, total_kv_read_wall_time: 1m4.9s, tikv_wall_time: 1m7.6s} | keep order:false | N/A | N/A |
| └─TableReader_35(Probe) | 300005811.00 | 300005811 | root | | time:9.93s, loops:293422, cop_task: {num: 7790, max: 0s, min: 0s, avg: 41.7ms, p95: 78.2ms, tot_proc: 5m15.5s, tot_wait: 256.5ms, copr_cache_hit_ratio: 0.00, build_task_duration: 198.3µs, max_distsql_concurrency: 15}, rpc_info:{Cop:{num_rpc:7790, total_time:5m24.4s}} | data:TableFullScan_34 | 4.63 MB | N/A |
| └─TableFullScan_34 | 300005811.00 | 300005811 | cop[tikv] | table:lineitem | tikv_task:{proc max:0s, min:0s, avg: 39ms, p80:56ms, p95:74ms, iters:324135, tasks:7790}, scan_detail: {total_process_keys: 299970579, total_process_keys_size: 58988557058, total_keys: 299978350, get_snapshot_time: 143.9ms, rocksdb: {key_skipped_count: 299970579, block: {cache_hit_count: 184621, read_count: 1691960, read_byte: 13.3 GB, read_time: 4.33s}}}, time_detail: {total_process_time: 5m15.5s, total_suspend_time: 592.4ms, total_wait_time: 256.5ms, total_kv_read_wall_time: 5m3.6s, tikv_wall_time: 5m16.9s} | keep order:false | N/A | N/A |
+-------------------------------+--------------+-----------+-----------+----------------+------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+-----------------------------------------------------------------------------------------------------------------------------------------+----------+---------+
具体分析各个算子的时间：
通过实现非等值 Join 条件的向量化执行，显著提升了 Probe 阶段的效率。具体而言， Hash Join Probe 时间提升了近 10 倍 ，而查询的整体时间也实现了接近 5 倍 的提升。
TPCH-50 benchmark
最后，我们对 TPC-H 50 基准测试在新旧 Hash Join 中的性能进行了对比测试：
从测试结果来看， TPC-H 整体查询时间提升了约 15% 。具体到 Hash Join Build 和 Hash Join Probe 的时间， Hash Join Build 时间提升了约 10 倍 ，而 Hash Join Probe 时间提升了约 2.7 倍 。
整体查询时间的提升幅度不如 Hash Join 本身的提升幅度，主要原因是对于许多查询而言，Join 操作并非唯一的瓶颈。在新的 Hash Join 提升了连接性能之后，其他算子（如表扫描、聚合等）可能会成为新的瓶颈。
5. 总结
本文介绍了 TiDB 中新 Hash Join 的设计与实现。相较于 TiDB 中现有的 Hash Join ，新 Hash Join 在 Build 端引入了并发构建机制和新的哈希表，使得 Build 性能提升了约 10 倍 。在 Probe 端，通过优化的实现方式， Probe 性能提升了 2 - 4 倍 。此外，针对 Hash Join 的 Spill 操作，设计了全新的算法。该算法不仅能够更彻底地通过 Spill 释放内存占用，还使 Spill 性能提升了 5 - 10 倍 。