Bucket Shuffle Join

功能简介

Bucket Shuffle Join旨在为某些Join查询提供基于本地的优化，减少数据在节点间的传输耗时来加速查询。Bucket Shuffle Join的设计、实现和效果的详细信息，请参见ISSUE 4394。

名称解释

• 左表：Join查询时左边的表，进行Probe操作，可被Join Reorder调整顺序。

• 右表：Join查询时右边的表，进行Build操作，可被Join Reorder调整顺序。

工作原理

SelectDB支持的常规分布式Join方式包括Shuffle Join和Broadcast Join。这两种Join都会导致相当大的网络开销。

例如，当前存在A表与B表的Join查询，它的Join方式为Hash Join，不同Join类型的开销如下：

• Broadcast Join：如果根据数据分布，查询规划出A表有3个执行的HashJoinNode，那么需要将B表全量的数据发送到这3个HashJoinNode。这次查询的网络开销是3倍的B表数据量，内存开销也是3倍的B表数据量。

• Shuffle Join：Shuffle Join会根据哈希计算，将A、B两张表的数据分散到集群的节点之中，所以这次操作的网络开销为A表数据量+B表数据量，内存开销为B表数据量。

FE中保存了SelectDB每个表的数据分布信息。如果Join语句命中了表的数据分布列，则应该使用数据分布信息来减少Join语句的网络与内存开销，这就是Bucket Shuffle Join的思路来源。

上图展示了Bucket Shuffle Join的工作原理。在SQL语句中，A表与B表进行了Join操作，并且Join的等值表达式与A表的数据分布列相吻合。Bucket Shuffle Join会根据A表的数据分布信息，将B表的数据发送到对应的A表的数据存储计算节点。Bucket Shuffle Join开销如下：

• 网络开销：B表数据量。小于常规Join方式的网络开销，其中Broadcast Join为3倍的B表数据量，Shuffle Join为A表数据量+B表数据量

• 内存开销：B表数据量。小于等于常规Join方式的内存开销，其中Broadcast Join为3倍的B表数据量，Shuffle Join为B表数据量。

相比于Broadcast Join与Shuffle Join， Bucket Shuffle Join有着较为明显的性能优势，可以减少数据在节点间的传输耗时和Join时的内存开销。相对于SelectDB原有的Join方式，Bucket Shuffle Join有如下优点：

• Bucket Shuffle Join降低了网络与内存开销，使一些Join查询具有了更好的性能。尤其是当FE能够执行左表的分区裁剪与桶裁剪时。

• 与Colocate Join不同，Bucket Shuffle Join对于表的数据分布方式并没有侵入性，这对于您来说是透明的。对于表的数据分布没有强制性的要求，不容易导致数据倾斜问题。

• 可以为Join Reorder提供更多可能的优化方向。

使用方式

设置Session变量

将Session变量enable_bucket_shuffle_join设置为true，则FE在进行查询规划时就会默认将能够转换为Bucket Shuffle Join的查询自动规划为Bucket Shuffle Join。

set enable_bucket_shuffle_join = true;

在FE进行分布式查询规划时，优先选择的顺序为Colocate Join > Bucket Shuffle Join > Broadcast Join > Shuffle Join。但是如果您显式hint了Join的类型，则上述的选择优先顺序不生效，优先选择hint方式的Join类型。例如：

SELECT * FROM test JOIN [shuffle] baseall ON test.k1 = baseall.k1;

查看Join的类型

通过explain命令来查看Join是否为Bucket Shuffle Join：

|   2:HASH JOIN                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                      |
|   |  join op: INNER JOIN (BUCKET_SHUFFLE)                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                          |
|   |  hash predicates:                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                              |
|   |  colocate: false, reason: table not in the same group                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                          |
|   |  equal join conjunct: `test`.`k1` = `baseall`.`k1`                                         

在Join类型之中会指明使用的Join方式为：BUCKET_SHUFFLE。

Bucket Shuffle Join的规划规则

在绝大多数情况下，您只需将Session变量的开关设为true，就能透明地体验Bucket Shuffle Join所带来的性能提升。然而，如果了解Bucket Shuffle Join的规划规则，您就能编写出更高效的SQL语句。

• Bucket Shuffle Join只生效于Join条件为等值的场景，原因与Colocation Join类似，都依赖hash计算来确定数据分布。

• 在等值Join条件中包含两张表的分桶列，当左表的分桶列为等值的Join条件时，它有很大概率会被规划为Bucket Shuffle Join。

• 由于不同数据类型的hash值计算结果不同，因此，Bucket Shuffle Join要求左表分桶列的类型与右表等值Join列的类型保持一致，否则无法进行对应的规划。

• Bucket Shuffle Join只作用于SelectDB原生的OLAP表。对于ODBC，MySQL，ES等外表，当其作为左表时无法进行有效规划。

• 对于分区表而言，由于每个分区的数据分布规则可能不同，因此只有在左表为单分区时，Bucket Shuffle Join才能生效。所以在SQL执行之中，需要尽量使用where条件，以确保分区裁剪的策略能够生效。

• 如果左表为Colocate的表，那么它每个分区的数据分布规则是确定的。在Colocate表上，Bucket Shuffle Join的表现更好。