作者 | 宇毅

前言

近年來，數據庫系統服務的數據量呈指數級增長，同時也面臨處理的業務需求愈發復雜、實時性要求越來越高等挑戰。單機數據庫系統已經逐漸不能滿足現代的數據庫服務要求，因此分布式數據庫/數據倉庫得到了越來越廣泛地運用。

在實時分析(OLAP)領域，分布式數據倉庫可以充分發揮系統的分布式特點，將復雜的OLAP任務分解下發到系統中的所有節點進行計算提升分析性能;分布式數據倉庫也可以比較方便地對系統節點進行擴容，應對用戶業務數據量增加的需求。但是分布式數據倉庫用戶無法避免的一個問題是：隨著數據倉庫集群規模增大，擴容帶來的性價比愈發降低。

造成這種現象的一個原因是，表連接(Join)作為數據庫業務中最廣泛使用的算子之一，在分布式計算中依賴系統節點間的數據交互;當分布式集群規模增大時，節點之間的數據交互代價會明顯增加，這種情況下非常考驗分布式系統的網絡處理能力，并依賴用戶的數據表設計和SQL編寫能力以緩解數據交互壓力。

針對這個問題，業界不同的分布式數據庫系統提出了不同的Join運行時過濾(Runtime Filter)算法。AnalyticDB for PostgreSQL(以下簡稱ADB PG)是一款PB級的MPP架構云原生數據倉庫，同樣也面臨著上述問題的挑戰。本文從ADB PG架構設計的角度出發，探討Runtime Filter在ADB PG中的實現方案，并介紹了基于Bloom Filter的ADB PG Dynamic Join Filter功能技術細節。

ADB PG架構簡介

ADB PG基于開源項目Greenplum構建，在單機PostgreSQL的基礎上進行擴展，將多個PG服務同時啟動在單個或多個服務器上并組成集群，以分布式的形式提供數據庫服務。ADB PG將每一個PG服務稱為一個Segment，并引入了Slice的概念。Slice用于解決分布式系統中的網絡結構，當數據庫涉及到MPP多階段計算時，例如Hash Join左右表的Join Key不滿足相同的Hash分布，那么就需要對Join Key通過網絡傳輸進行重分布，ADB PG將網絡傳輸的前后階段切分為不同的Slices。以下是一個ADB PG集群示意圖。

在這種架構下如何解決大規模集群下表連接Join的性能問題呢?業界解決這個問題的一個方案是引入網絡代理節點，同一機器內的Segment將網絡數據發送至本地代理節點，由代理節點與其它機器上的代理節點進行網絡收發工作以減少網絡擁塞。該方案對ADB PG架構的挑戰較大，且沒有從根本上減少Join的網絡Shuffle開銷。因此為了從Join根源上減少Join計算的數據量，ADB PG設計并實現了Join Runtime Filter方案。

Runtime Filter和Bloom Filter

Runtime FIlter的目的是在Join計算前篩選掉一部分數據，需要一個Filter的實現“載體”。在結合ADB PG的架構設計、存儲層和網絡層的特點后，我們選擇使用Bloom Filter作為Runtime Filter的實現形式。

Bloom Filter是一種概率數據結構，通常被用于判斷一個元素是否屬于一個集合。Bloom Filter的優點是其空間效率非常高，計算性能通常也高;缺點是存在陽性誤判率false positive，但是不存在false negative，即Bloom Filter判斷一個元素是否屬于集合的結果不是單純的true or false，而是"possible true" or "false"。

上圖是一個標準Bloom Filter的計算思路示意圖，其中的0、1為Bloom Filter用于表示集合信息的bit array，即每一位用一個bit存儲。上方x,y,z表示向Bloom Filter中插入的三個元素，分別使用3種hash算法計算hash值后在bit array中置位。而下方為判斷元素w是否屬于集合，由于3個hash值中的某一位沒有在bit array中被置位，可以肯定的是w不屬于集合。

Bloom Filter通常由以下幾個參數描述：

• m --- Bloom Filter bit array的大小m bits
• k --- 使用的hash函數個數k
• p --- 誤判率
• n --- Bloom Filter插入的元素個數

我們省略推導過程，直接將各個參數的關系給出：

當Bloom Filter足夠大時，可以簡化為：

在設計Bloom Filter時，n和m我們可以根據實際計算場景提前確定，上述公式可以視為自變量為k，應變量為p的函數p(k)，此函數通常在k > 0時通常不是單調的(由n:m確定)。因此Bloom Filter在設計時要考慮如何確定hash函數k的個數以獲得最小的誤判率p。根據上式可以計算得到當p為極小值時，對應k的值為：

Bloom Filter的參數設計：

如何將Bloom Filter應用至ADB PG Join過濾優化，我們首先要設計選擇Bloom Filter的參數。對于Bloom Filter插入元素的個數n，可以直接使用執行計劃中獲得的Join右表計劃行數;而為了獲得理想的過濾率，減少誤判率p，ADB PG使用了PG高版本Bloom Filter的思路，設計Bloom FIlter大小Bytes為n的2倍，即總體n:m達到1:16。在這個設計下，可以計算得到最佳的k取值為11，p(k)函數如下圖所示，當k = 11時可以取得最小的p = 0.046%

k = 11意味著對于每一個元素，都需要計算11個hash值以插入到Bloom Filter bit array中，這對于ADB PG是無法接受的，構建Bloom Filter的代價明顯過大。在構建Bloom Filter時，ADB PG會綜合誤判率、hash計算等因素考慮，選擇合適的k值。

在確定構建Bloom Filter的基本原則后，接下來就是工程實現問題。Bloom Filter的工程實現非常簡單高效，通常我們可以直接使用bitset數組來建立Bloom Filter，通過位操作實現Bloom Filter的插入和查找。下圖為向一個Bloom Filter bitset數組中插入元素的計算示意圖。

Dynamic Join Filter in ADB PG

在完成ADB PG Hash Join的Bloom Filter設計后，接來下討論如何將Bloom Filter應用至Join的Runtime Filter中。ADB PG將基于Bloom Filter的Runtime Filter命名為Dynamic Join Filter。

1.Dynamic Join Filter的實現方式

由于ADB PG優化器通常會選擇將右表作為小表，左表作為大表，因此ADB PG將Dynamic Join Filter的設計特點為單向過濾的，即僅用于右表過濾左表，暫不考慮左表過濾右表的形式;同時我們也可以將Dynamic Join Filter靈活應用于Hash Join左表鏈路不同算子的過濾中。

由于Hash Join的形式不同，Dynamic Join Filter的實現形式可以總結為Local Join和MPP Join兩種形式，并根據Runtime Filter是否具有下推算子的能力做進一步區分。

Local Join

Local Join是指左右表的Join Key均滿足相同Hash分布，無需再Shuffle數據。此時Hash、Hash Join和左表Scan處于同一個Slice內部，即同一個進程中，我們可以直接在進程空間內將Bloom Filter傳遞給左表Scan算子過濾輸出。

MPP Join

MPP Join是指左右表的Join Key均不滿足相同Hash分布，需要針對Join Key Shuffle數據。在前文介紹過，ADB PG的Hash Join和Hash算子一定處于同一個Slice內部，因此基于基本原則只需要考慮左表Shuffle的情況，即左表在Hash Join前存在Motion的場景。

MPP Join存在的另一種情況是，左表Motion下不是簡單的Scan，也沒有關聯信息將Join Key的Bloom Filter下推至Scan。那么以減少網絡傳輸數據量為最后準則，將Bloom Filter過濾放在Motion前，減少Motion Sender的數據。

2.Bloom Filter網絡傳輸

Dynamic Join Filter在各個計算節點上建立了一個Local Bloom Filter，每個計算節點需要收集所有其它節點的Bloom Filter，并在本地組成完整的Bloom Filter后才能開始過濾計算。我們將Bloom Filter的收發分為兩種模式：全量傳輸和位傳輸。在發送前我們可以判斷兩種模式的數據量大小，并自適應選擇數據量小的模式。

Bloom Filter全量傳輸

Bloom Filter位傳輸

性能測試

接下來我們對ADB PG Dynamic Join Filter的性能表現測試。測試集群為ADB PG公有云搭建的實例，測試使用TPC-H 1TB測試集(scale = 10000)，測試通過開啟\關閉Dynamic Join Filter功能對比執行性能。下圖展示了TPC-H執行性能有差異的Query測試結果：

可以看到Dynamic Join Filter在Q5、Q8、Q9和Q17上均獲得了較大的性能提升，其中Q17的優化性能最佳，執行時間137s優化至8s。而Q10存在略微的性能回退：10s回退至12s，原因在于Q10的Join Key是完全匹配的，Dynamic Join Filter無法做到動態提前過濾，而優化器未能準確估算代價導致計劃仍然使用了Dynamic Join Filter。此外Q20也因為優化器下推規則的的原因沒有選擇Dynamic Join Filter，實際上經過分析Q20與Q17類似，比較適合使用Dynamic Join Filter。為了解決這些問題，ADB PG優化器相關功能仍在開發迭代中。

總結&未來規劃

Dynamic Join Filter根據ADB PG架構設計、存儲層和網絡層特點，使用Bloom Filter作為Join Runtime Filter的實現形式，在TPC-H測試中取得了明顯的性能提升成果。未來我們將從以下幾個方面做進一步的開發和優化，提升客戶使用體驗：

完善Dynamic Join Filter功能，支持各種模式的Hash Join，并進一步推廣到Merge Sort Join、NestedLoop Join的優化中;

提升優化器的代價估算模型精度，完善優化器下推規則;

Runtime Filter自適應調度。

歡迎訪問云原生數據倉庫ADB PG主頁，了解更多：https://help.aliyun.com/product/35364.html