概述

Runtime Filter為某些Join查詢在運行時動態生成過濾條件，來減少數據的掃描計算，避免不必要的I/O和網絡傳輸，從而加速查詢。它的設計、實現和效果詳情請參見ISSUE 6116。

名詞解釋

• 左表：Join查詢時左邊的表，進行Probe操作。可被Join Reorder調整順序。

• 右表：Join查詢時右邊的表，進行Build操作。可被Join Reorder調整順序。

• Fragment：FE會將具體的SQL語句轉化為對應的執行片段（Fragment），然后下發到分布式集群中的BE節點進行執行。BE上執行對應Fragment，并將結果匯聚返回給FE。

基本原理

Runtime Filter在查詢規劃時動態生成，由HashJoin算子（HashJoinNode）中將Join過程中的右表轉換為過濾條件，下推給數據掃描算子（ScanNode），然后在左表掃描過程中進行裁剪過濾。這種方式大幅降低查詢過程中的數據讀取和計算，提升了查詢性能。

例如當前存在T1表與T2表的Join查詢，它的Join方式為HashJoin，T1是一張事實表，數據行數為1000000，T2是一張維度表，數據行數為200。在以上場景下，常規的HashJoin的實際情況如下。

|          >      HashJoinNode      <
|         |                          |
|         | 1000000                  | 200
|         |                          |
|   OlapScanNode              OlapScanNode
|         ^                          ^   
|         | 1000000                  | 200
|        T1                          T2
|

因此T1表需要掃描大量數據，并進行大量的Hash Join計算。

如果主動將T2將掃描的數據記錄交給HashJoinNode后，HashJoinNode會根據T2的數據計算出一個過濾條件，比如T2數據的最大/最小值，或者構建一個Bloom Filter，接著將這個過濾條件發給等待掃描T1的ScanNode。后者應用這個過濾條件，將過濾后的數據交給HashJoinNode，從而減少Probe Hashtable的次數和網絡開銷。這個過濾條件就是Runtime Filter，效果如下。

|          >      HashJoinNode     <
|         |                         |
|         | 6000                    | 200
|         |                         |
|   OlapScanNode              OlapScanNode
|         ^                         ^   
|         | 1000000                 | 200
|        T1                         T2
|

如果能進一步將過濾條件（Runtime Filter）下推到存儲引擎，則某些情況下可以利用索引進行裁剪數據。大幅減少實際讀取的數據量，從而大大降低掃描耗時，效果如下。

|          >      HashJoinNode     <
|         |                         |
|         | 6000                    | 200
|         |                         |
|   OlapScanNode              OlapScanNode
|         ^                         ^   
|         | 6000                    | 200
|        T1                        T2
|

通過上述分析，發現Runtime Filter和謂詞下推、分區裁剪不同，Runtime Filter是在運行時動態生成的過濾條件，即在查詢運行時解析Join on Clause確定過濾表達式，并將表達式廣播給正在讀取左表的ScanNode，從而減少查詢過程中數據的讀取和計算，大幅提升查詢性能。

Runtime Filter類型

SelectDB提供了三種不同的Runtime Filter類型。

• IN類型：利用HashSet結構實現IN過濾條件，下推到數據掃描節點。IN的優點是過濾效果明顯且快速。缺點方面：首先，它只適用于BroadCast；其次，當它右表超過一定數據量的時候就會失效。當前SelectDB配置的數據量限制為1024，即右表如果大于1024，IN類型的Runtime Filter就直接失效。

• Bloom Filter類型：利用哈希表的數據構造一個Bloom Filter，然后下推到查詢數據的掃描節點。Bloom Filter的特點是通用，適用于各種類型、效果也較好。缺點是它的配置比較復雜且計算較高。

• MinMax類型：通過右表數據確定一個Range范圍后，下推給數據掃描節點。MinMax的優點是開銷比較小。缺點是對于非數值列的效用不大。

適用場景

Runtime Filter主要用于大表Join小表的優化，如果左表的數據量太小，或者右表的數據量太大，則Runtime Filter可能不會取得預期效果。Runtime Filter適用的場景有以下兩個要求。

• 左表大右表小。因為構建Runtime Filter需要承擔計算成本，包括一些內存的開銷。

• 左右表Join出來的結果很少。左右表Join出來的結果很少說明這個Join可以過濾掉左表的絕大部分數據。

使用方式

查詢

SelectDB默認開啟Runtime Filter功能。SelectDB在處理用戶查詢時，會自動根據表、查詢語句情況，生成IN類型或Bloom Filter類型的Runtime Filter，進行查詢優化。

Runtime Filter查詢選項

runtime_filter_mode

用于控制Runtime Filter在查詢執行的最小單元之間傳輸的范圍。

取值范圍：數字（0，1，2）或者相對應的助記符字符串（OFF，LOCAL，GLOBAL）。默認2（GLOBAL）。

• LOCAL：相對保守，構建的Runtime Filter只能在同一個查詢執行的最小單元的同一個Fragment中使用，即Runtime Filter生產者（構建Filter的HashJoinNode）和消費者（使用Runtime Filter的ScanNode）在同一個Fragment，例如Broadcast Join的一般場景。

• GLOBAL：相對激進，除滿足LOCAL策略的場景外，還可以將Runtime Filter合并后通過網絡傳輸到不同執行單元上的不同Fragment中使用，例如Runtime Filter生產者和消費者在不同Fragment，比如Shuffle Join。

通常，GLOBAL策略可以在更廣泛的場景對查詢進行優化。但在有些Shuffle Join中，生成和合并Runtime Filter的開銷超過給查詢帶來的性能優勢，可以更改為LOCAL策略。如果集群中涉及的Join查詢不會因為Runtime Filter而提高性能，您可以將設置更改為OFF，從而完全關閉該功能。

在不同Fragment上構建和應用Runtime Filter時，需要合并Runtime Filter的原因和策略詳情請參見ISSUE 6116(opens new window)。

runtime_filter_type

指定使用的Runtime Filter類型。

取值范圍：數字（1，2，4，8，16）或者相對應的助記符字符串（IN，BLOOM_FILTER，MIN_MAX，IN_OR_BLOOM_FILTER，BITMAP_FILTER）。默認值為8（IN_OR_BLOOM_FILTER）。使用多個時用逗號分隔，注意需要加引號，或者將任意多個類型的數字相加，示例如下。

set runtime_filter_type="BLOOM_FILTER,IN,MIN_MAX";

上述設置等價于如下設置。

set runtime_filter_type=7;

Runtime Filter類型的具體含義如下表。

runtime_filter_wait_time_ms

Runtime Filter的等待耗時。

取值范圍：整數，單位ms，默認為1000。

在開啟Runtime Filter后，左表的ScanNode會為每一個分配給自己的Runtime Filter等待一段時間再掃描數據，即如果ScanNode被分配了3個Runtime Filter，那么它最多會等待3000ms。

因為Runtime Filter的構建和合并均需要時間，ScanNode會嘗試將等待時間內到達的Runtime Filter下推到存儲引擎，如果超過等待時間后，ScanNode會使用已經到達的Runtime Filter直接開始掃描數據。

如果Runtime Filter在ScanNode開始掃描之后到達，則ScanNode不會將該Runtime Filter下推到存儲引擎，而是對已經從存儲引擎掃描上來的數據，在ScanNode上基于該Runtime Filter使用表達式過濾。之前已經掃描的數據則不會應用該Runtime Filter。這樣得到的中間數據規模會大于最優解，但可以避免嚴重的裂化。

如果集群比較繁忙，并且集群上有許多資源密集型或長耗時的查詢，可以增加等待時間，以避免復雜查詢錯過優化機會。如果集群負載較輕，并且集群上有許多只需要幾秒的小查詢，可以減少等待時間，以避免每個查詢增加1s的延遲。

runtime_filters_max_num

每個查詢生成的Runtime Filter中Bloom Filter數量的上限。

取值范圍：整數，默認10。

目前僅對Bloom Filter的數量進行限制，因為相比MinMax Filter和IN Predicate，Bloom Filter構建和應用的代價更高。

如果生成的Bloom Filter超過允許的最大數量，則保留選擇性大的Bloom Filter，選擇性大意味著預期可以過濾更多的行。這個設置可以防止Bloom Filter耗費過多的內存開銷而導致潛在的問題。

選擇性=(HashJoinNode Cardinality / HashJoinNode left child Cardinality)
-- 因為目前FE拿到Cardinality不準，所以這里Bloom Filter計算的選擇性與實際不準，因此最終可能只是隨機保留了部分Bloom Filter。

Bloom Filter長度相關參數

包括runtime_bloom_filter_min_size、runtime_bloom_filter_max_size、runtime_bloom_filter_size，用于確定Runtime Filter使用的Bloom Filter數據結構的大?。ㄒ宰止潪閱挝唬?。

取值范圍：整數。

因為需要保證每個HashJoinNode構建的Bloom Filter長度相同才能合并，所以目前在FE查詢規劃時計算Bloom Filter的長度。

如果能拿到Join右表統計信息中的數據行數（Cardinality），會嘗試根據Cardinality估計Bloom Filter的最佳大小，并四舍五入到最接近的2的冪（以2為底的log值）。如果無法拿到右表的Cardinality，則會使用默認的Bloom Filter長度runtime_bloom_filter_size。runtime_bloom_filter_min_size和runtime_bloom_filter_max_size用于限制最終使用的Bloom Filter長度最小和最大值。

更大的Bloom Filter在處理高基數的輸入集時更有效，但需要消耗更多的內存。例如查詢中需要過濾高基數列（比如含有數百萬個不同的取值），可以增加runtime_bloom_filter_size的值進行一些基準測試，這有助于使Bloom Filter過濾的更加精準，從而獲得預期的性能提升。

Bloom Filter的有效性取決于查詢的數據分布，因此通常僅對一些特定查詢額外調整其Bloom Filter長度，而不是全局修改。一般僅在對大表間join的某些長耗時查詢進行調優時，才需要調整此查詢選項。

查看Query生成的Runtime Filter

explain命令可以顯示的查詢計劃中包括每個Fragment使用的Join on Clause信息，以及Fragment生成和使用Runtime Filter的注釋，從而確認是否將Runtime Filter應用到了期望的Join on Clause上。

• 生成Runtime Filter的Fragment包含的注釋例如runtime filters: filter_id[type] <- table.column。

• 使用Runtime Filter的Fragment包含的注釋例如runtime filters: filter_id[type] -> table.column。

以下示例中的查詢使用了一個ID為RF000的Runtime Filter。

CREATE TABLE test (t1 INT) DISTRIBUTED BY HASH (t1) BUCKETS 2;
INSERT INTO test VALUES (1), (2), (3), (4);

CREATE TABLE test2 (t2 INT) DISTRIBUTED BY HASH (t2) BUCKETS 2;
INSERT INTO test2 VALUES (3), (4), (5);

EXPLAIN SELECT t1 FROM test JOIN test2 where test.t1 = test2.t2;
+-------------------------------------------------------------------+
| Explain String                                                    |
+-------------------------------------------------------------------+
| PLAN FRAGMENT 0                                                   |
|  OUTPUT EXPRS:`t1`                                                |
|                                                                   |
|   4:EXCHANGE                                                      |
|                                                                   |
| PLAN FRAGMENT 1                                                   |
|  OUTPUT EXPRS:                                                    |
|   PARTITION: HASH_PARTITIONED: `default_cluster:ssb`.`test`.`t1`  |
|                                                                   |
|   2:HASH JOIN                                                     |
|   |  join op: INNER JOIN (BUCKET_SHUFFLE)                         |
|   |  equal join conjunct: `test`.`t1` = `test2`.`t2`              |
|   |  runtime filters: RF000[in] <- `test2`.`t2`                   |
|   |                                                               |
|   |----3:EXCHANGE                                                 |
|   |                                                               |
|   0:OlapScanNode                                                  |
|      TABLE: test                                                  |
|      runtime filters: RF000[in] -> `test`.`t1`                    |
|                                                                   |
| PLAN FRAGMENT 2                                                   |
|  OUTPUT EXPRS:                                                    |
|   PARTITION: HASH_PARTITIONED: `default_cluster:ssb`.`test2`.`t2` |
|                                                                   |
|   1:OlapScanNode                                                  |
|      TABLE: test2                                                 |
+-------------------------------------------------------------------+
-- 上面`runtime filters`的行顯示了`PLAN FRAGMENT 1`的`2:HASH JOIN`生成了ID為RF000的IN Predicate，
-- 其中`test2`.`t2`的key values僅在運行時可知，
-- 在`0:OlapScanNode`使用了該IN Predicate用于在讀取`test`.`t1`時過濾不必要的數據。

SELECT t1 FROM test JOIN test2 where test.t1 = test2.t2; 
-- 返回2行結果[3, 4];

-- 通過query的profile（set enable_profile=true;）可以查看查詢內部工作的詳細信息，
-- 包括每個Runtime Filter是否下推、等待耗時、以及OLAP_SCAN_NODE從prepare到接收到Runtime Filter的總時長。
RuntimeFilter:in:
    -  HasPushDownToEngine:  true
    -  AWaitTimeCost:  0ns
    -  EffectTimeCost:  2.76ms

-- 此外，在profile的OLAP_SCAN_NODE中還可以查看Runtime Filter下推后的過濾效果和耗時。
    -  RowsVectorPredFiltered:  9.320008M  (9320008)
    -  VectorPredEvalTime:  364.39ms

Runtime Filter的規劃規則

• 只支持對Join on Clause中的等值條件生成Runtime Filter，不包括Null-safe條件，因為其可能會過濾掉Join左表的null值。

• 不支持將Runtime Filter下推到Left Outer、Full Outer、Anti Join的左表。

• 不支持Src expr或Target expr是常量的場景。

• 不支持Src expr和Target expr相等的場景。

• 不支持Src expr的類型等于HLL或者BITMAP。

• 僅支持將Runtime Filter下推給OlapScanNode。

• 不支持Target expr包含NULL-checking表達式，例如COALESCE/IFNULL/CASE，因為當Outer Join上層其他Join的Join on Clause包含NULL-checking表達式并生成Runtime Filter時，將這個Runtime Filter下推到Outer Join的左表時可能導致結果不正確。

• 不支持Target expr中的列（slot）無法在原始表中找到某個等價列。

• 在以下場景不支持列傳導：

  • Join on Clause包含A.k = B.k and B.k = C.k時，目前C.k只可以下推給B.k，而不可以下推給A.k；

  • Join on Clause包含A.a + B.b = C.c，如果A.a可以列傳導到B.a，即A.a和B.a是等價的列，那么可以用B.a替換A.a，然后可以嘗試將Runtime Filter下推給B（如果A.a和B.a不是等價列，則不能下推給B，因為Target expr必須與唯一一個Join左表綁定）；

• Target expr和Src expr的類型必須相等，因為Bloom Filter基于hash，若類型不等則會嘗試將Target expr的類型轉換為Src expr的類型。

• 不支持PlanNode.Conjuncts生成的Runtime Filter下推，與HashJoinNode的eqJoinConjuncts和otherJoinConjuncts不同，PlanNode.Conjuncts生成的Runtime Filter在測試中發現可能會導致錯誤的結果，例如IN子查詢轉換為Join時，自動生成的Join on Clause將保存在PlanNode.Conjuncts中，此時應用Runtime Filter可能會導致結果缺少一些行。