MPP Join RuntimeFilter

一 runtime filter

MPP: maassively parallel processing

RuntimeFIlter: 多用于两表Join 时, 通过减少大表返回行的,减少网络传输、减少数据量、 进而加速Join过程的一种方法

RuntimeFilter: 最关键的实现点 就是Bloom FIlter, 将小表Join Key的Min Max值传递给大表Scan算子,或者Scan 的下一个算子,在讲过滤过的数据,传递给Join Node

Runtime Filter

二 基础知识

Join : 我们将Join 的小表成为Build 表,大表称之为probe

因为我们需要将小表的数据按照Join key , 进行Join Builder, 构建Hash Table(shuffle join下 左表也需要)

Bloom filter: bloom fitler 是一种过滤算法, 它可以快速将不属于当前MinMax区间的大部分数据快速过滤掉,

bloom filter 不关注数据类型

个人理解 runtimefiter 的缺点就是, probe 表需要等bloom filter 构建完成进行扫描,就可以理解为需要build 表扫描完成, 构建min max ,然后才可以开始扫描probe

三 RuntimeFilter 分类

1 Local RuntimeFilter

它其实是在MPP下Runtime Filter 的特殊场景, 即Hash Join 为Broadcast Join 的情况

特点: broadcast join 、build 表数据量非常少、probe表与Join 表在同一个Plan Fragement

LolcalRF 也称为犬粮RF, 是根据小表(Buld表,右表)的全量数据构建的一个RF. 又因为broad cast join ,会讲小表进行广播, 那么每个Executor都可以拥有它,生成RF, 最后将RF下推到Probe节点

2 Global Runtime FIlter

即:Shuffle join 使用, Shuffle join 会将左右表数据进行重新打散 , Global Runtime Filter 也需要按照 hask key 进行拆分和合并, 分发到对应的Plan Fragement Instance 中

所以此时我理解就需要一个Gloabl Runtime Filter Coodinator 在做这份工作

Shuffle Join 将build 和probe 表分发成N份,(常规情况下 是Plan Fragement Instance 数量 ==N), 然后一个Hash Join Instance 来处理其中的一份,如下图 N=3

build 表在构建Hash表, Global Runtime Filter 分发到对应的instance 上, 总体称之为一个完成的GRF, 发送给probe scan ,进行数据的过滤

这里我的理解是, 全量的GRF 发送到每个probe scan 算子

三 Runtime Filter生成的代价估算

Runtime Filter的生成、传输和检查会引入额外的开销，如果不加节制地滥用，不但不会提升性能，反而会导致性能的下降。由于代价估算和实现的复杂性，大多数开源系统中都只支持在broadcast join中实现Runtime Filter，比如Trino（原Presto）中就是这样的。这个做法的好处是实现简单，现有系统的改动较小，但同时也会失去很多优化的机会。

在PolarDB-X中我们将Runtime Filter的生成规则与优化器的统计信息有效地结合，通过多个纬度的数据来决定是否需要生成Runtime Filter：

1. probe端的数据量的大小。如果probe端的数据量过小，即便被过滤很多的数据，其性能提升也无法弥补bloom filter的额外开销，此时我们会放弃生成bloom filter。
2. bloom filter的大小。bloom filter的大小由输入的数量和fpp（错误率）决定，并和输入的数量成正比。当bloom filter太大，不仅会增大网络传输的数据，也会增大内存占用，因此我们将bloom filter的大小限制在一定范围内。
3. 过滤比例。当生成的bloom filter的过滤比例太小时，将其下推到join的probe端不仅不会起到任何的效果，而且精确的过滤比例的计算是一个比较复杂的过程，这里我们使用一个近似的公式来估算过滤性：

。只有当过滤比大于一定阀值时我们才会生成runtime filter。