Bucket Shuffle Join

本文介绍云数据库 SelectDB 版支持通过Bucket Shuffle Join进行查询优化,能够减少数据在节点间的传输耗时和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的思路来源。

image.png

上图展示了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的表现更好。