在关系型数据库的查询中 Join 是一个十分常见的操作，经过将几个表关联起来，用户能够在遵守数据库设计范式的前提下高效取得信息。在剖析类查询中，大表之间（或大表与小表）的 Join 一般运用 Hash Join 完结，这一般也是查询的功能瓶颈之一，因而如何优化 Join 的查询功能也是核算引擎的要点。

Runtime Filter 是在数据库中广泛运用的一种优化技能，其根本原理是经过在 JOIN 的 probe 端（具体来说是经过削减 probe 端的 scan 数据量）提早过滤掉那些不会命中 Join 的输入数据来大幅削减 Join 中的数据传输和核算，然后削减全体的履行时刻。例如关于下面这条句子的原始履行方案如下，其中 sales 是一个现实表， items 是一个维度表：

SELECT * FROM sales JOIN items ON sales.item_id = items.id WHERE items.price > 100

如上图左半部分所示，在进行 join 运算的时候不只需求把全量的 sales 数据传输到 join 算子里去，而且每一行 sales 数据都需求进行 join 运算（包括算哈希值、比较运算等）。这里假如 items.price > 100 的挑选率比较高，比方达到 50%，那么 sales 表中的大部分数据是肯定不会被 join 上，假如提早进行过滤掉，能够削减数据的传输和核算的开支。

上图的右半部分则是加入了 runtime filter 之后的履行方案，从图中能够看到在进行 join 的 build 端拉取数据的进程中新增了一个 RuntimeFilterBuilder算子，这个算子的作用便是在运转的进程中收集build端的信息形成runtime filter，并且发送到probe端的scan节点中去，让probe端的节点能够在scan就削减输入的数据，然后完结功能的提高。

在当时的大多数体系中runtime filter所需求的算子都是在优化器的CBO阶段之后插入进物理履行方案的，运用的是一种根据规矩的优化方法。可是在[3]中指出假如将runtime filter对履行方案所带来的影响在CBO阶段纳入考虑，则能更进一步地优化履行方案。如下图是一个例子：

在这个例子中：

• 图(a)是一个原始的查询，需求对k，mk和t三个表进行join。

• 图(b)是在不考虑runtime filter的情况下进行CBO得到的物理履行方案。

• 图(c)是在(b)的基础上经过根据规矩的方法将runtime filter加入到物理履行方案中去。

• 图(d)则是将runtime filter放在CBO阶段中得到的物理履行方案，咱们能够看到图(d)得到的最优的物理履行方案的终究cost小于图(c)得到的方案。

可是假如直接将runtime filter加入到CBO中去，则会引起优化器的查找空间的指数级增加。这是因为现有的优化器的CBO阶段大多根据动态规划的算法，假如将runtime filter放入CBO中，则子方案的最优解依赖于查询方案中父节点下推的filter的组合和runtime filter应用到的表的方法，这种组合将会引起查找空间的爆炸。[3]证明了关于星型查询和雪花查询（即经过主键和外键将维度表和现实表关联起来进行join的查询），某些join顺序在加入runtime filter之后是等价的，然后确保了优化器在CBO阶段查找空间的线性增加。

PolarDB-X作为一个HTAP数据库，在满足高功能的oltp场景的一起，也能完结对海量数据的高功能的剖析场景。为满足客户大数据剖析的需求，咱们也在自研的MPP引擎中完结了Runtime Filter。其根本原理与上述根本相同，可是咱们针对分布式数据库的场景也做了一些专门的优化。

在PolarDB-X中咱们挑选运用bloom filter来过滤咱们的数据。bloom filter有着诸多的长处：

• **类型无关： **这一特性降低了咱们处理多品种型的完结复杂度

• **空间复杂度低： **能够提高传输效率和内存开支

• **时刻复杂度低： **这一时刻复杂度既包括生成bloom filter的开支，也指检查是否存在的时刻开支，较低的时刻复杂度确保了不会引进过多的开支

当然在其他的体系中也会包括一些其他品种的过滤器，比方在Spark SQL中假如碰到过滤的是分区列且build端的数据较小，则会挑选运用全量的输入数据进行动态分区的取舍；而假如查询的数据格局是parquet或许orc这样的带索引的格局，则会生成min/max这样简略的过滤器来过滤。但这些过滤器大都针对特定场景，不够通用。

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。

PolarDB-X中的MPP引擎是一个为交互式剖析而生的分布式的核算引擎，与Spark、Flink等不同的当地在于选用push的履行模型。这个模型的优点在于中间数据不必落盘，极大地减小了核算进程中等候的推迟，但也增加了Runtime Filter这一特性开发的复杂度。与大部分的开源核算引擎不同，PolarDB-X中的Runtime Filter不只仅支撑broadcast join，也同样支撑其他各种分布式 join算法。咱们依然运用上面的一个SQL句子举例子：

SELECT * FROM sales JOIN items ON sales.item_id = items.id WHERE items.price > 100

在敞开了runtime filter之后的物理履行逻辑如下所示：

如图所示，build端会将生成的bloom filter发送到coordinator，coordinator在等候各个partition的bloom filter都发送完结之后进行一次merge操作，将兼并好的bloom filter发送到FilterExec算子中去，然后完结过滤作用。经过coordinator兼并之后的bloom filter的巨细与单个的partition的bloom filter的巨细一样大，但为每个probe端只传输一次，极大地削减了数据的传输。一起FilterExec在等候bloom filter的进程中并不会阻塞住，而是经过异步的方法接收bloom filter，然后尽量削减 bloom filter生成给推迟带来的影响。

为了进一步削减数据的传输，咱们经过完结udf的方法将bloom filter下推到DN层，在DN端进行数据的过滤，然后大幅削减网络的开支。如下图所示，PolarDB-X会将bloom filter进一步下推至DN，削减了从DN拉取的数据量，然后削减了网络传输和数据解析的开支。

咱们对比了Runtime Filter在 TPCH 100G的数据集上的作用，其结果如下所示：

咱们能够看到关于耗时较长的大查询，如Q9和Q21咱们都取得了2～3倍的功能提高，而关于其他中型的查询也有1倍的功能提高，总体的功能提高在20%左右。

1. Bloom filter

2. Dynamic Filtering in Trino：trino.io/blog/2019/0…

3. Bitvector-aware Query Optimization for Decision Support Queries, SIGMOD 2020： dl.acm.org/doi/pdf/10.…

4. Query Evaluation Techinques for Large Databases：infolab.stanford.edu/\~hyunjung/…