全面解讀PostgreSQL和Greenplum的Hash Join

2019年10月15日，Pivotal中國研發中心副總經理兼Greenplum中文社區發起人姚延棟出席了於意大利舉行的PostgreSQL Conference Europe並發表了精彩的演講《How does Hash Join work in PostgreSQL and its derivates》。本文根據演講內容整理而成，供你們學習交流。

今天我將詳細介紹PostgreSQL和Greenplum的Hashjoin。之因此會選擇Hashjoin這個話題，是由於HashJoin是處理OLAP或者是分析型查詢（analytics queries）的重要武器。首先，咱們來看看 PostgreSQL 中的 Hashjoin 實現。

在介紹HashJoin實現以前，首先了解下什麼是 JOIN。根據維基百科（WIKIPedia），JOIN是關係數據庫中組合一個或者多個表中的columns的算子。
而JOIN 有多種類型， SQL 標準中定義了 INNER JOIN、LEFT JOIN、RIGHT JOIN、FULL OUTER JOIN 等四種類型，用集合論裏的操做很是容易理解。 咱們在下圖直觀的解釋了這四種JOIN類型的效用。

此外還有其中JOIN類型，譬如 SEMI JOIN和 ANTI JOIN。 這兩種JOIN不是 SQL 語法，然而一般用來實現某些 SQL 功能，後面會詳細介紹他們。

本文將使用這樣一個例子。 例子裏面包含兩張表：student表和score表，每張表都有幾條記錄。

首先咱們看下圖中JOIN對應的 SQL語句。 經過 explain 能夠查看這6個SQL例子的JOIN類型。 （須要設置 enable_mergejoin, enable_hashagg 爲 OFF，不然優化器可能會選擇其餘查詢計劃）

下圖展現了查詢結果，讓咱們對圖中JOIN的做用有個直觀的認知。

JOIN 有三種經典的實現算法：Nested Loop、Merge JOIN、Hash Join。他們各有優缺點，譬如 Nest loop 一般性能很差，可是適用於任何類型的JOIN；Merge join對預排序的數據性能很是好；HashJoin對大數據量一般性能最好，可是隻能處理equijoin，而不能處理譬如 c1 > c2 這樣的join條件。

Hashjoin 是一個經典算法，它包含2個phases，一個是 build phase，理想狀況下對小表構建 hash table，該表一般也稱爲 inner table；第二個phase爲 probe phase，掃描關聯的另外一張表的內容，並經過hash table 探測是否有匹配的行/元組，該表一般稱爲 outer table。

讓咱們先從 inner join入手。下圖有一個inner join的例子，左邊是其查詢計劃，右邊是圖形化的計劃樹。 其中 inner table 一般也稱爲 right table；outer table 稱爲 left table。

首先是 build phase，在這個階段，對掃描 inner table 的每個tuple，根據 join key 的值計算其hash值，並放到hash table的對應的桶中。掃描完inner table後，也完成了整個hash table 的構建。

第二個階段 probe phase，掃描outer table 的每個元組（tuple），計算該元組的hash值，而後根據計算的 outer table 的哈希值，去hash table中查詢是否有匹配元組，若是有而且知足全部查詢條件，則輸出該元組。依次處理outer table 的每個元組。

如今來下一個 full outer join 的例子，與inner join 不一樣的是，如何處理沒有匹配的tuple。 依然是先掃描 inner table 構建hash table，而後掃描 outer table。 若是join key 匹配且知足全部查詢條件，則輸出該tuple對應的關聯結果。若是hashtable 中沒有匹配的元組，則輸出該tuple，而且使用 null 填充關聯結果中 inner table 對應的column。  當掃描完outer table 後，再次掃描 hash table，找到全部不曾匹配過的 inner tuple，輸出該tuple，並使用 null 填充關聯結果中 outer table 對應的column。

這兒有個比較容易混淆的地方：JOIN 類型的SQL語義和 JOIN的內部實現類型。咱們經過一個例子來看，一樣是 LEFT JOIN的兩個 SQL，內部可使用 Hash Left JOIN 或者 Hash Right Join。 第一個例子是一張大表left join一張小表，它的內部是實現JOIN類型是 Left join；第二個例子是一張小表 left join一張大表，它的內部實現JOIN類型是 right join。 緣由是優化器儘可能選擇小表作內表，在其上構建hashtable。

下面咱們來看看 semi join。Semi join 一般用來實現 EXISTS。它和 inner join相似，不一樣支持是 semi join 只關心有沒有匹配，而不關心有多少元組匹配。

Anti JOIN 則是在沒有元組匹配時才輸出結果，用來實現 NOT EXISTS。

前面的實現看起來很是直觀優雅，卻沒有考慮一個問題：若是inner table 太大不能放到內存中怎麼辦？ 解決的思路很經典，分而治之。 Grace hash join 是經典的解決這一個問題的算法：它把inner table 和outer table 按照關聯鍵分紅多個分區，每一個分區保存到磁盤上，而後對每一個分區應用前面提到的 hash join 算法。每一個分區成爲一個 batch（一次批處理）。 基本思路是根據join key 計算其hash value，然而計算該hash值對應的batchno和bucketno：算法爲：

• bucketno = hashvalue MOD nbuckets

batchno = (hashvalue DIV nbuckets) MOD nbatch

• nbuckets 是buckets的個數，nbatch是batch的個數，二者都是2的冪，這樣能夠經過位運算得到 bucketno和batchno

Hybrid hash join是 grace hash join之上的一個優化：第一個batch沒必要寫入磁盤，能夠避免第一個batch的磁盤io。

hybrid hash join首先對 inner table進行分區/分batch，根據前面計算的算法計算batchno，若是tuple屬於batch 0，則加入內存中的hashtable中，不然則寫入該batch對應的磁盤文件中。batch 0不用寫入磁盤文件中。

而後對 outer table 進行分區/分batch，若是outer table 的tuple屬於 batch 0，則執行前面提到的hashjoin算法：判斷hashtable中是否存在匹配該outer tuple的inner tuple，若是存在且知足全部where條件，則找到了一個匹配，輸出結果，不然繼續下一個tuple。若是outer tuple 不屬於batch 0，則寫入該batch對應的磁盤文件中。

Outer table 掃描結束時， batch 0也處理完了。繼續處理 batch 1：加載 batch 1 的inner table臨時數據到內存中，構建hashtable，而後掃描batch 1 的outer table臨時數據，執行前面跳到的probe 操做。完成batch 1 後，繼續處理 batch 2，直到完成全部的batches。

下面這張圖介紹瞭如何判斷是否須要多個batch：若是inner table 的大小加上 buckets 的開銷小於 work_mem，則使用單個batch；不然須要使用多個batches。

算法的輸入：

• Plan_rows：預估的inner table 的行數
• Plan_width：預估的inner table 的平均行寬
• NTUP_PER_BUCKET：單個bucket的tuples數據，老版本這個數值是10，新的版本是1，假設hash衝突不多，平均一個bucket 含有一個 tuple
• Work_mem：爲hashjoin分配的內存配額

那麼若是 batch 0 仍然太大，內存不足以容納怎麼辦？

辦法是batches個數翻倍，從n變成2n。 這時會從新掃描 batch 0裏面的tuples，根據2n從新計算其所屬的batch，若是仍然屬於batch 0，則保留在內存中，不然從內存中移除，寫入到tuple對應的新batch中。

此時不會移動 Batch 文件中的已有的tuple，當處理該 batch 時會進行處理。

因爲 batch 數目發生了變化，那麼有些batch 裏面的tuple可能會不在屬於當前batch。Hybrid hash join 算法（取模操做）確保，batch 數目翻倍後，tuple 所屬 batch 只會向後，而不會向前。

處理 Batch i 時，若是該batch 的inner tuple太多，佔用空間太大，那麼有可能內存仍是放不下。

這會形成 batch 數目繼續翻番。以下圖所示。

當前batch裏的tuple所屬的batchno也會變化。 舉一個具體例子，假設 Nbatch = 10;  2次翻番後，Nbatch = 40； batch 3 中已有的tuple知足 hashvalue % 10 = 3, 因此batch 3 中的tuple的hashvalue多是  3, 13, 23, 33, 43, 53, … 當 nbatch從10變爲40時，hashvalue % 40 可能的結果是3, 13, 23, 33.

PostgreSQL 的hashjoin實如今經典的 Hybrid hash join之上還作了一些優化，一個重要的優化是對傾斜數據的優化。現實中的數據不少是非正態分佈的數據，譬如寵物，假設地球上每一個人都有一個寵物，那麼養貓或者養狗的人會佔據大多數。

Skew優化的核心思想是儘可能避免磁盤io：在 batch 0階段處理outer table最多見的 （MCV，Most common value） 數據。選擇 outer table 的MCV而不是inner table的 MCV的緣由是優化器一般選擇小表和正態分佈的表作 inner table，這樣outer table會更大，或者更大機率是非正態分佈。

首先是準備 skew hash table，包括三個步驟：

• 肯定 skew hash table 的大小。PostgreSQL 默認分配2% 的內存用戶構建skew hash table，並計算能容納多少 MCV tuples。
• 根據 pg_statistic syscache 數據，得到 outer table 的MCV 統計信息，對每一個mcv，計算其hash值，並放到其對應skew hash bucket 中，如今尚未處理inner table，因此該 bucket 指向 NULL。hash衝突解決方案是線性增加，若是當前slot被佔用了，則佔用下一個。 計算skew buckets 大小的時候，會確保 skew hashtable 足夠稀疏，避免轉一圈也找不到空閒slot。
• 填充skew hash table：掃描inner table 構建main hashtable 時，若是當前tuple 屬於skew hash table（對應的slot不爲空），則加入到skew hashtable 而非main hash table。

掃描outer table 時，若是是 MCV tuple，則使用skew hash table進行處理。 不然按照前面介紹的Hybrid hash join 算法處理。假設使用 skew 優化，50%的 MCVs 在 batch 0階段就處理了，那麼節約了大約 50% 的磁盤io。

這裏不介紹並行 JOIN，主要緣由是PostgreSQL hashjoin 的並行join實現看起來不優雅，引入了大約1倍的代碼量來處理並行hash join。 而 Greenplum 提供一個很是優雅的方案來處理並行hash join，代碼幾乎不須要改動。

接下來，咱們來說講  Greenplum 中的 HashJoin。

首先介紹下Greenplum。 Greenplum 本質上是不少 PostgreSQL 節點組成的集羣，然而單單把不少PostgreSQL節點放在一塊兒，不能給用戶提供一個透明的，知足ACID的邏輯數據庫。

爲此 Greenplum 團隊在分佈式數據存儲、分佈式查詢優化、分佈式執行、存儲器、事務管理、併發控制、集羣管理等領域作了大量工做，覺得用戶提供一個高性能的、線性擴展的、功能齊全的邏輯數據庫。

這是 Greenplum 的一個典型拓撲結構。 Greenplum 是CS 模式，有master和segment節點，每一個節點有本身的存儲、內存和計算單元，之間經過網絡進行通訊，這種架構也稱爲無共享MPP架構。整個架構從磁盤、segments、網絡和master 等各個層次都是高可用的。

Greenplum 有兩個關鍵概念：

• 分佈策略：控制每一個tuple保存在那個segment上，目前支持hash分佈、隨機分佈、複製表，也支持自定義
• Motion：在不一樣的segments之間傳輸數據，有三種方式：Gather，重分佈和廣播

這個例子裏面兩張表的分佈鍵都是學生的id，因此例子 SQL 的join操做能夠在本機執行，而後由master的 Gather motion作一個結果的彙總。這種SQL的執行和PostgreSQL 的查詢執行很是類似。

這個例子裏student表按照學生id分佈數據到不一樣segments，而 score 表按照score 表的id分佈數據到不一樣segments上，這樣同一個學生的score 信息可能分佈在不一樣的節點上，因此上頁圖片的查詢計劃會產生錯誤結果。爲了解決這個問題，查詢計劃中引入了一個 Broadcast motion。這樣 Hashjoin node 的outer node是一個廣播motion節點，因此能夠獲得outer table 的全量數據，保證hashjoin的結果是正確的。

這張圖片展現了上頁SQL的執行流程。有關 Slice、Gang 等更多信息，能夠參考 Greenplum 開發團隊出版的新書《Greenplum：從大數據戰略到實現》。

Greenplum 和 PostgreSQL 的Hashjoin實現基本類似。有幾個點作了加強：

• 支持對臨時batch文件進行壓縮，zstd 對壓縮解壓縮速度和壓縮比之間作了很好的平衡，因此採用了 zstd 壓縮算法 、
• 加入 Left anti semi join 類型以對 NOT IN場景優化。

可見單節點的 PostgreSQL 和並行數據庫Greenplum在hashjoin的執行層面都是基於 Hybrid hash join 算法，執行器層面實現細節幾乎沒有什麼不一樣，主要修改在於優化器層面。 其餘並行數據庫如 CitusDB 也是如此。

本文儘可能不涉及代碼細節，而是從邏輯層面講清楚 Hahsjoin 的實現邏輯。若是感興趣的話，能夠參考代碼。理解了處理邏輯，代碼看起來就比較容易了。相應的代碼在 nodehash.c 和 nodehashjoin.c

主要的代碼邏輯在 execHashJoin() 中，該函數實現了一個狀態機，其中有6個主要狀態，狀態變換大致如上上圖所示，僅作參考。

得到更多關於Greenpum的技術乾貨，請訪問Greenplum中文社區網站。