问题背景

列存数据库大多采用无序的数据组织形式，无法像传统行存借助索引精确定位到特定的数据，因此很多查询操作都依赖全表扫描。Data skipping / pruning是减少数据访问的常用手段，也被很多列存系统采用来优化扫描操作。pruning 的实现依赖于两点：1）query中的过滤条件，例如select col from table where col = constant中的col = constant；2）存储层的pruner，通常是数据块的统计信息或者更复杂的用于过滤的数据结构。pruning在实际访问数据之前，会预先利用query中的过滤条件访问pruner，识别过滤掉查询无关的数据，减少不必要的数据访问，从而节省I/O和计算，提升查询性能。
优化pruning可以从两个方面入手：1）充分挖掘利用query中的过滤条件；2）提高pruner的过滤精度。这里我们主要考虑第一点。query中最简单的过滤条件就是查询中显式指定的col = constant这种predicate，我们称其为静态的过滤条件，可以直接在优化阶段将其下推到底层扫描算子进行过滤。与之相对的，动态的过滤条件只能在执行期间获得，例如join或者子查询操作。传统的pruning只利用了静态的过滤条件 (Static pruning)，没有利用动态过滤条件进行优化。考虑到分析型负载涉及大量join操作，如果join selectivity较低，实则有很大的过滤空间。因此，我们希望扩大传统pruning的作用范围，利用运行时的信息动态生成过滤条件来做数据过滤（Dynamic pruning），进一步减少扫描中的数据访问。
dynamic pruning 的关键在于如何利用运行时的信息生成动态的过滤条件。业界常用的一种手段是runtime filter。Runtime filter 通常是一种针对join的优化技术，在join运行时根据一侧的数据动态生成filter，然后在另一侧借助filter提前过滤掉那些不会命中join的数据来减少join算子的输入。我们可以进一步将runtime filter下推到扫描算子，然后借助存储层的pruner进行过滤，这样就可以达到dynamic pruning的目的。

我们举个例子说明利用runtime filter和pruner实现dynamic pruning的适用场景。以下面这条SQL为例，典型的fact table join dimension table。

SELECT /*+ QB_NAME(q1) JOIN_FIXED_ORDER(@q1) */
COUNT(*)
FROM item, inventory
WHERE i_item_sk = inv_item_sk
AND i_item_sk / 204000 <= 0.001

利用 join condition i_item_sk = inv_item_sk和 item 侧（build side）运行时的数据生成 runtime filter inventory.inv_item_sk BTW 1 AND 204，并将其下推到底层扫描算子，然后对 invenotry 侧 (probe side) 的数据做过滤，这一方面减少了pack的读取，另一方面减少了后续join的运算。可以看到，借助runtime filter，invenotry这一侧的数据从 399330000 减小到 399330，selectivity为0.001，过滤掉了绝大多数数据。同时借助dynamic pruning过滤掉的pack数目imci_pruner_rejected = 5815，避免读取大量的不相关pack。

目标

目前系统存储层有pack粒度的pruner（min/max, bloom filter, cmap, surf, 默认min/max），支持static pruning来处理查询中显式的predicate。为进一步扩大pruning的作用范围，本文考虑单机场景下，利用runtime filter来对join进行优化，并结合已有的pruner来实现dynamic pruning，避免join中不必要的I/O和计算，从而提升查询性能。

因此核心是如何设计runtime filter来最大化收益，我们从其自身特点出发

• 对runtime的要求：
  • 何时构建
• 对filter的要求
  • filter 的构建和应用代价
  • filter的准确性
  • filter 需要和存储层的pruner适配才能实现dynamic pruning

设计

cost model

cost model 可以帮助我们分析 runtime filter 的代价和收益，指导runtime filter的设计
对于 hash join 操作来讲
: build side, : probe side
: 构建哈希表的代价， : 扫描哈希表的代价，: hashjoin 输出的代价
：HashJoin算子的代价
对于 runtime filter 来讲
: build side 构建runtime filter的代价
: probe side 应用runtime filter的代价
: runtime filter的selectivity，i.e., probe side 经runtime filter过滤之后满足条件的数据占原始数据的比例。取决于runtime filter本身的准确性和join selectivity 。代价估计阶段可以转化为semi join进行估计。
：引入runtime filter后HashJoin算子的代价
不考虑 pruning 的影响，则runtime filter应用前后的join代价为

进一步考虑pruning
: pack size
: tablescan算子中单个pack的扫描代价 (有I/O和全内存两种情况)
: probe侧 pruner的selectivity，i.e., probe 侧经runtime filter pruning之后仍需扫描的pack数目占原始pack数目的比例。取决于runtime filter的selectivity，pruner准确性和数据分布。
：CTableScan算子的代价
：引入runtime filter之后CTableScan算子的代价
受pruning的影响，runtime filter应用前后的扫描代价为

因此 runtime filter 结合dynamic pruning的收益来自两部分 ，要使 尽可能大，这要求

• R 尽可能小，S 尽可能大 (依赖于 join order)
• %22%20aria-hidden%3D%22true%22%3E%0A%20%3Cuse%20xlink%3Ahref%3D%22%23E1-MJMATHI-3BB%22%20x%3D%220%22%20y%3D%220%22%3E%3C%2Fuse%3E%0A%3C%2Fg%3E%0A%3C%2Fsvg%3E#card=math&code=%5Clambda&id=yPTEx)和 尽可能小 (取决于 filter准确性、pruner准确性以及数据的分布)
• 和尽可能小（取决于filter的类型）

构建

由于runtime本身的特点，何时构建filter并不是十分直接的。我们可以将runtime filter的构建分为两个过程：1）生成filter算子；2）填入运行时的数据。在第一个过程中，根据join condition生成filter算子并进行filter下推，这里起到plan中占位的作用；在第二个过程中，执行阶段根据运行时的数据向上一步生成的filter填入内容。拆成两部分的好处是，只有第二个过程必须依赖运行时的数据，而第一个过程只依赖查询语句自身。在不同阶段生成filter算子会有不一样的作用， 区别在于是否在优化阶段考虑runtime filter

• 在优化阶段生成
  • 优点：将runtime filter纳入优化阶段考虑会扩大搜索空间，理论上有可能生成更优的执行计划
  • 缺点：引入了额外的优化代价
• 在优化阶段后生成
  • 优点：优化阶段没有额外开销
  • 缺点：优化器对runtime filter没有感知，可能会生成次优的执行计划

在优化阶段考虑runtime filter会更复杂，是否能生成更优的执行计划比较依赖filter selectivity estimation的准确性。由于额外的优化代价和其收益难以评估，目前实现暂时没有考虑和优化器的关系，而是在最优物理执行计划确定后，直接生成runtime filter。
关于优化阶段考虑runtime filter的相关工作，学术界有一些相关的探索，留在文末讨论。

filter类型

runtime filter 本质上是对 join build侧数据组成的集合的一种描述，我们目前pack级别的pruner也是pack内部数据集合的一种描述，若要支持 dynamic pruning，其实是对这两个集合进行比较，判断是否有交集，如果没有则可以直接过滤。
filter需要综合考虑：

• 构建应用的开销和准确性之间的trade-off
• 能够适配存储层的pruner

我们考虑最常用的三种filter类型：
Range
将build侧数据的min/max用作filter，要求join key为数值类型，配合min/max pruner，可以实现dynamic pruning。优点是代价小并且比较通用。
Set
直接将build侧数据组成的集合用作filter。这要求build侧distinct value较少，不然probe侧过滤代价会比较高。
**Bloom Filter **
假设 build side 构建一个bloom filter bf1，probe side join key 列上有bloom filter类型的pruner bf2。我们希望的是，通过比较bf1和bf2，可以判断能否过滤掉pack。如果要对两个bloom filter直接比较，要求两个bf的映射关系和映射空间相同，即 k （hash function的个数）和 m（filter长度）相同。如果 bf1 & bf2==0 ，则两个集合元素一定没有重合，可以直接过滤，否则可能存在重合，无法过滤。这要求bf1构造的时候根据bf2进行适配，但有两个问题：1）目前实现的bf2 有两种，它们的m不同，这个信息对bf1不可见； 2）bf2的m较大，bf1如果用相同的m构建开销会很大。

综合考虑， 目前系统内实现了最通用且开销最小的range filter。

实现

适用范围

join strategy一般有 NestedLoopJoin, SortMergeJoin, HashJoin三种。runtime filter的目的是借助运行时的数据信息，用一侧的数据来对另一侧进行过滤。HashJoin的实现是待一侧build hash table完成后，另一侧再开始probe hash table进行匹配。如果在hashjoin build hash table的同时生成runtime filter然后对probe侧的数据提前过滤，这种实现是十分直接的。因此runtime filter的作用范围局限在 hashjoin 类型的算子（包括 HashJoin和HashMatch）。此外，由于是build侧生成filter，然后对probe侧进行过滤，所以join 类型局限于 inner join，left outer join, left semi join这几种。

构建流程

**1. generate **
物理执行计划确定后遍历plan生成runtime filter，并将runtime filter尽可能下推到join probe侧的最底层。这个阶段用来生成filter在plan中的占位，同时建立 hashjoin -> filter之间的映射关系，方便在运行时直接利用映射关系将build侧的数据填入probe侧的对应位置。

• 构建runtime filter（默认左侧 build, 右侧 probe），hashjoin中每一对 <left_expr, right_expr> 可以对应一个runtime filter。
  • 目前只支持join key为int类型
  • 目前只支持right_expr中只有单个field的场景。因为如果涉及多表，则无法下推到底层的tablescan算子，暂时不做考虑。
• 将生成的runtime filter下推到hash join probe侧的最底端。
• 为每个hashjoin会维护一个map来记录有效的expr_id -> runtime filter的映射，这样在之后的build阶段可以直接将表达式结果传输给相应的runtime filter。

**2. build **
HashJoin在build hash table的同时，利用left_expr的值生成runtime filter，然后依赖当前hashjoin算子维护的map，赋值给对应的runtime filter（包括 Optimizer::Predicate 和 Expressions::RTExprTree）。
**3. apply **
由于在generate过程中已经将runtime filter下推到CTableScan算子，然后build过程中会依据build side的数据更新runtime filter，因此apply阶段可以直接复用之前predicate的pruning和evaluate的逻辑。

测试

测试分了两部分，micro-benchmark用来评估runtime filter具体的代价和收益，分析其适用场景；标准的benchmark TPC-H和TPC-DS来验证 runtime filter 对性能的影响。
衡量指标：

• filter selectivity
• pruner selectivity
• query运行时间
• runtime filter加入前后的加速比speedup = t / t_rf，其中 t为没有 runtime filter 的运行时间，t_rf 为加入runtime filter 的运行时间

micro-benchmark

TPC-DS 100G的数据，选取 item 和 inventory 两个表（数据量见下表），典型的 dimension table 和 fact table 进行join操作。验证不同runtime filter selectivity下，runtime filter加入前后的加速比。

小表build 大表probe

我们首先选取runtime filter最适合的场景，小表build hashtable，大表probe hashtable。SQL如下，join key item_sk 数据范围 [1, 204000]，通过参数p调整filter的selectivity。

SELECT /*+ QB_NAME(q1) JOIN_FIXED_ORDER(@q1) */
COUNT(*)
FROM item, inventory
WHERE i_item_sk = inv_item_sk
AND i_item_sk / 204000 <= @p

验证了下面这样三个场景，其中pruner为minmax类型，imci_max_dop=1

• I/O + pruning
• 全内存 + pruning
• 全内存无pruning

**I/O + pruning **
下表为不同selectivity下query的执行时间(ms)

下图展示了不同selectivity下runtime filter的加速比。其中total为总运行时常，hashjoin build为build侧时常，hashjoin probe为probe侧时常。

全内存 + pruning
下表为不同selectivity下query的执行时间(ms)

全内存无pruning
下表为不同selectivity下query的执行时间(ms)

大表build 小表probe

为进一步验证极端情况，调换item inventory两表的join顺序，大表build，小表probe。
SQL如下，join key item_sk 数据范围 [1, 204000]，通过参数p调整filter的selectivity。

SELECT
COUNT(*)
FROM item, inventory
WHERE i_item_sk = inv_item_sk
AND inv_item_sk / 204000 <= @p

结论

• runtime filter 适合join左右两表数据量相差大，小表build，大表probe，且小表对大表的过滤率高的场景
• runtime filter 引入的开销不大，无法加速的查询不会引起明显的性能回退
• runtime filter 结合 pruner 可以进一步提高加速比，在selectivity低的场景下尤为明显

TPC-H

TPC-H的22条query中

• Q1和Q6只涉及单表，不包含HashJoin
• 其余query均包含HashJoin/HashMatch
  • 其中 Q5 Q7 Q8 Q9调整过顺序后的版本为 better5，better7，better8，better9也进行了测试

uniform, scale=100

对于标准的TPC-H数据集（uniform random distribution），runtime filter几乎没有效果，selectivity基本都在0.99左右。下表是scale=100时，所有query的运行时间总和。

|

zipf=2，scale=100

为验证有数据倾斜的场景，调整TPC-H数据集分布（zipf distribution, zipf value可以取任意正实数，通常取值在[0-4]之间，值越大倾斜程度越高），实验选取了zipf value=2。
下图展示了不同query间开runtime filter之后的加速比和runtime filter selectivity之间的关系，其中query按照加速比进行了排序。

可以看到，对于大部分query，都遵循selectivity越低，加速比越高的规律。
但也有一些例外，诸如 Q2, Q7, Q9, Q16, Q17，这些query的selectivity都在0.5以下，但加速比都不明显。原因是，目前的unblocking且不支持join reorder的执行方式下，这些query的plan都出现了大表build，小表probe的情况，这使得查询耗时主要集中在HashJoin的build阶段，因而对于probe阶段的优化在整体query中并不明显。反观调整过顺序的better5，better7，better8，better9，其中better7和better5 selectivity低都有比较好的加速比，better8和better9的过滤比例虽然不高，但也有提升。

TPC-DS

为进一步衡量 runtime filter 的效果，验证其在TPC-DS benchmark下的性能，scale=100。
下图展示了不同query间开runtime filter之后的加速比和runtime filter selectivity之间的关系，其中query按照加速比进行了排序。

加速有限的原因分析：

• hashjoin在query中的时间占比可能不是最主要的，瓶颈在其他算子
• selectivity 较高
• selectivity 低，但是目前缺失 join reorder的能力，生成的plan可能是大表build，小表probe，导致hashjoin主要的耗时在hash build 阶段，对probe阶段的优化在总体query中的占比小

和优化器的关系

传统的runtime filter只考虑了查询的执行阶段，业界系统的实现也都只是在查询的执行阶段或者是优化的最后阶段生成runtime filter，然后直接在执行阶段应用。可以说在之前的做法中，优化器对runtime filter是没有感知的。
这会导致两个问题：

• 对于runtime filter不适合的场景可能会造成性能回退。对于这个问题，业界目前的解法通常都是基于一些规则来识别规避这种场景（例如在build侧设定一个阈值，当build侧数据量超过该阈值时不生成runtime filter）。
• 优化阶段的cardinality估计可能不准确。因为优化阶段没有考虑runtime filter，但在执行时期使用runtime filter，如果runtime filter selectivity很低，那优化时期估计的cardinality会明显偏高，这会进一步影响优化器最优计划的选择（例如join order）。

相关工作

近年有一些相关的工作将runtime filter纳入优化阶段考虑。《Bitvector-aware Query Optimization for Decision Support Queries》在优化器阶段考虑了runtime filter的加入对于不同join order的影响。《Pushing Data-Induced Predicates Through Joins in Big-Data Clusters》则是直接在优化器阶段借助数据的元信息生成filter并加入执行计划中，从而无需在查询执行阶段动态生成filter，避免了runtime filter在执行期间的构建开销。

Bitvector-aware Query Optimization for Decision Support Queries

大部分系统是在优化的最后阶段生成runtime filter，此时 join order 确定，runtime filter在build侧构建，probe侧应用，相当于优化阶段并没有考虑runtime filter对plan cost的影响。以下面这条SQL查询来解释 runtime filter 对 plan 的影响。

SELECT COUNT(*)
FROM movie_keyword mk, title t, keyword k
WHERE mk.movie_id = t.id AND mk.keyword_id = k.id AND t.title LIKE '%(' AND k.keyword LIKE '%ge%'

上图是对这条SQL生成的不同plan
b) 不考虑filter，优先选取小表build，然后大表probe，得到的最优执行计划
c) 在b的基础上加上runtime filter，相当于HJ1和HJ2生成filter分别作用于mk，mk的cardinality由4.5M下降到113k是两次filter作用的结果
d) 优化时考虑filter的影响。mk过滤一次，来自HJ1，t过滤一次，来自mk。因为HJ1对于mk的过滤效果很好，使得mk用于过滤t要比t来过滤mk有更好的效果。
e) 由于优化阶段没有考虑runtime filter，导致代价计算时join输入的cardinality要比实际运用runtime filter之后的更高
可以发现，优化阶段忽略了runtime filter对join cardinality的影响，得到的plan可能是suboptimal的（b+c）。而在优化阶段考虑runtime filter可能生成更优的执行计划 (d)。

Pushing Data-Induced Predicates Through Joins in Big-Data Clusters

不同于runtime filter，作者提出在优化阶段推导生成 induced predicates (diPs) 来实现dynamic pruning。
参考下图，SQL为select count(*) from lineitem, part where lineitem.partkey = part.partkey and part.size=2。最左边的为原始plan，仅有 predicate part.size=2作用在part表上。假设part表上维护了pruner信息，我们可以在优化阶段借助 predicate 对part表的pack进行过滤，然后根据剩余pack的pruner信息推导生成关于join key的new predicate，再利用join key的关系将其推到lineitem表。这样在优化阶段就有作用在lineitem表上，可以对该表进行pruning。

diP有两大优点：

1. 可能生成更优的执行计划。dip的出发点是尽可能将已有的predicate作用到其他table上，来实现更广泛的过滤。diP的构建不局限于build侧还是probe侧，不受join order的影响，能够覆盖到更多的复杂join语句甚至是非join类型的语句。
2. 消除了查询执行时期构建runtime filter的代价。

Reference

Pushing Data-Induced Predicates Through Joins in Big-Data Clusters, VLDB 2019
Bitvector-aware Query Optimization for Decision Support Queries, SIGMOD 2020

runtime filter在业内系统中的运用
Doris runtime filter: https://doris.apache.org/zh-CN/advanced/join-optimization/runtime-filter.html#%E5%90%8D%E8%AF%8D%E8%A7%A3%E9%87%8A
Databricks dynamic file pruning: https://databricks.com/blog/2020/04/30/faster-sql-queries-on-delta-lake-with-dynamic-file-pruning.html
Presto dynamic filter: https://trino.io/blog/2019/06/30/dynamic-filtering.html
PolarDB-X runtime filter: https://zhuanlan.zhihu.com/p/354754979
StarRocks: https://forum.starrocks.com/t/topic/1530
Impala: https://impala.apache.org/docs/build/html/topics/impala_runtime_filtering.html#runtime_filtering, https://impala.apache.org/docs/build/html/topics/impala_partitioning.html#dynamic_partition_pruning

tpc-h zipf distribution generator
https://www.microsoft.com/en-us/download/details.aspx?id=52430
https://github.com/YSU-Data-Lab/TPC-H-Skew.git