GaussDB(DWS)分布式Join原理浅析

前言

单机场景下两表关联（join）比较简单，但是分布式场景下，由于数据分布在不同的Datanode（DN），DWS是如何对两表进行Join的呢？为什么有时候两表关联时会出现Stream算子，有时候没有呢？接下来，本文简单介绍下GaussDB(DWS)在分布式下的两表Join的实现方法。

几种典型场景

GaussDB(DWS)当前支持表，Hash分布表和复制表。其中Hash表的数据根据分布键的Hash值分布在不同DN，复制表在每个DN节点之间都有全量数据。因此，DWS中两表join可以分为如下几种场景。

说明：前置场景如下，其中t1、t2表为Hash表，tt1、tt2表为复制表，为了方便执行计划展示，已设置enable_fast_query_shipping为off，建表语句如下：

create table t1(a int, b int, c int) distribute by hash(a);
create table t2(a int, b int, c int) distribute by hash(a);
create table tt1(a int, b int, c int) distribute by replication;
create table tt2(a int, b int, c int) distribute by replication;

场景一：Hash表和Hash表Join

分布式场景下Hash表和Hash表进行关联的核心思想总结成一句话就是：两表满足join条件的数据分布在同一个DN上，不在同一个DN上的数据一定不满足join条件

1. join列都包含分布列

这种场景本身就满足核心思想：两表满足join条件的数据分布在同一个DN上，不在同一个DN上的数据一定不满足join条件，因此可以直接在各DN上进行Join，各DN执行完成后结果返回给CN，执行计划如下：

postgres=# explain select * from t1 left join t2 on t1.a = t2.a;
                                 QUERY PLAN                                  
-----------------------------------------------------------------------------
  id |          operation           | E-rows | E-memory | E-width | E-costs 
 ----+------------------------------+--------+----------+---------+---------
   1 | ->  Streaming (type: GATHER) |     20 |          |      24 | 32.56   
   2 |    ->  Hash Left Join (3, 4) |     20 | 1MB      |      24 | 26.56   
   3 |       ->  Seq Scan on t1     |     20 | 1MB      |      12 | 13.13   
   4 |       ->  Hash               |     21 | 16MB     |      12 | 13.13   
   5 |          ->  Seq Scan on t2  |     20 | 1MB      |      12 | 13.13   
 
 Predicate Information (identified by plan id)
 ---------------------------------------------
   2 --Hash Left Join (3, 4)
         Hash Cond: (t1.a = t2.a)

2. join条件只包含一张表的分布列

该场景下，为了满足核心思想，需要将不包含分布列的一侧做重分布，从而满足要求。执行计划如下：

postgres=# explain select * from t1 left join t2 on t1.b = t2.a;
                                       QUERY PLAN                                       
----------------------------------------------------------------------------------------
  id |                operation                | E-rows | E-memory | E-width | E-costs 
 ----+-----------------------------------------+--------+----------+---------+---------
   1 | ->  Streaming (type: GATHER)            |     20 |          |      24 | 33.75   
   2 |    ->  Hash Left Join (3, 5)            |     20 | 1MB      |      24 | 27.75   
   3 |       ->  Streaming(type: REDISTRIBUTE) |     20 | 2MB      |      12 | 14.31   
   4 |          ->  Seq Scan on t1             |     20 | 1MB      |      12 | 13.13   
   5 |       ->  Hash                          |     21 | 16MB     |      12 | 13.13   
   6 |          ->  Seq Scan on t2             |     20 | 1MB      |      12 | 13.13   
 
 Predicate Information (identified by plan id)
 ---------------------------------------------
   2 --Hash Left Join (3, 5)
         Hash Cond: (t1.b = t2.a)

3. join条件不包含分布列

同理，该场景下，由于join条件没有包含分布键，因此为了满足核心思想，需要对两表都做redistribute，执行计划如下：

postgres=# explain select * from t1 left join t2 on t1.b = t2.b;
                                        QUERY PLAN                                         
-------------------------------------------------------------------------------------------
  id |                 operation                  | E-rows | E-memory | E-width | E-costs 
 ----+--------------------------------------------+--------+----------+---------+---------
   1 | ->  Streaming (type: GATHER)               |     20 |          |      24 | 34.93   
   2 |    ->  Hash Left Join (3, 5)               |     20 | 1MB      |      24 | 28.93   
   3 |       ->  Streaming(type: REDISTRIBUTE)    |     20 | 2MB      |      12 | 14.31   
   4 |          ->  Seq Scan on t1                |     20 | 1MB      |      12 | 13.13   
   5 |       ->  Hash                             |     21 | 16MB     |      12 | 14.31   
   6 |          ->  Streaming(type: REDISTRIBUTE) |     20 | 2MB      |      12 | 14.31   
   7 |             ->  Seq Scan on t2             |     20 | 1MB      |      12 | 13.13   
 
 Predicate Information (identified by plan id)
 ---------------------------------------------
   2 --Hash Left Join (3, 5)
         Hash Cond: (t1.b = t2.b)

场景二：复制表和复制表Join

由于复制表在每个DN上有一份全量数据，该场景下相当于单机场景，任选一个DN进行两遍关联即可，不做赘述。

场景三：Hash表和复制表Join

1. Hash表left join 复制表

hash表left join复制表的时候，由于复制表在每个DN上都有一份全量数据，因此不管join条件是否包含hash表的分布键，都不会产生redistribute算子。

postgres=# explain select * from t1 left join tt1 on t1.b = tt1.b;
                                  QUERY PLAN                                  
------------------------------------------------------------------------------
  id |           operation           | E-rows | E-memory | E-width | E-costs 
 ----+-------------------------------+--------+----------+---------+---------
   1 | ->  Streaming (type: GATHER)  |     20 |          |      24 | 39.66   
   2 |    ->  Hash Right Join (3, 4) |     20 | 1MB      |      24 | 33.66   
   3 |       ->  Seq Scan on tt1     |     40 | 1MB      |      12 | 20.20   
   4 |       ->  Hash                |     21 | 16MB     |      12 | 13.13   
   5 |          ->  Seq Scan on t1   |     20 | 1MB      |      12 | 13.13   
 
 Predicate Information (identified by plan id)
 ---------------------------------------------
   2 --Hash Right Join (3, 4)
         Hash Cond: (tt1.b = t1.b)

2. 复制表left join hash表

这种场景下，如果join列不包含hash表的分布列，是会产生redistribute算子的，可以想想为什么（提示：和left join的补NULL有关）。

Redistribute还是Broadcast?

前面介绍的几种场景中，出现的stream算子都是redistribute，什么时候会出现broadcast算子呢？试想一下场景一中的第三种情况，join两侧都需要redistribute，实际场景中，left join的左侧一般都很大，数据量很多，如果集群规模也很大，这时对左表做redistribute的代价是比较高的，这种情况下，如果右侧的表恰好又比较小，那么可以对右侧表做Broadcast，减少大表重分布的开销，达到如下效果：

右侧的小表做Broadcast后，相当于复制表，可以和左侧的表直接Join，执行计划如下：

postgres=# explain select * from t1 left join t2 on t1.b = t2.b;
                                       QUERY PLAN                                       
----------------------------------------------------------------------------------------
  id |                operation                | E-rows | E-memory | E-width | E-costs 
 ----+-----------------------------------------+--------+----------+---------+---------
   1 | ->  Streaming (type: GATHER)            |    100 |          |      24 | 24.07   
   2 |    ->  Hash Left Join (3, 4)            |    100 | 1MB      |      24 | 18.07   
   3 |       ->  Seq Scan on t1                |    100 | 1MB      |      12 | 2.50    
   4 |       ->  Hash                          |     40 | 16MB     |      12 | 15.18   
   5 |          ->  Streaming(type: BROADCAST) |     40 | 2MB      |      12 | 15.18   
   6 |             ->  Seq Scan on t2          |     20 | 1MB      |      12 | 13.13   
 
 Predicate Information (identified by plan id)
 ---------------------------------------------
   2 --Hash Left Join (3, 4)
         Hash Cond: (t1.b = t2.b)

两表Join场景下减少Stream算子的优化建议

大集群高并发场景下，Stream算子过多可能会导致通信的性能瓶颈，引起性能劣化，因此需要尽可能减少Stream算子，根据上面介绍的几种场景，常用的方法如下：

1. 建表时合理选取分布列

在满足数据均匀分布的基础上，选取经常被用作join条件的列作为分布键，尽量不要使用sequence id等无意义的分布键。是否满足均匀分布可以使用函数table_skewness查看，示例：

postgres=# select * from table_skewness('t1','b');
 seqnum | num |  ratio   
--------+-----+----------
 0      | 100 | 100.000%
(1 row)

结果表明该列倾斜率为100%，不适合做分布列。

2. 合理使用复制表

如果join右侧的维表比较小，且不会频发更新，可以考虑将内表建成复制表。

3. 增加包含分布键的join条件

某些场景下可以适当增加包含分布列的join条件，达到减少stream算子的目的。

想了解GuassDB(DWS)更多信息，欢迎微信搜索“GaussDB DWS”关注微信公众号，和您分享最新最全的PB级数仓黑科技，后台还可获取众多学习资料~