1.背景

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

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

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

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

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

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

2.Impala VS Presto

2.1 Impala

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

Impala Remote模式

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

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

2.2 Presto

presto local模式

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

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

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

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

• DynamicFilter
• DynamicFilterSource

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

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

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

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

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

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

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

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