Semi-Join的并行加速

前提条件

PolarDB集群版本需为PolarDB MySQL版8.0版本且修订版本满足以下条件：

• 8.0.1.0.5 或以上。

• 8.0.2.2.7 或以上。

如何查看集群版本，请参见查询版本号。

背景信息

MySQL 5.6.5引入了Semi-Join半连接，当外表在内表中找到匹配的记录之后，Semi-Join会返回外表中的记录。但即使在内表中找到多条匹配的记录，外表也只会返回已经存在于外表中的记录。而对于子查询，外表的每个符合条件的元组都要执行一轮子查询，效率比较低下。此时使用半连接操作优化子查询，会减少查询次数，提高查询性能。其主要思路是将子查询上拉到父查询中，这样内表和外表是并列关系，外表的每个符合条件的元组，只需要在内表中找符合条件的元组即可，所以效率会大大提高。

策略

Semi-Join主要使用了如下策略：

• DuplicateWeedout Strategy

  该策略创建由row id组成唯一ID的临时表，再通过该唯一ID达到去重目的。

  

• Materialization Strategy

  该策略将nested tables物化到临时表中，再通过查找物化表或者遍历物化表查找外表的方法达到去重目的。

  

• Firstmatch Strategy

  该策略采用顺序查找表的方式，找到第一个匹配的记录后立即跳转到最后一个外表，并对外表的下一条记录执行JOIN操作，从而达到去重的目的。

  

• LooseScan Strategy

  该策略对内表基于索引（Index）进行分组，分组后与外表执行JOIN（进行Condition的匹配）操作，如果存在匹配的记录，则提取外表的记录，内表选取下一个分组继续进行计算，从而达到去重目的。

  

语法

Semi-Join通常使用IN或EXISTS作为连接条件。

• IN

  SELECT * FROM Employee WHERE DeptName IN (   SELECT DeptName   FROM Dept )

• EXISTS

  SELECT * FROM Employee WHERE EXISTS (   SELECT 1   FROM Dept   WHERE Employee.DeptName = Dept.DeptName )

并行Semi-Join性能提升

对于选择Semi-Join策略的查询，PolarDB对Semi-Join所有策略实现了并行加速。通过拆分Semi-Join的任务，多线程模型并行运行任务集，强化去重能力，使查询性能得到了显著的提升。在PolarDB 8.0.2.2.7后支持对物化策略（Semi-Join Materialization）的多阶段并行查询，进一步提升了Semi-Join的查询性能，以Q20为例进行说明。

SELECT
   s_name,
   s_address 
FROM
   supplier,
   nation 
WHERE
   s_suppkey IN 
   (
      SELECT
         ps_suppkey 
      FROM
         partsupp 
      WHERE
         ps_partkey IN 
         (
            SELECT
               p_partkey 
            FROM
               part 
            WHERE
               p_name LIKE '[COLOR]%' 
         )
         AND ps_availqty > ( 
         SELECT
            0.5 * SUM(l_quantity) 
         FROM
            lineitem 
         WHERE
            l_partkey = ps_partkey 
            AND l_suppkey = ps_suppkey 
            AND l_shipdate >= date('[DATE]') 
            AND l_shipdate < date('[DATE]') + interval '1' year ) 
   )
   AND s_nationkey = n_nationkey 
   AND n_name = '[NATION]' 
ORDER BY
   s_name;

本文例子中，子查询和外层查询都以并行度（DOP）为32并行执行，子查询首先并行生成物化表，之后外层查询也并行的进行后续处理，充分发挥CPU的处理能力，将查询并行能力最大化。下文展示了在标准TPC-H中，SCALE为100 GB的数据量热数据场景下，开启并行后多阶段的并行处理能力。

并行的执行计划如下：

 -> Sort: <temporary>.s_name  (cost=5014616.15 rows=100942)
    -> Stream results
        -> Nested loop inner join  (cost=127689.96 rows=100942)
            -> Gather (slice: 2; workers: 64; nodes: 2)  (cost=6187.68 rows=100928)
                -> Nested loop inner join  (cost=1052.43 rows=1577)
                    -> Filter: (nation.N_NAME = 'KENYA')  (cost=2.29 rows=3)
                        -> Table scan on nation  (cost=2.29 rows=25)
                    -> Parallel index lookup on supplier using SUPPLIER_FK1 (S_NATIONKEY=nation.N_NATIONKEY), with index condition: (supplier.S_SUPPKEY is not null), with parallel partitions: 863  (cost=381.79 rows=631)
            -> Single-row index lookup on <subquery2> using <auto_distinct_key> (ps_suppkey=supplier.S_SUPPKEY)
                -> Materialize with deduplication
                    -> Gather (slice: 1; workers: 64; nodes: 2)  (cost=487376.70 rows=8142336)
                        -> Nested loop inner join  (cost=73888.70 rows=127224)
                            -> Filter: (part.P_NAME like 'lime%')  (cost=31271.54 rows=33159)
                                -> Parallel table scan on part, with parallel partitions: 6244  (cost=31271.54 rows=298459)
                            -> Filter: (partsupp.PS_AVAILQTY > (select #4))  (cost=0.94 rows=4)
                                -> Index lookup on partsupp using PRIMARY (PS_PARTKEY=part.P_PARTKEY)  (cost=0.94 rows=4)
                                -> Select #4 (subquery in condition; dependent)
                                    -> Aggregate: sum(lineitem.L_QUANTITY)
                                        -> Filter: ((lineitem.L_SHIPDATE >= DATE'1994-01-01') and (lineitem.L_SHIPDATE < <cache>((DATE'1994-01-01' + interval '1' year))))  (cost=4.05 rows=1)
                                            -> Index lookup on lineitem using LINEITEM_FK2 (L_PARTKEY=partsupp.PS_PARTKEY, L_SUPPKEY=partsupp.PS_SUPPKEY)  (cost=4.05 rows=7)

在标准TPC-H 100 GB的数据量，热数据场景下，串行的执行时间：

多机并行开启情况下的执行时间：

可以看到执行时间从43.52秒减少到了2.29秒，性能提升19倍。