Join優(yōu)化技術(shù)之Runtime Filter
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ù)。
但是實現(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ò)傳輸就可以直接應用，典型的情況時BROADCAST HASH JOIN的時候，JOIN和左表的HDFSTableScan是在一個Fragment中實現(xiàn)的（在一個線程中），由于每一個節(jié)點上運行的JOIN都會獲取到所有的右表數(shù)據(jù)，因此都能夠build出完整的基于右表數(shù)據(jù)的RF信息，然后直接將這個信息交給左表的Scan算子，不需要經(jīng)過任何的網(wǎng)絡(luò)傳輸。
RuntimeFilter實現(xiàn)層面是采用Bloom Filter，上圖中會每個結(jié)點從HDFS讀取數(shù)據(jù)，然后傳到JoinNode，最后都到了協(xié)調(diào)者，統(tǒng)一merge后進行分發(fā)處理。
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端頁面（使用“廣播”連接時應該是這種情況）。
下圖中紅色箭頭表示發(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
