TiDB 源碼閱讀系列文章（九）Hash Join

06-27

來自專欄 TiDB 的後花園

TiDB 目前獲得了廣泛的關注，特別是一些技術愛好者希望能夠參與這個項目。由於整個系統的複雜性，很多人並不能很好的理解整個項目。我們希望通過 TiDB 源碼閱讀系列文章自頂向下，由淺入深，講述 TiDB 的技術原理以及實現細節，幫助大家掌握這個項目。
本文是 TiDB 源碼閱讀系列文章的第九篇。內文詳細介紹了 TiDB Hash Join 的實現以及幾種常見的問題，enjoy～

什麼是 Hash Join

Hash Join 的基本定義可以參考維基百科：Hash join。簡單來說，A 表和 B 表的 Hash Join 需要我們選擇一個 Inner 表來構造哈希表，然後對 Outer 表的每一行數據都去這個哈希表中查找是否有匹配的數據。

我們不用「小表」和「大表」這兩個術語是因為：對於類似 Left Outer Join 這種 Outer Join 來說，如果我們使用 Hash Join，不管 Left 表相對於 Right 表而言是大表還是小表，我們都只能使用 Right 表充當 Inner 表並在之上建哈希表，使用 Left 表來當 Outer 表，也就是我們的驅動表。使用 Inner 和 Outer 更準確，沒有迷惑性。在 Build 階段，對 Inner 表建哈希表，在 Probe 階段，對由 Outer 表驅動執行 Join 過程。

TiDB Hash Join 實現

TiDB 的 Hash Join 是一個多線程版本的實現，主要任務有：

Main Thread，一個，執行下列任務：

讀取所有的 Inner 表數據；
根據 Inner 表數據構造哈希表；
啟動 Outer Fetcher 和 Join Worker 開始後台工作，生成 Join 結果，各個 goroutine 的啟動過程由 fetchOuterAndProbeHashTable 這個函數完成；
將 Join Worker 計算出的 Join 結果返回給 NextChunk 介面的調用方法。

Outer Fetcher，一個，負責讀取 Outer 表的數據並分發給各個 Join Worker；
Join Worker，多個，負責查哈希表、Join 匹配的 Inner 和 Outer 表的數據，並把結果傳遞給 Main Thread。

接下來我們細緻的介紹 Hash Join 的各個階段。

Main Thread 讀 Inner 表數據

讀 Inner 表數據的過程由 fetchInnerRows 這個函數完成。這個過程會不斷調用 Child 的 NextChunk 介面，把每次函數調用所獲取的 Chunk 存儲到 innerResult 這個 List 中供接下來的計算使用。

2. Main Thread 構造哈希表

構造哈希表的過程由 buildHashTableForList 這個函數完成。

我們這裡使用的哈希表（存儲在變數 hashTable 中）本質上是一個 MVMap。MVMap 的 Key 和 Value 都是 []byte 類型的數據，和普通 map 不同的是，MVMap 允許一個 Key 擁有多個 Value。這個特性對於 Hash Join 來說非常方便和實用，因為表中同一個 Join Key 可能對應多行數據。

構造哈希表的過程中，我們會遍歷 Inner 表的每行數據（上文提到，此時所有的數據都已經存儲在了 innerResult 中），對每行數據做如下操作：

計算該行數據的 Join Key，得到一個 []byte，它將作為 MVMap 的 Key；
計算該行數據的位置信息，得到另一個 []byte，它將作為 MVMap 的 Value；
將這個 (Key, Value) 放入 MVMap 中。

3. Outer Fetcher

Outer Fetcher 是一個後台 goroutine，他的主要計算邏輯在 fetchOuterChunks 這個函數中。

它會不斷的讀大表的數據，並將獲得的 Outer 表的數據分發給各個 Join Worker。這裡多線程之間的資源交互可以用下圖表示：

上圖中涉及到了兩個 channel：

outerResultChs[i]：每個 Join Worker 一個，Outer Fetcher 將獲取到的 Outer Chunk 寫入到這個 channel 中供相應的 Join Worker 使用；
outerChkResourceCh：當 Join Worker 用完了當前的 Outer Chunk 後，它需要把這個 Chunk 以及自己對應的 outerResultChs[i] 的地址一起寫入到 outerChkResourceCh 這個 channel 中，告訴 Outer Fetcher 兩個信息：

我提供了一個 Chunk 給你，你直接用這個 Chunk 去拉 Outer 數據吧，不用再重新申請內存了；
我的 Outer Chunk 已經用完了，你需要把拉取到的 Outer 數據直接傳給我，不要給別人了。

所以，整體上 Outer Fetcher 的計算邏輯是：

i. 從 outerChkResourceCh 中獲取一個 outerChkResource，存儲在變數 outerResource 中；

ii. 從 Child 拉取數據，將數據寫入到 outerResource 的 chk 欄位中；

iii. 將這個 chk 發給需要 Outer 表的數據的 Join Worker 的 outerResultChs[i] 中去，這個信息記錄在了 outerResource 的 dest 欄位中。

4. Join Worker

每個 Join Worker 都是一個後台 goroutine，主要計算邏輯在 runJoinWorker4Chunk 這個函數中。Join Worker 的數量由 tidb_hash_join_concurrency這個 session 變數來控制，默認是 5 個。

上圖中涉及到兩個 channel：

joinChkResourceCh[i]：每個 Join Worker 一個，用來存 Join 的結果；
joinResultCh：Join Worker 將 Join 的結果 Chunk 以及它的 joinChkResourceCh 地址寫入到這個 channel 中，告訴 Main Thread 兩件事：

我計算出了一個 Join 的結果 Chunk 給你，你讀到這個數據後可以直接返回給你 Next 函數的調用方；
你用完這個 Chunk 後趕緊還給我，不要給別人，我好繼續幹活。

所以，整體上 Join Worker 的計算邏輯是：

i. 獲取一個 Outer Chunk；

ii. 獲取一個 Join Chunk Resource；

iii. 查哈希表，將匹配的 Outer Row 和 Inner Rows 寫到 Join Chunk 中；

iv. 將寫滿了的 Join Chunk 發送給 Main Thread。

5. Main Thread

主線程的計算邏輯由 NextChunk 這個函數完成。主線程的計算邏輯非常簡單：

i. 從 joinResultCh 中獲取一個 Join Chunk；

ii. 將調用方傳下來的 chk 和 Join Chunk 中的數據交換；

iii. 把 Join Chunk 還給對應的 Join Worker。

Hash Join FAQ

1. 如何確定 Inner 和 Outer 表？

Left Outer Join：左表是 Outer 表，右表是 Inner 表；
Right Outer Join：跟 Left Outer Join 相反，右表是 Outer 表，左表是 Inner 表；
Inner Join：優化器估算出的較大表是 Outer 表，較小的表是 Inner 表；
Semi Join、Anti Semi Join、Left Outer Semi Join 或 Anti Left Outer Semi Join：左表是 Outer 表，右表是 Inner 表。

2. Join Key 中 NULL 值的問題

NULL 和 NULL 不等，所以：

在用 Inner 表建 NULL 值的時候會忽略掉 Join Key 中有 NULL 的數據（代碼在這裡）；
當 Outer 表中某行數據的 Join Key 中有 NULL 值的時候我們不會去查哈希表（代碼在這裡）。

3. Join 中的 4 種 Filter

Inner 表上的 Filter：這種 Filter 目前被優化器推到了 Hash Join Inner 表上面，在 Hash Join 實現的過程中不用考慮這種 Filter 了。推下去的原因是能夠儘早的在 coprocessor 上就把不能匹配到的 Inner 表數據給過濾掉，給上層計算減壓。
Outer 表上的 Filter：這種 Filter 的計算目前在 join2Chunk 中，由 Join Worker 進行。當 Join Worker 拿到一個 Outer Chunk 以後需要先計算 Outer Filter，如果通過了 Outer Filter 再去查哈希表。
兩個表上的等值條件：這就是我們說的 Join Key。比如 A 表和 B 表的等值條件是：A.col1=B.col2 and A.col3=B.col4，那麼 A 表和 B 表上的 Join Key 分別是 (col1, col3) 和 (col2, col4)。
兩個表上的非等值條件：這種 Filter 需要在 Join 的結果集上計算，如果能夠過這個 Filter 才認為兩行數據能夠匹配。這個 Filter 的計算過程交給了 joinResultGenerator。

4. Join 方式的實現

目前 TiDB 支持的 Join 方式有 7 種，我們使用 joinResultGenerator 這個介面來定義兩行數據的 Join 方式，實現一種具體的 Join 方式需要特殊的去實現 joinResultGenerator 這個介面，目前有 7 種實現：

semiJoinResultGenerator：實現了 Semi Join 的鏈接方式，當一個 Outer Row 和至少一個 Inner Row 匹配時，輸出這個 Outer Row。
antiSemiJoinResultGenerator：實現了 Anti Semi Join 的鏈接方式，當 Outer Row 和所有的 Inner Row 都不能匹配時才輸出這個 Outer Row。
leftOuterSemiJoinResultGenerator：實現了 Left Outer Semi Join 的鏈接方式，Join 的結果是 Outer Row + 一個布爾值，如果該 Outer Row 能和至少一個 Inner Row 匹配，則輸出該 Outer Row + True，否則輸出 Outer Row + False。
antiLeftOuterSemiJoinResultGenerator：實現了 Anti Left Outer Semi Join 的鏈接方式，Join 的結果也是 Outer Row + 一個布爾值，不同的是，如果該 Outer Row 不能和任何 Inner Row 匹配上，則輸出 Outer Row + True，否則輸出 Outer Row + False。
leftOuterJoinResultGenerator：實現了 Left Outer Join 的鏈接方式，如果 Outer Row 不能和任何 Inner Row 匹配，則輸出 Outer Row + NULL 填充的 Inner Row，否則輸出每個匹配的 Outer Row + Inner Row。
rightOuterJoinResultGenerator：實現了 Right Outer Join 的鏈接方式，如果 Outer Row 不能和 Inner Row 匹配，則輸出 NULL 填充的 Inner Row + Outer Row，否則輸出每個匹配的 Inner Row + Outer Row。
innerJoinResultGenerator：實現了 Inner Join 的鏈接方式，如果 Outer Row 不能和 Inner Row 匹配，不輸出任何數據，否則根據 Outer Row 是左表還是右表選擇性的輸出每個匹配的 Inner Row + Outer Row 或者 Outer Row + Inner Row。