Apache Doris Join优化原理详解

背景与目的掌握Apache Doris连接优化方法及其实现原理为阅读代码提供了理论基础。

Doris数据分区的不同连接方法取决于对Doris中数据分区方法的透彻理解。所以这里先列出必要的基础知识。

首先，在Doris中，数据是以表格的形式进行逻辑描述的。

在Doris的存储引擎中，用户数据被横向划分为若干个数据片(也称为数据桶)。每个数位板包含几个数据行。各个平板的数据之间没有交集，物理上独立存储。

一个平板电脑只属于一个数据分区。一个分区包含几个平板电脑。因为平板电脑是物理独立存放的，所以可以认为隔板也是物理独立的。平板电脑是数据移动、复制和其他操作的最小物理存储单元。

几块隔板形成一张桌子。分区可以被视为最小的逻辑管理单元。只能对一个分区进行数据的导入和删除。

Doris支持两层数据分区。第一层是Partition，支持范围和列表的划分。第二层是Bucket(Tablet)，只支持哈希分区。您也可以只使用一层分区。当使用一层分区时，仅支持桶分区。

解释下图中桌子、分区和桶(平板电脑)之间的关系:

按照表范围的方式对日期字段进行分区，得到N个分区。每个分区用相同的哈希方法将数据分成M个Bucket(Tablet)。逻辑上，桶1可以包含由N个分区划分的数据，例如平板11、平板21和平板N1

特别注意:

Doris中分区和桶的定义可能与其他一些数据库系统不同。下面是一个特定的表构建语句作为示例:

如果不存在，则创建表example _ db . expandle _ range _ TBL(` user _ id ` large int NOT NULL COMMENT "用户id "，` date` DATE NOT NULL COMMENT "数据注入的日期和时间"，` timestamp ` datetime NOT NULL COMMENT "数据注入的时间戳"，city` VARCHAR(20) COMMENT "用户的城市"，` age` smallcomment "用户的年龄"，Sex` TINYINT COMMENT "用户的性别"，` last _ visit _ DATE ` datetime replace default " 1970-01-01 00 ` cost ` bigint sum default " 0 " comment " total consumption users "，" max _ dwell _ time ` int max default " 0 " comment " maximum stay time users "，" min _ dwell _ time ` int min default " 99999 " comment " user ' s minimum stay time ")engine = OLAP聚合键(` user _ id '，` date '，` timestamp '，` city '，` age '，` sex `)按范围分区(` date `)(分区` p201701 `值小于(" 2017-) 绿色:分区，在本例中，日期字段用于分区。

蓝色突出显示:Bucket，在本例中，user_id字段用作分发列表。

PartitionPartition列可以指定一个或多个列，并且分区列必须是键列。理论上，分区数量没有上限。当不使用partition构建表时，系统将自动生成一个与表名同名的全值分区。该分区对用户不可见，不能删除。创建分区时，不能添加具有重叠范围的分区。

有两种分区方式:

分区模式范围的一般用法通常是按时间分区，方便管理新旧数据。List支持更多类型，分区值是枚举值。只有当数据是目标分区的枚举值之一时，才能命中该分区。

如果在Bucket中使用Partition，分布式语句描述每个分区中数据的分区规则。如果不使用Partition，它描述了整个表的数据分区规则对桶列的选择，这是查询吞吐量和查询并发性之间的权衡:如果选择多个桶列，数据分布会更均匀。如果一个查询条件不包含所有桶列的等价条件(意味着不能做桶裁剪来缩小数据查询的范围)，那么查询会同时触发所有桶扫描，这样查询的吞吐量会增加，单个查询的延迟会降低。该方法适用于高吞吐量、低并发的查询场景。如果只选择了一个或几个桶列，对应的点查询只能触发一次桶扫描(也就是说可以做桶裁剪，缩小数据查询范围)。此时，当多个点查询并发时，这些查询分别触发不同桶扫描的概率较大，且每个查询之间的IO影响较小(尤其是不同桶分布在不同磁盘上时)，因此该方法适用于高并发点查询场景，理论上不存在桶拆分次数上限。

加入模式

总的来说，作为一个分布式MPP数据库，需要将数据加入到洗牌加入的过程中。需要对数据进行分割和调度，以确保最终的连接结果是正确的。举个简单的例子，假设关系S和R被连接，N代表连接计算中涉及的节点数；t代表关系的元组数。

目前多丽丝支持以上四种加入方式。这四种方法的灵活性和适用性由高到低，对数据分布的要求也越来越严格。但是通过减少网络开销，Join计算的性能越来越好。

连接方式的选择是FE生成分布式计划阶段需要考虑的问题之一。在FE的分布式规划中，优先级顺序是(期望性能最好的总是优先):co locate join ->；桶洗牌加入-& gt；广播加入->随机加入.

共存和桶混洗不能兼得。当它们无法使用时，多丽丝会自动尝试广播加入，如果估计的小桌子太大，会自动切换到洗牌加入。

但是，用户可以通过显式提示来强制所需的联接类型，例如:

select * from测试联接

在上面的例子中，Join的等价表达式命中了表A(左边的表)的数据分布列表。Bucket Shuffle Join会根据表A的数据分布信息将表B(右表)的数据发送到表A对应的数据计算节点。

定性分析:

减少了网络和内存开销(不比广播和混洗Join差)，使一类Join查询有更好的性能。特别是当FE可以对左表进行分区切割和桶切割时，不同于Colocate Join，它对表的数据分布是无创的，对用户是透明的。对表的数据分布没有强制要求(不需要在建表语句中显式设置colocate_with属性)，不易造成数据倾斜的问题可以为Join Reorder提供更多可能的优化空间。

Plan RuleBucket Shuffle Join仅适用于连接条件相等的情况。原因类似于共址联接。它们都依赖哈希来计算确定的数据分布。两个表的桶列包含在等效的连接条件中。当左表的桶列是等价的连接条件时，有很大概率会被规划为桶混洗连接。因为不同的数据类型有不同的哈希值计算结果，所以Bucket Shuffle Join要求左表中的Bucket列类型应该与右表中等价的Join列类型一致。否则相应的策划就做不出来。桶洗牌加入只适用于多丽丝的本地OLAP表。对于ODBC，MySQL，ES等外观，当它是左表时无法计划生效。对于分区表，由于每个分区的数据分布规则可能不同，桶洗牌Join只能保证左表在单个分区时生效。所以在SQL的执行中，要尽量使用where条件，使分区裁剪的策略有效。如果左表是并置表，那么每个分区的数据分布规则是确定的，桶混洗Join在并置表上的表现更好。

同位连接可以理解为在数据分布满足一定条件的前提下，减少所有不必要的网络传输开销，实现完全计算本地化以加快查询。同时，由于没有网络传输开销，be节点可以有更高的并发性，从而进一步提高加入性能。

要理解这种算法，首先需要知道两个术语:

协同定位组(CG):一个CG将包含一个或多个表。同一组中的表具有相同的着色组模式，以及相同的数据碎片分布共存组模式(CGS):用于描述CG中的表以及与着色相关的一般模式信息。包括桶列类型、桶号、副本号和桶序列的概念:

根据桶列值散列和桶的数量，一个表的数据最终将落入一个桶中。假设一个表的桶号是8，那么总共有

您可以通过查询计划检查查询是否使用了协同定位连接。同时，计划交换节点被删除，ScanNode将被直接设置为Hash Join节点的子节点。

DESC SELECT * FROM tbl1内连接TB L2 ON(TBL 1 . k2 = TB L2 . k2)；Hash Join节点中会显示-4 +: --cololocate:true/false cololocate Join非常适用于几个表按照相同字段划分为桶，按照固定字段高频连接的场景。这样可以提前将数据存储在同一个桶中，实现本地计算。

运行时过滤器优化Doris在进行哈希连接计算时，在右表中构建一个哈希表，左表流经右表的哈希表得到连接结果。运行时过滤器充分利用哈希表构建阶段对表做一些额外的事情。

在右表生成哈希表时，还会生成一个基于哈希表数据的过滤条件，然后下推到左表的数据扫描节点。这样，Doris可以在运行时过滤数据。

如果左表是大表，右表是小表，那么下推到左表的过滤条件可以在读取数据时提前过滤掉Join层的大部分待过滤数据(如果可以下推到引擎层，也可以使用Doris的键列过滤延迟物化)，从而大大提高Join查询的性能。

运行时过滤器在查询规划期间生成，内置于HashJoinNode中，并在ScanNode中应用。比如T1(行数10w)和T2(行数2k)的Join运算:

| >；HashJoinNode & lt| | | | | 100000 | 2000 | | | | olapscannode olapscannode | ^ ^ | | 100000 | 2000 | T1 |显然，扫描T2的数据要比t1快得多。如果我们在扫描T1之前等待一段时间，在T2将扫描的数据记录交给hashjonode之后，hashjonode会根据T2数据计算出一个过滤条件。例如T2数据的最大值和最小值，或者建立布隆过滤器，然后将该过滤器条件发送给等待T1被扫描的ScanNode。后者应用这个过滤条件，将过滤后的数据交给HashJoinNode，从而减少了探测哈希表的次数和网络开销。该过滤条件为运行时过滤，效果如下:

| >；HashJoinNode & lt| | | | | 6000 | 2000 | | | | olapscannode olapscannode | ^ ^ | | 100000 | 2000 | T2 |如果可以将运行时过滤器下推到存储引擎，在某些情况下，可以使用索引(如联接列是键列，可以使用延迟物化能力)直接减少扫描数据量，从而大大减少扫描时间。效果如下:

| >；HashJoinNode & lt| | | | | 6000 | 2000 | | | | olapscannode olapscannode | ^ ^ | | 6000 | 2000 | T2 |可以看出，与谓词下推和分区裁剪不同，运行时过滤是运行时动态生成的过滤条件，即在查询运行时，解析连接条件确定过滤表达式，并将表达式下推至读取左表的ScanNode，从而减少扫描数据量，进一步减少探针哈希表数量，避免不必要的IO和网络传输由于其运行时有效的特性，它被正式视为自适应查询执行的应用程序。

根据上面的例子可以推断，当场景满足以下条件时，使用运行时滤镜的效果会更好:

左表大，右表小(当右表有附加过滤条件时会更理想)，因为构建运行时过滤器需要承担计算成本，包括一些内存开销，开销直接取决于右表的实际数据量。连接左表和右表的结果很少，这意味着左表中的大部分数据都可以通过运行时过滤器进行过滤。Doris支持三种运行时过滤器:

一个是IN，它很容易理解，将一个hashset下推到数据扫描节点。第二个是BloomFilter，就是从哈希表的数据构造一个BloomFilter，然后把这个BloomFilter下推到查询数据的扫描节点。最后一个是MinMax，是一个范围。右表数据确定范围后，下推到数据扫描节点。工作原理和优缺点总结如下:

运行时类型的工作原理适用场景的优缺点。在IN子查询的方式上，实现是将一个Hashset下推到扫描节点广播Join，开销小，过滤效果明显，速度快。当右表超过一定的数据量，就会失效。目前Doris配置的阈值是1024Min/Max，然后下推到一般开销低的扫描节点，只对数值型有效。对于数值以外的类型，不能使用BloomFilter。通过右边的表构建一个BloomFilter，然后将其下推到扫描节点。具有通用性，适用于各种类型，效果好，配置复杂，计算成本高。当过滤速率较低或左表数据较少时，可能会导致性能下降。一些使用注意事项(比较细节，考虑后面结合代码进一步理解):

打开运行时过滤器后，左表中的ScanNode在扫描数据前，会为分配给自己的每个运行时过滤器等待一段时间，也就是说，如果ScanNode被分配了三个运行时过滤器，最多等待3000ms。

因为构建和合并Runtimefilters需要时间，所以ScanNode会尝试将在等待时间内到达的Runtimefilters推送到存储引擎。如果超过等待时间，ScanNode将使用已经到达的Runtimefilters直接开始扫描数据。

如果运行时过滤器在ScanNode开始扫描之后到达，则ScanNode不会将运行时过滤器下推到存储引擎，而是基于ScanNode上的运行时过滤器过滤从存储引擎扫描的数据，并且不会将运行时过滤器应用于先前扫描的数据，这样获得的中间数据的大小将大于最优解，但可以避免严重恶化。

如果集群很忙，并且集群上有许多资源密集型或耗时的查询，您可以考虑增加等待时间，以避免复杂查询错过优化机会。如果集群负载较轻，并且集群上有许多只需要几秒钟的小查询，那么可以考虑减少等待时间，以避免每个查询增加1s的延迟。

前两个表的运行时过滤器为连接重新排序的优化铺平了道路。再看Join Reorder的优化，逻辑关系就可以理顺了。

Doris目前的Join Reorder算法是基于RBO的，其逻辑描述如下:

试着把大桌子和小桌子连接起来。它生成的中间结果是将条件连接表尽可能小的向前放，也就是说，尝试过滤条件连接表。散列连接比嵌套循环连接具有更高的优先级，因为散列连接本身比嵌套循环连接快得多。可以发现，前两个是朝着让“右表”更小的方向优化的，而最后一个是从算法的性能上考虑的。

联接优化建议联接列应该是相同的简单类型；同类型避免了强制转换操作，而简单类型具有良好的连接计算性能。Join列最好是Key列，因为Key列可以充分利用Doris的延迟物化，减少IO提高性能。最好利用联合定位来连接大型表，这相当于已经完成了预洗牌。在实际操作中，可以直接加入计算，无需洗牌操作，完全避免了大表的洗牌网络开销。使用运行时过滤器，当连接过滤器较高时效果显著。根据三个运行时过滤器的特点，需要判断连接的合理性来选择最合适的涉及多个表的连接。尽量确保& ldquo大左，小右& rdquo原则上来说，HashJoin比NLJ好。必要时，可以通过SQL重写和提示来调整联接顺序REF。

https://www.jb51.net/article/266004.htm

https://www.jb51.net/article/266000.htm

以上是Apache Doris加入优化原理的详细内容。关于Apache Doris Join优化的更多信息，请关注主机频道zhujipindao中的其他相关文章。com！