Runtime Filter Optimization Technology

This document introduces the principles and usage of Runtime Filter, a query performance optimization technology.

1 What is Runtime Filter

RuntimeFilter is one of the key technologies to improve the performance of YMatrix execution engine. RuntimeFilter refers to a filter (Filter) that is dynamically built when valuating in the execution engine after the optimizer generates a physical execution plan, which is different from the filter pre-planned by the optimizer.

Join is one of the most complex operators in SQL queries.
Taking Hash Join as an example, the execution engine will first scan the inner table (usually a table with smaller data volume) to create a Hash Table, then iterate over the Outer Table, calculate its Hash value and match it according to the Join conditions. The former is called the build side (construction side), and the latter is called the probe side (probe side). Since the appearance generally has many rows, table scanning, data transmission, calculation of hash values, and matching processes are time-consuming. If the appearance can be filtered first, there is a chance to get query acceleration. Runtime Filter filters data during Join runtime to reduce the I/O generated in stages such as scanning of external data and shuffle (the process of bringing together data distributed on different nodes according to certain rules), to achieve the effect of improving query performance. With Runtime Filter enabled, using YMatrix to query the Q17 statement for the TPC-H test set is nearly 20 times better than not enabled.

2 How to use Runtime Filter

In YMatrix, Runtime Filter is turned on by default. The cost-based optimizer refers to the execution cost to decide whether to enable Runtime Filter.

Notes!
If you want to close manually, you can set the following settings in the session: set mx_enable_runtime_filter=off;.

For more related parameters, see Runtime Filter System Configuration Parameters.

3 EXPLAIN Information Interpretation

Runtime Filter's planning and execution information can be seen through EXPLAIN.

Notes!
To display Runtime Filter information in the EXPLAIN plan, you need to add the VERBOSE keyword.

3.1 Plan information.

Create the rt_ao_t1 table.

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

Create the rt_ao_t2 table.

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

Plan 1 From the plan below, you can see two Runtime Filter Type keywords, located under the Hash node and under MxVScan. in:

local means that the Runtime Filter pass is local and does not pass the Motion node.

initiator represents the initiator of the Runtime Filter (i.e., the Hash node), and target represents the receiver of the Runtime Filter (i.e., the MxVScan node)

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)

Plan 2 The difference between this plan and the previous one is that the Hash Join operation needs to pass the Motion node. Therefore, the type of Runtime Filter is global. In addition, the Runtime Filter is also marked on the Motion node, marked as broker, which means that the Runtime Filter needs to be transmitted through this node.

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 Execution Information

If you want to obtain information about Runtime Filter during the execution phase, you can obtain it through EXPLAIN ANALYZE.

The information contained in it is as follows:

Type of Filter (BloomFilter in the example)
Column: Create the column of the Filter (c1 in the example)
keys: Filter's key value size (3385 in the example, unit row)
mem: Save the memory size required for Filter (128 KB in the example)
inputrows: filter input number of rows (3385 rows in the example)
outputrows: Filter filtered row count (3385 rows in the example)

batch: The number of batches processed by Filter (53 times in the example)

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 What kind of query will use RuntimeFilter?

Runtime Filter is optimized for Hash Join, so as long as you use Hash Join query, you will try to use Runtime Filter. However, whether to use it in the end depends on the cost calculated by the optimizer during the planning and execution stages.
Runtime Filter will intervene in the planning stage and execution stage of the query plan. If it is found that the amount of internal and external data that are connected does not meet the corresponding rules, the generation will be cancelled; if it is found that the amount of estimated data and actual data is large, the execution will be cancelled during the execution stage.

5 How to know if the query uses RuntimeFilter?

Use EXPLAIN VERBOSE to get the query plan. If the Runtime Filter keyword is included in the plan, the Runtime Filter will be used during the execution phase.