作者 | 劉垚

編輯 | 爾悅

小 T 導讀：在(zai)使用或者實(shi)現(xian)分布(bu)式數(shu)(shu)據庫（Distributed Database）時(shi)，會面臨(lin)把一個表的(de)(de)數(shu)(shu)據按(an)照一定(ding)的(de)(de)策略分散到各個數(shu)(shu)據庫節點上(shang)的(de)(de)情況，隨(sui)之而來的(de)(de)是多節點數(shu)(shu)據查詢(xun)復雜(za)性的(de)(de)問題，例(li)如 Join 和子(zi)查詢(xun)。本文將(jiang)會為你(ni)解讀分布(bu)式數(shu)(shu)據庫下子(zi)查詢(xun)和 Join 等復雜(za) SQL 如何實(shi)現(xian)，來幫(bang)助你(ni)更好地解決上(shang)述問題。

首(shou)先簡單講一下(xia) SQL 的(de)執行(xing)過程：

SQL ==> Parser ==> Translate & Semantic Check ==> Optimizer ==> Coordinator ==> Executer

• Parser 產生的是語法樹，即 Abstract Syntax Tree；
• Translate & Semantic Check，這一步會從 Catalog 讀取元數據，用元數據完善語法樹，便于 Optimizer 使用。例如：常見的 select * from tableA，一般會在這一步把“*”換成 tableA 的列；
• Optimizer 產生的是優化之后的邏輯執行計劃，即 Optimized Logical Plan，執行計劃是個有向無環圖，即 DAG；
• Coordinator 負責分發邏輯執行計劃給各個節點去計算；
• Executer 會把邏輯執行計劃轉成物理執行計劃，即 Physical Plan。

開源的數據庫有很多，我們可以結合一些主流數據庫的源代碼來理解子查詢和 Join 的實現方式，比如關系型數據庫 ：Impala、Presto、ClickHouse，時序數據庫（Time- Series Database）： TDengine 等。下(xia)面從子查詢和 Join 兩部分進行分析(xi)。

子查詢部分

邏輯(ji)執行(xing)計(ji)劃有多種(zhong) Node，分別(bie)對(dui)應(ying)著(zhu) SQL 中的(de)(de)各種(zhong)計(ji)算(suan)(suan)，包括 Scan Node、Join Node、Aggregate Node、Sort Node、Project Node 等(deng)等(deng)，相(xiang)應(ying)的(de)(de)物理執行(xing)計(ji)劃的(de)(de)算(suan)(suan)子為 Scan Operator 、Join Operator、Aggregate Operator、Sort Operator、Project Operator 等(deng)等(deng)。而數(shu)(shu)據(ju)庫一(yi)般沒有計(ji)算(suan)(suan)子查詢的(de)(de)算(suan)(suan)子，這是(shi)因(yin)為將抽象語法樹轉成邏輯(ji)執行(xing)計(ji)劃之(zhi)后，就(jiu)已經沒有子查詢的(de)(de)概(gai)念了(le)，其運(yun)行(xing)邏輯(ji)是(shi)數(shu)(shu)據(ju)算(suan)(suan)子之(zhi)間自下(xia)而上逐(zhu)層(ceng)傳遞，并逐(zhu)層(ceng)計(ji)算(suan)(suan)，并不(bu)特(te)別(bie)計(ji)算(suan)(suan)子查詢。下(xia)面講一(yi)下(xia)分布式(shi)數(shu)(shu)據(ju)庫針對(dui)子查詢的(de)(de)一(yi)些相(xiang)關處理。

首(shou)先(xian)，分布式數(shu)據庫的(de)優化器會將子(zi)查(cha)詢扁(bian)平(ping)(ping)化處(chu)理，這種(zhong)方(fang)式一(yi)般分為兩(liang)種(zhong)，一(yi)種(zhong)是(shi)直接在語(yu)法樹（AST）上做子(zi)查(cha)詢扁(bian)平(ping)(ping)化（Subquery Flatten），另外(wai)一(yi)種(zhong)是(shi)在生(sheng)成邏輯執行計劃(hua)時進行扁(bian)平(ping)(ping)化。這兩(liang)種(zhong)方(fang)式本質上大同(tong)小異，都要保證語(yu)義的(de)等價性。但(dan)也并不是(shi)所(suo)有的(de)子(zi)查(cha)詢都能扁(bian)平(ping)(ping)化，有如下幾(ji)種(zhong)特殊情況(kuang)：

• 子查詢和父查詢都有聚集函數
• 子查詢有聚集函數，并且父查詢有分組計算（Group By）
• 子查詢有聚集函數，并且用子查詢聚集函數的結果關聯（Join）父查詢的表
• 父查詢有聚集函數，并且子查詢有分組計算（Group By）
• 子查詢有 Limit（限制返回結果的行數），并且父查詢有過濾條件（Where）或者分組計算、排序（Order By）
• 其他

基于 AST 進行(xing)(xing)子(zi)查詢扁平化(hua)時，需(xu)要先遍歷語法數(shu)據(ju)，并按規則(ze)進行(xing)(xing)判斷，進而去除不必要的(de)子(zi)查詢。對于生(sheng)成邏輯執(zhi)行(xing)(xing)計劃(hua)時的(de)子(zi)查詢扁平化(hua)，在(zai)生(sheng)成 Plan Node 時需(xu)要先去除冗余的(de) Node，舉個例子(zi)，SQL：select colA from (select * from tA) group by colA;

一般來說，邏輯執行(xing)計劃會有(you)多個子(zi)計劃，通常(chang)在需要網(wang)絡傳(chuan)輸時才會產生子(zi)計劃，需要注意的是子(zi)計劃和子(zi)查詢(xun)之間并沒有(you)必然的聯系，即有(you)子(zi)查詢(xun)不一定對應一個子(zi)計劃。

Join 部分

首先(xian)，分布(bu)式數據庫會對(dui) Join 進行(xing)優(you)化(hua)(hua)，包(bao)括 Join 消(xiao)除(chu)（例如基(ji)于(yu)(yu)主鍵(jian)外(wai)(wai)鍵(jian)去除(chu)不(bu)必要的 Join）、外(wai)(wai)連接消(xiao)除(chu)（Outer Join 轉成(cheng) Inner Join）、Join Order 優(you)化(hua)(hua)（基(ji)于(yu)(yu)數據的統計信息，用動態規劃算法、貪心算法或遺傳(chuan)算法等優(you)化(hua)(hua) Table 的 Join 順序）等等。

再(zai)講一下 Join 的三(san)種基本算法：Hash Join（必須(xu)要有等(deng)值(zhi)連接(jie)條(tiao)件，例如(ru) t1.colA = t2.colB）、Merge Join（左表和右表的數據都是(shi)有序的，按連接(jie)條(tiao)件中的列有序）、Nestloop Join（含有非(fei)等(deng)值(zhi)連接(jie)條(tiao)件并(bing)且數據無序）。在實際當中，會把(ba)三(san)種算法進行混合使用，這(zhe)是(shi)因為 Join 條(tiao)件可以同時包含等(deng)值(zhi)連接(jie)和非(fei)等(deng)值(zhi)連接(jie)，例如(ru) t1.colA = t2.colB AND t1.colC > t2.colC

Hash Join

在進行(xing) Join Order 優(you)化(hua)時(shi)，優(you)化(hua)器會調整左(zuo)(zuo)表(biao)(biao)和右(you)表(biao)(biao)的順(shun)序(xu)，一般把小表(biao)(biao)放(fang)右(you)邊，大表(biao)(biao)放(fang)左(zuo)(zuo)邊，并且選擇(ze) Join 模式：Shuffle Join（按照關(guan)聯條件，同時(shi) shuffle 左(zuo)(zuo)表(biao)(biao)和右(you)表(biao)(biao)，然后再計(ji)算 Join） 或(huo) Boradcast Join（把右(you)表(biao)(biao)廣(guang)播到左(zuo)(zuo)表(biao)(biao)所(suo)在的節點(dian)，注意左(zuo)(zuo)表(biao)(biao)不動，然后再計(ji)算 Join）。一般是基于(yu)代價去(qu)選擇(ze) Join Order 優(you)化(hua)，但考慮到統計(ji)信(xin)息可(ke)能會存(cun)在誤差(cha)，因此很多數據庫可(ke)以通過(guo) Hint、Query Option 等方式，由(you)用戶來(lai)指定 Join 順(shun)序(xu)、Join 模式等。

Hash Join 是目前(qian)最常(chang)用(yong)的(de) Join 算法(fa)，大部分數(shu)據庫都實現了(le) Hash Join。這種算法(fa)會(hui)先讀取右(you)表，并把右(you)表的(de)數(shu)據放(fang)入 Hash Map 里，如(ru)果存(cun)不下就會(hui)放(fang)入外存(cun)。通常(chang)情況下，各(ge)個數(shu)據庫都會(hui)實現自己的(de) Hash Map，很少直接使用(yong) STL 或 Boost 等第三方庫中的(de) Hash Map，原因主要有兩(liang)點(dian)：

• 定制化 Hash Map 會提升 Join 計算速度。
• 定制化 Hash Map 能更準確地控制內存使用，當內存不足時，會使用外存，定制化 Hash Map 可以根據 Join 算法，優化 Swap 機制，減少 Swap 的數據量。Hash Map 的結構如下：

右表(biao)可能含有重(zhong)(zhong)復(fu)的(de)數據(ju)，所以(yi)會有 Duplicate Node。這(zhe)里的(de)重(zhong)(zhong)復(fu)數據(ju)是(shi)指 Join Key（Join 條件對(dui)應(ying)的(de)列(lie)(lie)）的(de)數據(ju)重(zhong)(zhong)復(fu)，并且其他列(lie)(lie)不重(zhong)(zhong)復(fu)，所以(yi)要(yao)(yao)分別緩(huan)存。注(zhu)意上(shang)述圖中(zhong)，是(shi)通過 Hash 算法解決 Hash 沖突(tu)的(de)問題，即不會把不同(tong)(tong)的(de) Join Key 放(fang)在同(tong)(tong)一個桶(tong)中(zhong)。當然，現實操作(zuo)中(zhong)也(ye)有把不同(tong)(tong)的(de) Join Key 放(fang)在同(tong)(tong)一個桶(tong)中(zhong)的(de)情況，那(nei)需(xu)要(yao)(yao)遍歷(li) List 才能確定查找的(de) Join Key 是(shi)否(fou)存在。

Merge Join

Merge Join 一般(ban)是在左表(biao)和右表(biao)的數據是有序的情況下使用(yong)。例(li)如(ru)時序數據庫 TDengine，數據按時間(jian)戳列有序，那么用(yong)時間(jian)戳列做 Join 時，TDengine database 會(hui)用(yong) Merge Join 來計算，這樣(yang)的一個好處(chu)是處(chu)理速(su)度非(fei)常快，并且占用(yong)內存非(fei)常小。

Nestloop Join

這種 Join 算法(fa)速度非(fei)常慢，但對(dui)于全功能數據庫而言是(shi)不可(ke)缺少的。使(shi)用這種算法(fa)時，可(ke)以(yi)結合(he)索引來(lai)提速。

總(zong)結(jie)而言，Hash Join 使用最廣，適用于很多數據(ju)分(fen)析的場景，并且(qie)大部(bu)分(fen)數據(ju)庫都支持；Merge Join 一(yi)般是(shi)在左右(you)表數據(ju)有序時才會使用，不需要緩(huan)存(cun)數據(ju)，所以(yi)使用內存(cun)非(fei)常少(shao)，計算速度是(shi)三種 Join 算法中最快的；Nestloop Join 性能很差，分(fen)布式(shi)數據(ju)庫一(yi)般很少(shao)使用，有些分(fen)布式(shi)數據(ju)庫就(jiu)不支持，可以(yi)通(tong)過索引來加(jia)速 Nestloop Join。

寫在最后

上面我們(men)對子查詢和 Join 兩種復雜 SQL 的(de)實(shi)現(xian)方式做了具體(ti)解讀，大(da)家(jia)可以結合一些開源數據庫(ku)的(de)源代(dai)碼來理解，像 TDengine 的(de)源代(dai)碼都可以在 上看到，如果(guo)你對時序數據庫(ku)的(de)復雜 SQL 實(shi)現(xian)有興(xing)趣，這就是(shi)一個不錯的(de)觀摩(mo)對象。也(ye)歡(huan)迎(ying)大(da)家(jia)在下(xia)方評論區進行交流。