全部產品
Search

搜尋本產品

    ApsaraDB for SelectDB：Runtime Filter

    文件中心

    ApsaraDB for SelectDB：Runtime Filter

    更新時間：Jul 06, 2024

    本文介紹ApsaraDB for SelectDB提供的Runtime Filter的使用方式和注意事項，作為您進行Join最佳化的參考。

    概述

    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的下推策略，包括OFF、LOCAL、GLOBAL三種策略，預設設定為GLOBAL策略。

    runtime_filter_type

    指定使用的Runtime Filter類型。大多數情況下只需要調整這一個選項，其他選項保持預設即可。

    包括Bloom Filter、MinMax Filter、IN Predicate、IN Or Bloom Filter、Bitmap Filter，預設會使用IN Or Bloom Filter。部分情況下同時使用Bloom Filter、MinMax Filter、IN Predicate時效能更高。

    runtime_filter_wait_time_ms

    左表的ScanNode等待每個Runtime Filter的時間，單位為ms，預設為1000。

    runtime_filters_max_num

    每個查詢可應用的Runtime Filter中Bloom Filter的最大數量，預設10。

    runtime_bloom_filter_min_size

    Runtime Filter中Bloom Filter的最小長度，預設1048576（1 MiB）。

    runtime_bloom_filter_max_size

    Runtime Filter中Bloom Filter的最大長度，預設16777216（16 MiB）。

    runtime_bloom_filter_size

    Runtime Filter中Bloom Filter的預設長度，預設2097152（2 MiB）。

    runtime_filter_max_in_num

    如果join右表資料行數大於這個值，將不產生IN Predicate，預設1024。

    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類型的具體含義如下表。

    參數名稱

    參數說明

    IN Predicate

    根據Join on Clause中Key列在右表上的所有值構建IN Predicate，使用構建的IN Predicate在左表上過濾，相比Bloom Filter構建和應用的開銷更低，在右表資料量較少時效能更高。

    • IN Predicate已實現合并方法。

    • 當同時指定IN Predicate和其他Filter，並且in的過濾數值沒達到runtime_filter_max_in_num時，會嘗試把其他filter去除掉。原因在於IN Predicate是精確的過濾條件，即使沒有其他filter也可以高效過濾。目前僅在Runtime filter的生產者和消費者處於同一個Fragment時才會有去除非In Filter的邏輯。

    Bloom Filter

    有一定的誤判率，導致過濾的資料比預期少一點，但不會導致最終結果不準確，在大部分情況下Bloom Filter都可以提升效能或對效能沒有顯著影響，在少部分情況下會導致效能降低。

    • Bloom Filter構建和應用的開銷較高，所以當過濾率較低時，或者左表資料量較少時，Bloom Filter可能會導致效能降低。

    • 目前只有左表的Key列應用Bloom Filter才能下推到儲存引擎，並且測試結果顯示Bloom Filter不下推到儲存引擎時往往會導致效能降低。

    • 目前Bloom Filter僅在ScanNode上使用運算式過濾時有短路（short-circuit）邏輯，即當假陽性率過高時，不繼續使用Bloom Filter，但當Bloom Filter下推到儲存引擎後沒有短路邏輯，所以當過濾率較低時可能導致效能降低。

    MinMax Filter

    包含最大值和最小值，從而過濾小於最小值和大於最大值的資料，MinMax Filter的過濾效果與Join on Clause中Key列的類型和左右表資料分布有關。

    • 當Join on Clause中Key列的類型為int/bigint/double等時，極端情況下，如果左右表的最大最小值相同則沒有效果，反之右表最大值小於左表最小值，或右表最小值大於左表最大值，則效果最好。

    • 當Join on Clause中Key列的類型為varchar等時，應用MinMax Filter往往會導致效能降低。

    IN or Bloom Filter

    根據右表在執行過程中的真實行數，由系統自動判斷使用IN Predicate還是 Bloom Filter。

    • 預設在右表資料行數少於102400時會使用IN Predicate（可通過session變數中的runtime_filter_max_in_num調整），否則使用Bloom Filter。

    Bitmap Filter

    • 當前僅當In Subquery操作中的子查詢返回Bitmap列時會使用Bitmap Filter。

    • 當前僅在向量化引擎中支援Bitmap 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。
    說明

    僅在對大表間Join的某些長耗時查詢進行調優時，才需要調整此查詢選項。

    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可能會導致結果缺少一些行。