TiDB 源碼閱讀系列文章（十一）Index Lookup Join

TiDB 源碼閱讀系列文章（十一）Index Lookup Join

來自專欄 TiDB 的後花園27 人贊了文章

作者：徐懷宇

什麼是 Index Lookup Join

Nested Loop Join

在介紹 Index Lookup Join 之前，我們首先看一下什麼是 Nested Loop Join（NLJ）。 NLJ 的具體定義可以參考 Wikipedia。NLJ 是最為簡單暴力的 Join 演算法，其執行過程簡述如下：

• 遍歷 Outer 表，取一條數據 r；
• 遍歷 Inner 表，對於 Inner 表中的每條數據，與 r 進行 join 操作並輸出 join 結果；
• 重複步驟 1，2 直至遍歷完 Outer 表中的所有數據。

NLJ 演算法實現非常簡單並且 join 結果的順序與 Outer 表的數據順序一致。

但是存在性能上的問題：執行過程中，對於每一條 OuterRow，我們都需要對 Inner 表進行一次全表掃操作，這將消耗大量時間。

為了減少對於 Inner 表的全表掃次數，我們可以將上述步驟 1 優化為每次從 Outer 表中讀取一個 batch 的數據，優化後的演算法即 Block Nested-Loop Join（BNJ），BNJ 的具體定義可以參考 Wikipedia。

Index Lookup Join

對於 BNJ 演算法，我們注意到，對於 Outer 表中每個 batch，我們並沒有必要對 Inner 表都進行一次全表掃操作，很多時候可以通過索引減少數據讀取的代價。Index Lookup Join（ILJ） 在 BNJ 基礎上進行了改進，其執行過程簡述如下：

• 從 Outer 表中取一批數據，設為 B；
• 通過 Join Key 以及 B 中的數據構造 Inner 表取值範圍，只讀取對應取值範圍的數據，設為 S；
• 對 B 中的每一行數據，與 S 中的每一條數據執行 Join 操作並輸出結果；
• 重複步驟 1，2，3，直至遍歷完 Outer 表中的所有數據。

TiDB Index Lookup Join 的實現

TiDB 的 ILJ 運算元是一個多線程的實現，主要的線程有： Main Thead，Outer Worker，和 Inner Worker：

• Outer Worker 一個：
• 按 batch 遍歷 Outer 表，並封裝對應的 task
• 將 task 發送給 Inner Worker 和 Main Thread
• Inner Worker N 個：
• 讀取 Outer Worker 構建的 task
• 根據 task 中的 Outer 表數據，構建 Inner 表的掃描範圍，並構造相應的物理執行運算元讀取該範圍內的 Inner 表數據
• 對讀取的 Inner 表數據創建對應的哈希表並存入 task
• Main Thread 一個：
• 啟動 Outer Worker 及 Inner Workers
• 讀取 Outer Worker 構建的 task，並對每行 Outer 數據在對應的哈希表中 probe
• 對 probe 到的數據進行 join 並返回執行結果

這個運算元有如下特點：

• Join 結果的順序與 Outer 表的數據順序一致，這樣對上一層運算元可以提供順序保證；
• 對於 Outer 表中的每個 batch，只在 Inner 表中掃描部分數據，提升單個 batch 的處理效率；
• Outer 表的讀數據操作，Inner 表的讀數據操作，及 Join 操作並行執行，整體上是一個並行+Pipeline 的方式，儘可能提升執行效率。

執行階段詳述

TiDB 中 ILJ 的執行階段可劃分為如下圖所示的 5 步：

1. 啟動 Outer Worker 及 Inner Workers

這部分工作由 startWorkers 函數完成。該函數會 啟動一個 Outer Worker 和多個 Inner Worker 和 多個 Inner Worker。Inner Woker 的數量可以通過 tidb_index_lookup_concurrency 這個系統變數進行設置，默認為 4。

2. 讀取 Outer 表數據

這部分工作由 buildTask 函數完成。此處主要注意兩點：

第一點，對於每次讀取的 batch 大小，如果將其設置為固定值，則可能會出現如下問題：

• 若設置的 batch 值較大，但 Outer 表數據量較小時。各個 Inner Worker 所需處理的任務量可能會不均勻，出現數據傾斜的情況，導致並發整體性能相對單線程提升有限。
• 若設置的 batch 值較小，但 Outer 表數據量較大時。Inner Worker 處理任務時間短，需要頻繁從管道中取任務，CPU 不能被持續高效利用，由此帶來大量的線程切換開銷。此外, 當 batch 值較小時，同一批 inner 表數據能會被反覆讀取多次，帶來更大的網路開銷，對整體性能產生極大影響。

因此，我們通過指數遞增的方式動態控制 batch 的大小（由函數 increaseBatchSize 完成），以避免上述問題，batch size 的最大值由 session 變數 tidb_index_join_batch_size 控制，默認是 25000。讀取到的 batch 存儲在 lookUpJoinTask.outerResult 中。

第二點，如果 Outer 表的過濾條件不為空，我們需要對 outerResult 中的數據進行過濾（由函數 VectorizedFilter 完成）。outerResult 是 Chunk 類型（Chunk 的介紹請參考 TiDB 源碼閱讀系列文章（十）），如果對滿足過濾條件的行進行提取並重新構建對象進行存儲，會帶來不必要的時間和內存開銷。VectorizedFilter 函數通過一個長度與 outerResult 實際數據行數相等的 bool slice 記錄 outerResult 中的每一行是否滿足過濾條件以避免上述開銷。 該 bool slice 存儲在 lookUpJoinTask.outerMatch 中。

3. Outer Worker 將 task 發送給 Inner Worker 和 Main Thread

Inner Worker 需要根據 Outer 表每個 batch 的數據，構建 Inner 表的數據掃描範圍並讀取數據，因此 Outer Worker 需要將 task 發送給 Inner Worker。

如前文所述，ILJ 多線程並發執行，且 Join 結果的順序與 Outer 表的數據順序一致。 為了實現這一點，Outer Worker 通過管道將 task 發送給 Main Thread，Main Thread 從管道中按序讀取 task 並執行 Join 操作，這樣便可以實現在多線程並發執行的情況下的保序需求。

4. Inner Worker 讀取 inner 表數據

這部分工作由 handleTask 這個函數完成。handleTask 有如下幾個步驟:

• constructDatumLookupKeys 函數計算 Outer 表對應的 Join Keys 的值，我們可以根據 Join Keys 的值從 Inner 表中僅查詢所需要的數據即可，而不用對 Inner 表中的所有數據進行遍歷。為了避免對同一個 batch 中相同的 Join Keys 重複查詢 Inner 表中的數據，sortAndDedupDatumLookUpKeys 會在查詢前對前面計算出的 Join Keys 的值進行去重。
• fetchInnerResult 函數利用去重後的 Join Keys 構造對 Inner 表進行查詢的執行器，並讀取數據存儲於 task.innerResult 中。
• buildLookUpMap 函數對讀取的 Inner 數據按照對應的 Join Keys 構建哈希表，存儲於 task.lookupMap 中。

上述步驟完成後，Inner Worker 向 task.doneCh 中發送數據，以喚醒 Main Thread 進行接下來的工作。

5. Main Thread 執行 Join 操作

這部分工作由 prepareJoinResult 函數完成。prepareJoinResult 有如下幾個步驟：

• getFinishedTask 從 resultCh 中讀取 task，並等待 task.doneCh 發送來的數據，若該 task 沒有完成，則阻塞住；
• 接下來的步驟與 Hash Join類似（參考 TiDB 源碼閱讀系列文章（九）），lookUpMatchedInners 取一行 OuterRow 對應的 Join Key，從 task.lookupMap 中 probe 對應的 Inner 表的數據；
• 主線程對該 OuterRow，與取出的對應的 InnerRows 執行 Join 操作，寫滿存儲結果的 chk 後返回。

示例

CREATE TABLE `t` (`a` int(11) DEFAULT NULL,`pk` int(11) NOT NULL AUTO_INCREMENT,PRIMARY KEY (`pk`));CREATE TABLE `s` (`a` int(11) DEFAULT NULL,KEY `idx_s_a` (`a`));?insert into t(`a`) value(1),(1),(1),(4),(4),(5);insert into s value(1),(2),(3),(4);?select /*+ TIDB_INLJ(t) */ * from t left join s on t.a = s.a;

在上例中， t 為 Outer 表，s 為 Inner 表。 /** TIDN_INLJ */ 可以讓優化器儘可能選擇 Index Lookup Join 演算法。

設 Outer 表讀數據 batch 的初始大小為 2 行，Inner Worker 數量為 2。

查詢語句的一種可能的執行流程如下圖所示，其中由上往下箭頭表示時間線：

延展閱讀

（十）Chunk 和執行框架簡介

（九）Hash Join

（八）基於代價的優化

（七）基於規則的優化

（六）Select 語句概覽

（五）TiDB SQL Parser 的實現

（四）Insert 語句概覽

（三）SQL 的一生

（二）初識 TiDB 源碼

（一）序

推薦閱讀：