Join優(yōu)化技術(shù)之Runtime Filter

作者：lightcity 2022-10-30 10:24:43

RuntimeFilter實現(xiàn)層面是采用Bloom Filter，上圖中會每個結(jié)點從HDFS讀取數(shù)據(jù)，然后傳到JoinNode，最后都到了協(xié)調(diào)者，統(tǒng)一merge后進行分發(fā)處理。

1.背景

Runtime Filter又稱為Dynamic Filter，其目的在于通過在join的probe端提前過濾掉那些不會命中join的輸入數(shù)據(jù)來大幅減少join中的數(shù)據(jù)傳輸和計算，從而減少整體的執(zhí)行時間。簡單來說就是利用小表的Join keys基于大表Join keys構(gòu)造過濾器，來減少大表的數(shù)據(jù)讀取。

圖中左邊是正常掃描查詢計劃，右邊是加上Runtime Filter(Dynamic Filter)之后的掃描計劃，可以看到probe端在Join之前(Scan時)提前過濾掉數(shù)據(jù)。

SELECT * from fact_table A JOIN dimension_table B WHERE A.join_key = B.join_key;

但是實現(xiàn)層面的困難在于如何將Runtime Filter(Dynamic Filter) Builder端的數(shù)據(jù)轉(zhuǎn)發(fā)給probe端，因為這些算子可能在不同的節(jié)點上運行。

一般有如下幾種設(shè)計方案：

協(xié)調(diào)器(Remote模式)
Local模式

2.Impala VS Presto

2.1 Impala

Impala采用了這兩種方式、Presto采用了Local模式。

Impala Remote模式

Impala local表示生成的RF不需要通過網(wǎng)絡(luò)傳輸就可以直接應(yīng)用，典型的情況時BROADCAST HASH JOIN的時候，JOIN和左表的HDFSTableScan是在一個Fragment中實現(xiàn)的（在一個線程中），由于每一個節(jié)點上運行的JOIN都會獲取到所有的右表數(shù)據(jù)，因此都能夠build出完整的基于右表數(shù)據(jù)的RF信息，然后直接將這個信息交給左表的Scan算子，不需要經(jīng)過任何的網(wǎng)絡(luò)傳輸。

2.2 Presto

presto local模式

Presto 的Dynamic Filter包含 Partition Pruning（分區(qū)表）以及 Row filtering（非分區(qū)表），依賴于broadcast join，builder端比probe端的表小得多。在這種情況下，probe端掃描和join算子在同一個進程中運行——因此它們之間的消息傳遞變得更加簡單。

presto將生成的Dynamic Filter作為 TupleDomain 公開給Probe端的 PageSourceProvider。

TPC-DS運行結(jié)果顯示DF在部分query上有顯著受益，cost-based optimizer (CBO) 。

Presto的實現(xiàn)原理：

DynamicFilter
DynamicFilterSource

DynamicFilter 代表計劃的一部分，一旦過濾器數(shù)據(jù)準備好，它將在運行時進行實際過濾。

DynamicFilterSource 負責構(gòu)建運行時謂詞數(shù)據(jù)（例如布隆過濾器）并在準備好時將其傳遞給 DynamicFilter。除了構(gòu)建 Bloom 過濾器 DynamicFilterSource 節(jié)點外，它還作為傳遞節(jié)點將接收到的數(shù)據(jù)轉(zhuǎn)發(fā)到 Join 節(jié)點。

代碼實現(xiàn)角度：DynamicFilterSource算子作為一個簡單的“pass-through”管道，同時保存輸入的頁信息。收集的頁面值用于創(chuàng)建RunTime Filter約束（用于內(nèi)部連接中的probe端表掃描）。注意該算子僅支持小表builder端頁面（使用“廣播”連接時應(yīng)該是這種情況）。

下圖中紅色箭頭表示發(fā)送謂詞（例如布隆過濾器）時的通信。這里可以使用標準的 Presto 數(shù)據(jù)通信方式（Pages over Exchanges）將數(shù)據(jù)從 DFS 傳遞到 DF。實現(xiàn)另一個“元數(shù)據(jù)”協(xié)議似乎過于復雜。

如果將此連接公開為直接的父子關(guān)系，因為這將導致計劃不再是樹而是 DAG。這將破壞（或至少使）通過訪問者的當前遍歷方法。無論如何，當在優(yōu)化期間圍繞計劃樹移動一個或另一個時，需要保持 DFS 到 DF 的關(guān)系。

因此，最終實現(xiàn)手段是提供一個DynamicFilterSource算子作為通信管道。

https://varada.io/blog/presto/dynamic-filtering-for-highly-selective-join-optimization/

https://github.com/prestodb/presto/pull/9453/commits

責任編輯：武曉燕來源：光城