Runtime Filter 优化技术

本文档介绍了查询性能优化技术 Runtime Filter 的原理及使用方法。

1 什么是 Runtime Filter

RuntimeFilter 是提升 YMatrix 执行引擎性能的关键技术之一，RuntimeFilter 是指在优化器生成物理执行计划后，在执行引擎中估价时，动态构建的过滤器（Filter），区别于优化器预先规划的 Filter。

Join 是 SQL 查询中最复杂的算子之一。
以 Hash Join 为例，执行引擎会首先扫描内表（Inner Table）（通常也是数据量较小的表）建立 Hash Table（哈希表），然后遍历外表（Outer Table），计算其 Hash 值并根据 Join 条件进行匹配。前者称为 build side（构造侧），后者称为 probe side（探查侧）。由于外表一般行数很多，表扫描、数据传输以及计算 Hash 值、匹配过程都比较耗时。如果能对外表先进行过滤，则有机会获得查询加速。 Runtime Filter 在 Join 运行时对数据进行过滤，以减少对外表的数据的扫描以及 shuffle（将分布在不同节点的数据按照一定的规则汇集到一起的过程）等阶段会产生的 I/O，来达到提升查询性能的效果。在启用 Runtime Filter 的情况下，使用 YMatrix 查询TPC-H 测试集的 Q17 语句就比不启用有接近 20 倍的性能提升。

2 怎么用 Runtime Filter

在 YMatrix 中，Runtime Filter 默认会打开。基于代价的优化器会参考执行代价来决定是否启用 Runtime Filter。

注意！
如果想要手动关闭则在会话中做如下设置即可：set mx_enable_runtime_filter=off;。

更多相关参数请见 Runtime Filter 系统配置参数。

3 EXPLAIN 信息解读

Runtime Filter 的计划和执行信息可以通过 EXPLAIN 看到。

注意！
要在 EXPLAIN 计划里显示 Runtime Filter 信息需要加上 VERBOSE 关键字。

3.1 计划信息。

创建 rt_ao_t1 表。

CREATE TABLE rt_ao_t1
(
    c1 int,
    c2 int
)
WITH (appendoptimized=true, orientation=column, compresslevel=1)
DISTRIBUTED BY(c1);

创建 rt_ao_t2 表。

CREATE TABLE rt_ao_t2
(
    c1 int,
    c2 int
)
WITH (appendoptimized=true, orientation=column, compresslevel=1)
DISTRIBUTED BY(c1);

计划1 从下面的计划可以看到两处 Runtime Filter Type 关键字，分别位于 Hash 节点下面和 MxVScan 下面。其中：

local 表示 Runtime Filter 传递是本地的，没有通过 Motion 节点。

initiator 表示 Runtime Filter 的发起方（即 Hash 节点）、target 表示 Runtime Filter 的接收方（即 MxVScan 节点）

EXPLAIN (VERBOSE) SELECT * FROM rt_ao_t1 t1, rt_ao_t2 t2 WHERE  t1.c1 = t2.c1;
                                                                                    QUERY PLAN                                                                                        
-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
Gather Motion 3:1  (slice1; segments: 3)  (cost=73.47..284.43 rows=10000 width=16)
Output: t1.c1, t1.c2, t2.c1, t2.c2
->  Hash Join  (cost=73.47..151.10 rows=3333 width=16)
      Output: t1.c1, t1.c2, t2.c1, t2.c2
      Hash Cond: (t1.c1 = t2.c1)
      ->  Custom Scan (MxVScan) on public.rt_ao_t1 t1  (cost=0.00..31.80 rows=3333 width=8)
            Output: t1.c1, t1.c2
            RuntimeFilterType: local, target
      ->  Hash  (cost=31.80..31.80 rows=3333 width=8)
            Output: t2.c1, t2.c2
            RuntimeFilterType: local, initiator
            ->  Custom Scan (MxVScan) on public.rt_ao_t2 t2  (cost=0.00..31.80 rows=3333 width=8)
                  Output: t2.c1, t2.c2
Optimizer: Postgres query optimizer
Settings: enable_hashjoin=on, enable_mergejoin=off, enable_nestloop=off, mx_enable_runtime_filter=on, mx_interconnect_compress=on, mx_runtime_join_ratio=0, mx_runtime_min_outer_rows=0
(15 rows)

计划 2 这个计划和上一个不同之处在于，Hash Join 操作需要通过 Motion 节点。所以 Runtime Filter 的类型是 global，另外在 Motion 节点上也对 Runtime Filter 做了标记，标识为 broker，意思是 Runtime Filter 要通过这个节点做传输。

EXPLAIN (VERBOSE) SELECT * FROM rt_ao_t1 t1, rt_ao_t2 t2 WHERE  t1.c2 = t2.c2;
                                                                                    QUERY PLAN                                                                                        
-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
Gather Motion 3:1  (slice1; segments: 3)  (cost=140.13..417.77 rows=10000 width=16)
Output: t1.c1, t1.c2, t2.c1, t2.c2
->  Hash Join  (cost=140.13..284.43 rows=3333 width=16)
      Output: t1.c1, t1.c2, t2.c1, t2.c2
      Hash Cond: (t1.c2 = t2.c2)
      ->  Custom Scan (MxVMotion) Redistribute Motion 3:3  (slice2; segments: 3)  (cost=0.00..98.47 rows=3333 width=8)
            Output: t1.c1, t1.c2
            Hash Key: t1.c2
            RuntimeFilterType: global, broker
            ->  Custom Scan (MxVScan) on public.rt_ao_t1 t1  (cost=0.00..31.80 rows=3333 width=8)
                  Output: t1.c1, t1.c2
                  RuntimeFilterType: global, target
      ->  Hash  (cost=98.47..98.47 rows=3333 width=8)
            Output: t2.c1, t2.c2
            RuntimeFilterType: global, initiator
            ->  Custom Scan (MxVMotion) Redistribute Motion 3:3  (slice3; segments: 3)  (cost=0.00..98.47 rows=3333 width=8)
                  Output: t2.c1, t2.c2
                  Hash Key: t2.c2
                  ->  Custom Scan (MxVScan) on public.rt_ao_t2 t2  (cost=0.00..31.80 rows=3333 width=8)
                        Output: t2.c1, t2.c2
Optimizer: Postgres query optimizer
Settings: enable_hashjoin=on, enable_mergejoin=off, enable_nestloop=off, mx_enable_runtime_filter=on, mx_interconnect_compress=on, mx_runtime_join_ratio=0, mx_runtime_min_outer_rows=0
(21 rows)

3.2 执行信息

如果想要获取 Runtime Filter 在执行阶段的信息，则可以通过 EXPLAIN ANALYZE 获得。

其中包含的信息如下：

Filter 的类型（示例中为 BloomFilter）
Column：创建 Filter 的列（示例中为 c1）
keys：Filter 的键值大小（示例中为 3385，单位行）
mem：保存 Filter 需要的内存大小（示例中为 128 KB）
inputrows：过滤器输入行数（示例中为 3385 行）
outputrows：Filter 过滤后的行数（示例中为 3385 行）

batch：Filter 处理的批次数（示例中为 53 次）

EXPLAIN ANALYZE SELECT * FROM rt_ao_t1 t1, rt_ao_t2 t2 WHERE  t1.c1 = t2.c1;
                                                             QUERY PLAN                                                                 
-------------------------------------------------------------------------------------------------------------------------------------------
Gather Motion 3:1  (slice1; segments: 3)  (cost=73.47..284.43 rows=10000 width=16) (actual time=11.124..15.917 rows=10000 loops=1)
->  Hash Join  (cost=73.47..151.10 rows=3333 width=16) (actual time=8.923..13.268 rows=3385 loops=1)
      Hash Cond: (t1.c1 = t2.c1)
      ->  Custom Scan (MxVScan) on rt_ao_t1 t1  (cost=0.00..31.80 rows=3333 width=8) (actual time=1.709..3.520 rows=3385 loops=1)
            RuntimeFilter: BloomFilter, column:c1, keys:3385, mem(KB):128, inputrows:3385, outputrows:3385, batch:53
      ->  Hash  (cost=31.80..31.80 rows=3333 width=8) (actual time=4.813..4.813 rows=3385 loops=1)
            Buckets: 524288  Batches: 1  Memory Usage: 4229kB
            ->  Custom Scan (MxVScan) on rt_ao_t2 t2  (cost=0.00..31.80 rows=3333 width=8) (actual time=1.933..2.777 rows=3385 loops=1)
Planning Time: 3.398 ms
(slice0)    Executor memory: 94K bytes.
(slice1)    Executor memory: 9367K bytes avg x 3 workers, 9379K bytes max (seg1).  Work_mem: 4229K bytes max.
Memory used:  128000kB
Optimizer: Postgres query optimizer
Execution Time: 17.945 ms
(14 rows)

4 什么样的查询会使用 RuntimeFilter？

Runtime Filter 是针对 Hash Join 做得优化，所以只要用到 Hash Join 的查询都会尝试使用 Runtime Filter。但最终是否使用，要在计划和执行阶段根据优化器计算出的代价来决定。
Runtime Filter 在查询计划的计划阶段和执行阶段都会进行干预，计划阶段如果发现做连接的内表和外表数据量不满足相应规则，则取消生成；执行阶段如果发现预估数据量和实际数据量差别较大，则也会取消执行。

5 如何知道查询是否使用了 RuntimeFilter？

使用 EXPLAIN VERBOSE 来获取查询计划，如果计划中包含 Runtime Filter 关键字，则说明将会在执行阶段使用 Runtime Filter。