【技术干货】TDSQL 列存引擎 LibraDB 中的Right Semi/Anti Hash Join设计

导语

在探索关系型数据库的查询优化技术时，我们经常会遇到各种Join操作，它们是构建复杂查询的基础。在这些Join操作中，Semi/Anti Join是一类有趣且实用的概念，它在处理特定类型的查询时显示出其独特的效率和优势。Semi/Anti操作是特殊类型的连接，它的核心目的是从一个表中查询出与另一个表存在/不存在关联的行。

与传统连接不同的是，Semi Join仅关注是否存在至少一个匹配，而不关心有多少个匹配。这意味着，即使外表中的一行在内表中有多个匹配项，Semi也只会返回这一行一次，而不会重复返回; 而Anti Join关注的是在内表中没有匹配项的外表行。这种操作在数据分析和处理中非常有用，尤其是当我们只需要确认关联存在，而不需要关联的具体内容时，避免了传统Join操作中可能产生的大量重复数据从而加快了查询执行性能。

为防止歧义，本文所指Right侧为Hash Join的Build侧。本文旨在通过详细讲解Semi/Anti Join在LibraDB中的实现优化细节，帮助读者深入理解这些特殊Join操作的技术独特性，分享腾讯云数据库团队在数据库查询优化方面的技术探索和实践。

1.背景

在数据库查询优化的实践中，我们经常会遇到需要转换或重写查询的情况，以提高执行效率。考虑以下SQL查询，它旨在从表 T1中查找特定的行，这些行的 V2 列与表 T2 中至少一行的 V2 列值相等。

EXPLAIN SELECT t1.v1 FROM t1 WHERE EXISTS (SELECT t2.v1 FROM t2 WHERE t2.v2 = t1.v2);

在执行计划中，优化器决定采用 Semi Join 的策略，通过哈希连接（Hash Join）来实现。在这个过程中，表T2的数据被用作构建哈希表（Build table）的基础，而表T1的数据则用于Probe这个哈希表，以获取最终结果。这种改写在通用情况下明显优于子查询的原始执行方式，因为它减少了不必要的嵌套循环执行。

然而，这个执行计划并非没有缺点。如果表T1的数据量相对较小，而T2的数据量很大，那么在构建Hash Map的阶段，性能可能会遇到瓶颈。通常，优化器会倾向于选择数据量较小的表来构建哈希映射，因为这样可以减少内存占用并加快构建速度。

如果我们考虑直接在执行计划中交换表T2和表T1的角色，可能会认为这样做可以解决上述瓶颈问题。但是，在大多数情况下，这样的交换会导致错误的结果。

1.1 Broadcast T1的正确性问题

在分布式数据库系统中，数据通常被分散存储在多个节点上，这就要求查询执行计划能够有效地处理跨节点的数据操作。考虑一种情况，我们尝试通过将表T1的所有数据广播（Broadcast）到T2表数据所在的各个列存计算节点上，来执行一个 Right Semi Join。这种策略的初衷是利用广播来简化数据的查找过程，确保每个节点都有完整的T1数据集来与本地的T2数据进行匹配。

然而，这种方法在实际操作中可能会引入数据冗余。由于每个节点都会独立执行半连接操作，满足连接条件的T1表中的行可能会在多个节点上重复出现。如下图所示，如果T1中的某个值3在T2的多个分区(T2_P1、T2_P2分布在不同计算节点上)中都找到了匹配项，那么每个分区都会输出这个值3，导致最终结果中3被冗余地计算了多次。

这种情况下，虽然理论上我们期望的结果是唯一的3，但实际输出却可能包含多个3，从而违背了半连接的初衷——每个匹配的T1行应该只被输出一次。

1.2 按Inner Join半连接的语意错误问题

在上述的分布式查询例子中，如果T1表作为Probe侧，那么按照内连接的执行方式去输出满足条件的T1行是可行的。这是因为Probe操作通常涉及到在已构建的哈希表中查找匹配项，这样每个T1行都会与T2表中的匹配行进行一一对应，从而确保了输出的准确性。

然而，如果T1表作为构建（Build）侧，那么对于满足连接条件的T1行去输出就可能会产生冗余。因为在构建哈希表时，每个T2表中的行都会去T1的哈希表中查找匹配项，如果T1中的某个值在T2中有多个匹配，那么这个值就会被重复输出多次。这种情况下的输出与Semi Join的语义不符，因为Semi Join要求即使存在多个匹配，每个T1行也只能输出一次。

2.Right Semi/ Anti Join执行 

为了解决1.2所述正确性问题，对于Hash Join而言，在Right Semi/Anti Join场景下，需要能保证输出的是符合条件的Build侧数据，即已经构建成Hash Map中符合Join条件的数据，并且需要保证符合条件的行只输出一次。

它的执行逻辑和Inner Join存在一些差异，Inner Join可以在匹配的过程中流式地输出符合连接条件行，对于Right Semi而言，尽管逻辑上可以在匹配过程中动态删除Hash Map中行，但对于多线程并发的执行引擎而言需要付出额外保证并发安全的开销。

因此，LibraDB选择在Join过程中统一标记Match/UnMatch行，之后再统一输出符合条件行。除此之外，为了尽可能提升性能并解决实际场景中的内存问题和数据倾斜问题，LibraDB实现了多个关键特性，包括Shared-Hash Join、自动内存管理、Pipeline向量化执行引擎和Runtime Filter等。

2.1Shared -hash join

LibraDB采用了Shared-hash join的方法，这是一种内存友好的Join策略。在这种模式下，单个节点上的多个线程共享同一个Two-level Hash Map对象，协同进行构建操作, 这种设计的优势在于，它可以显著减少内存消耗，因为不需要为每个线程分配独立的Hash Map。

此外，通过在节点层面共享Hash Map，系统能够更有效地应对数据倾斜问题，从而提高整体的计算效率。

2.1.1 哈希函数特化

在理想情况下，如果我们能够提前了解数据的规模和分布，我们通常可以选择最适合的哈希函数。不同的哈希函数在碰撞率和性能上表现各异，而在Join操作中，我们无法预知Hash Map的负载因子。

因此，以哈希函数的性能为导向，可以在大多数工作负载上为Join操作取得相对较好的性能表现。实际上，在现代CPU执行体系下，哈希函数的指令越简单，越容易被CPU指令缓存，并且在很多场景下可以避免函数调用，从而实现内联执行。

因此，LibraDB执行引擎尽可能选择指令简单的哈希函数进行计算。尽管这些哈希函数在冲突表现上可能不尽如人意，但在大多数场景中获得的性能收益是值得的。

此外，一些硬件指令集支持的哈希函数相比软件实现有显著的性能提升。在SMHasher测试套件中，可以看到不同哈希函数的基准测试结果，例如xxhash就取得了不错的效果。

但在实际的Hash Table使用上哈希函数的选择仍然存在优化的空间，应该说要白盒化的去使用哈希函数而不是黑盒。例如，对于1个字节或者2个字节的这种16位以内的hash key, 如果可以保证Hash Map的槽位，无需使用哈希函数单独计算也是常见的优化手段。

2.1.2 自动内存管理

为了尽可能减少在链式哈希表在冲突检测时产生的缓存失效问题，LibraDB采用开放寻址的内存布局，保证内存的连续性来提升缓存的友好性。

在实际业务场景中，一个很常见的场景是内存无法存下整个哈希表，基于此类场景，LibraDB已支持将Join的Hash Map进行分区，分区落盘后可以将Build和Probe侧的数据按分区进行Join，可以有效解决内存不足的问题，这种分区的思想来源于Hybrid/Grace Hash Join。

但仍然需要解决应该在什么时候落盘，什么时候可以用更多内存的问题。如果只是设置一个内存阈值，将会陷入参数调整的困境，因为很难有一组参数可以适配所有场景。

如果抽象来看待这个问题，在执行引擎例如 Hash Join、Sort、Hash Group by 等算子都有持有大量内存做计算的需求，如果我们有一个全局的内存观察者和调度者视野，我们就可以有机会去为多个query的多个执行算子去自动化合理地分配内存使用。

当前LibraDB已支持全局内存自动管理，每个查询中的多个使用计算内存的算子会在全局内存管理器上注册自己的使用内存预估值，在真实处理数据的过程中不断和内存管理器进行交互，来动态决策能否扩大内存使用或是进入落盘对应逻辑或者是降低内存使用等等。

这并不意味着真实内存申请会在全局内存管理中发生，只是进行逻辑上的自动管理。这个特性对于产品的易用性、执行引擎的鲁棒性、并发Query的吞吐量都有不错的提升。

2.2线程安全的标记Match/UnMatch

相比Inner Join, 在执行Right Semi/Anti Join时，系统需要能够识别出在Hash Map中匹配或不匹配的行。LibraDB通过在Probe过程中对匹配的行进行标记来解决这一问题。

这个标记过程是可重入的，并且无需加锁，从而提高了并发性能。一旦Join Probe操作完成，执行引擎会来统一并行处理并输出这些标记过的行。如果执行引擎的数据流和线程是绑定的，任何一个等待Probe完成的过程需要引入复杂的线程同步机制。

而在LibraDB的Pipeline执行引擎中，由于数据流和线程解耦，调度框架可以根据各个执行算子的Port状态来保证这种先后的依赖关系被正确执行。

2.3Pipeline向量化执行

LibraDB中已实现Pipeline向量化执行引擎，执行引擎维护了算子执行的DAG图， 通过连接的上下游关系来控制执行流程，每个算子通过InputPort和OutputPort来实现数据和状态的流转，相邻算子之间会传递列存Block，并在算子计算内部尽可能向量化执行。

在Right Semi/Anti Join的场景中，Delayed Ports机制被引入以解决同步问题，并利用Pipeline调度执行框架实现了线程与数据的解耦。在这种机制下，Join操作的输出端口会连接到Delayed Input Port，该端口所连接的算子只有在Join操作完成后才会被调度执行。

这样的设计允许线程资源在不需要同步的情况下被释放，以便执行查询计划中其他Pipeline部分的任务，从而更合理地利用CPU资源。

 2.4并行输出

Join Probe操作完成后，Hash Map中仍然存储着匹配的行。根据Semi/Anti Join的语义，系统需要输出这些行。这一过程仍然可以并行执行，归功于可以将Hash Map按多个分片拆分给到多个获取数据的Source算子使用。

这种并行处理能够进一步提高查询的执行效率，确保即使在高并发的场景下，系统也能够快速准确地返回结果。

3.Optimizer 考虑

在TDSQL这样的分布式数据库系统中，在列存引擎中执行Join操作的实现通常涉及将参与连接的数据根据特定的Shuffle策略分发到多个节点进行处理。

为了解决1.1所述正确性问题，在数据库查询优化中，需要确保查询计划的正确性并选择最佳的连接顺序，这些都是优化器的关键任务。以下是针对上述问题，优化器需要考虑的几个方面：

3.1 Cost-based join order

在确定连接顺序时，优化器通常会采用动态规划方法来评估不同连接路径的代价，并选择成本最低的连接顺序。这个过程中需考虑分布式计算的额外代价以及Runtime Filter对Join order的影响。

在许多情况下，选择数据量较小的表作为哈希连接的Build侧是一种常见且有效的策略，因为这样可以减少内存占用并提高连接效率。

3.2 Shuffle方式的选择

对于Right Semi/Anti Join，选择合适的Shuffle方式是至关重要的。在分布式环境中，除了单机场景外，通常应避免对Right侧进行广播（Broadcast），因为这通常会导致错误的结果。

在这个场景中支持的Shuffle方式包括Colocate、Partition Wise、Repartition和Hash等。优化器基于表数据的物理位置分布和计算成本来决定最合适的Shuffle方式，以确保查询的高效执行。

3.3 Probe侧去重

对于连接的左侧即Probe侧，如果表的实际行数与不同值的数量（NDV）之间存在较大差异，那么在执行连接操作之前对左侧进行去重可能会带来性能上的提升。去重可以减少连接操作处理的数据量，从而减少计算的开销。

然而，这一步骤是可选的，因为它可能会引入额外的计算成本。优化器需要权衡去重操作的潜在好处与其成本，以确定是否执行去重操作。

3.4 Runtime Filter

在执行Right Semi/Anti Join时，一种提升性能的策略是利用 Runtime Filter 。这种技术涉及在Join操作的构建阶段生成一个动态的过滤器，然后将这个过滤器 Push Down 到 Probe 阶段的基表上。

通过这种方式，可以在数据实际参与Join之前，提前筛选掉不满足条件的记录，从而减少在网络上传输的数据量，显著提高整体查询的效率。

Runtime Filter的优势在于它是基于实际参与Join的数据动态生成的，因此能够更精确地反映当前查询的上下文。

这种精确性使得Runtime Filter成为一种强大的工具，它可以有效地减少无关数据的处理，尤其是在分布式数据库环境中，这对于降低网络I/O和减少不必要的数据计算尤为重要， 当前LibraDB已实现 Min/Max, Bloom, In-list三种不同的Global/Local Filter满足多种场景需要。

4.性能测试

这里在Right Semi/Anti场景简单测试了LibraDB执行性能， 其中机器CPU型号为: Intel(R) Xeon(R) Gold 6133 CPU @ 2.50GHz

SQL:

SELECT Count(A.v1)

FROM   (SELECT t1.v1

       FROM t1

       WHERE  EXISTS (SELECT t2.v1

                      FROM t2

                      WHERE t2.v2 = t1.v2)) A;

LibrDB执行计划示意:

性能如下图示, 这个场景中每列数据分布为[1,Size]，因此对于这个Semi Join是一个完全匹配的场景。从图中可以看到，在不同的数据量下均随并行度增加执行时间呈现减少趋势。

5.未来分享

在Parallel Hash Join的实现上，在提前已经打散例如基于Join Key进行Hash Shuffle，进一步地将数据分配到不同的Pipeline分支上而不是不同的计算节点上，让每个Pipeline或者线程执行时都能独立地构建自己的小Hash Map去Probe数据，这样的话能够在内存访问上能够进一步利用NUMA架构的特性，即有机会通过绑核来避免跨CPU的内存访问，从而获得更低的内存访问延迟来提升整体系统执行性能。

这种利用NUMA架构特性的优化或者充分利用硬件特点去进一步提升数据库性能的可能性已经成为当前数据库发展的热门方向。

在数据库执行引擎层面，业务中会发现存在越来越多的自适应能力的需求，这就要求执行引擎在各个场景都需要表现出良好的性能鲁棒性。

这种自适应能力不仅体现在功能层面，例如Hash Join解决倾斜的Hybrid Hash方案、两阶段聚合自适应Bypass处理、自适应的资源(DOP/内存等)调度管理、Runtime Filter自适应Disable/Enable能力、执行计划演进、各种减少Memory Stalling开销等等已经非常多的地方可以优化提升。但除此之外，如果深入实现细节，在代码微观层面，仍然存在很多未能解决的问题。

以分支预测为例，分支预测失败破坏CPU执行流水线已深入人心，但不意味着去掉分支预测直接计算在所有场景能取得最好的效果，因为实际上还取决于选择率，对于不同的数据分布，执行引擎是否有能力选择到不同的执行路径是值得探讨的。

又例如，当前部分厂商的实现对于Filter操作是采用的Select Vector的形式，即会重新组织出新的满足条件的数据，在Filter过滤性较差的情况下，这么做的开销显然不小。

总之，技术始终要为业务服务，TDSQL未来会与各类业务场景紧密结合，共同优化和提升。

﹀

﹀

﹀

-- 更多精彩 --

技术干货丨TDSQL 列存引擎 LibraDB 计算模型的设计与思考

技术干货丨 TDSQL for MySQL DDL执行框架

分布式数据库时代，需要什么样的产品？