在《探究Presto SQL引擎(1)-巧用Antlr》中，我們介紹了Antlr的基本用法以及如何使用Antlr4實(shí)現(xiàn)解析SQL查詢CSV數(shù)據(jù)，更加深入理解Presto查詢引擎支持的SQL語(yǔ)法以及實(shí)現(xiàn)思路。

本次帶來(lái)的是系列文章的第2篇，本文梳理了Join的原理，以及Join算法在Presto中的實(shí)現(xiàn)思路。通過(guò)理論和實(shí)踐的結(jié)合，可以在理解原理的基礎(chǔ)上，更加深入理解Join算法在OLAP場(chǎng)景下的工程落地技巧，比如火山模型，列式存儲(chǔ)，批量處理等思想的應(yīng)用。

一、背景

在業(yè)務(wù)開(kāi)發(fā)中使用數(shù)據(jù)庫(kù)，通常會(huì)有規(guī)范不允許過(guò)多表的Join。例如阿里巴巴開(kāi)發(fā)手冊(cè)中，有如下的規(guī)定：

【強(qiáng)制】超過(guò)三個(gè)表禁止Join。需要Join的字段，數(shù)據(jù)類型必須絕對(duì)一致；多表關(guān)聯(lián)查詢時(shí)，保證被關(guān)聯(lián)的字段需要有索引。說(shuō)明：即使雙表Join也要注意表索引、SQL性能。

在大數(shù)據(jù)數(shù)倉(cāng)的建設(shè)中，盡管我們有星型結(jié)構(gòu)和雪花結(jié)構(gòu)，但是最終交付業(yè)務(wù)使用的大多是寬表。

可以看出業(yè)務(wù)使用數(shù)據(jù)庫(kù)中的一個(gè)矛盾點(diǎn)：我們需要Join來(lái)提供靈活的關(guān)聯(lián)操作，但是又要盡量避免多表和大表Join帶來(lái)的性能問(wèn)題。這是為什么呢？

二、Join的基本原理

在數(shù)據(jù)庫(kù)中Join提供的語(yǔ)義是非常豐富的。簡(jiǎn)單總結(jié)如下：

通常理解Join的實(shí)現(xiàn)原理，從Cross Join是最好的切入點(diǎn)，也就是所謂的笛卡爾積。對(duì)于集合進(jìn)行笛卡爾積運(yùn)算，理解非常簡(jiǎn)單，就是窮舉兩個(gè)集合中元素所有的組合情況。在數(shù)據(jù)庫(kù)中，集合就對(duì)應(yīng)到數(shù)據(jù)表中的所有行(tuples)，集合中的元素就對(duì)應(yīng)到單行(tuple)。所以實(shí)現(xiàn)Cross Join的算法也就呼之欲出了。實(shí)現(xiàn)的代碼樣例如下：

List  r = newArrayList(
new Tuple(newArrayList(1,"a")),
new Tuple(newArrayList(2,"b")));
List  s = newArrayList(
new Tuple(newArrayList(3,"c")),
new Tuple(newArrayList(4,"d")));
int cnt =0;
for(Tuple ri:r){
for(Tuple si:s){
Tuple c = new Tuple().merge(ri).merge(si);
System.out.println(++cnt+": "+ c);
}
}
/**
? out:
1: [1, a, 3, c]
2: [1, a, 4, d]
3: [2, b, 3, c]
4: [2, b, 4, d]
*/

可以看出實(shí)現(xiàn)邏輯非常簡(jiǎn)單，就是兩個(gè)For循環(huán)嵌套。

2.1 Nested Loop Join算法

在這個(gè)基礎(chǔ)上，實(shí)現(xiàn)Inner Join的第一個(gè)算法就順其自然了。非常直白的名稱：Nested Loop，，實(shí)現(xiàn)關(guān)鍵點(diǎn)如下：

（來(lái)源：Join Processing in Relational Databases）

其中，θ操作符可以是：=, !=, <, >, ≤, ≥。

相比笛卡爾積的實(shí)現(xiàn)思路，也就是添加了一層if條件的判斷用于過(guò)濾滿足條件的組合。

對(duì)于Nested Loop算法，最關(guān)鍵的點(diǎn)在于它的執(zhí)行效率。假如參與Join的兩張表一張量級(jí)為1萬(wàn)，一張量級(jí)為10w，那么進(jìn)行比較的次數(shù)為1w*10w=10億次。在大數(shù)據(jù)時(shí)代，通常一張表數(shù)據(jù)量都是以億為單位，如果使用Nested Loop Join算法，那么Join操作的比較次數(shù)直接就是天文數(shù)字了。所以Nested Loop Join基本上是作為萬(wàn)不得已的保底方案。Nested Loop這個(gè)框架下，常見(jiàn)的優(yōu)化措施如下：

• 小表驅(qū)動(dòng)大表，即數(shù)據(jù)量較大的集作為于for循環(huán)的內(nèi)部循環(huán)。
• 一次處理一個(gè)數(shù)據(jù)塊，而不是一條記錄。也就是所謂的Block Nested Loop Join，通過(guò)分塊降低IO次數(shù)，提升緩存命中率。

值得一提的是Nested Loop Join的思想雖然非常樸素，但是天然的具備分布式、并行的能力。這也是為什么各類NoSQL數(shù)據(jù)庫(kù)中依然保留Nested Loop Join實(shí)現(xiàn)的重要一點(diǎn)。雖然單機(jī)串行執(zhí)行慢，但是可以并行化的話，那就是加機(jī)器能解決的問(wèn)題了。

2.2 Sort Merge Join算法

通過(guò)前面的分析可以知道，Nested Loop Join算法的關(guān)鍵問(wèn)題在于比較次數(shù)過(guò)多，算法的復(fù)雜度為O(m*n)，那么突破口也得朝著這個(gè)點(diǎn)。如果集合中的元素是有序的，比較的次數(shù)會(huì)大幅度降低，避免很多無(wú)意義的比較運(yùn)算。對(duì)于有序的所以Join的第二種實(shí)現(xiàn)方式如下所描述：

（來(lái)源：Join Processing in Relational Databases）

通過(guò)將JOIN操作拆分成Sort和Merge兩個(gè)階段實(shí)現(xiàn)Join操作的加速。對(duì)于Sort階段，是可以提前準(zhǔn)備好可以復(fù)用的。這樣的思想對(duì)于MySQL這類關(guān)系型數(shù)據(jù)庫(kù)是非常友好的，這也能解釋阿里巴巴開(kāi)發(fā)手冊(cè)中要求關(guān)聯(lián)的字段必須建立索引，因?yàn)樗饕ＷC了數(shù)據(jù)有序。該算法時(shí)間復(fù)雜度為排序開(kāi)銷O(mlog(m)+nlog(n))+合并開(kāi)銷O(m+n)。但是通常由于索引保證了數(shù)據(jù)有序，索引其時(shí)間復(fù)雜度為O(m+n)。

2.3 Hash Join算法

Sort Merge Join的思想在落地中有一定的限制。所謂成也蕭何敗蕭何，對(duì)于基于Hadoop的數(shù)倉(cāng)而言，保證數(shù)據(jù)存儲(chǔ)的有序性這個(gè)點(diǎn)對(duì)于性能影響過(guò)大。在海量數(shù)據(jù)的背景下，維護(hù)索引成本是比較大的。而且索引還依賴于使用場(chǎng)景，不可能每個(gè)字段都建一個(gè)索引。在數(shù)據(jù)表關(guān)聯(lián)的場(chǎng)景是大表關(guān)聯(lián)小表時(shí)，比如：用戶表(大表)--當(dāng)日訂單表(小表)；事實(shí)表(大表)–維度表(小表)，可以通過(guò)空間換時(shí)間?；叵胍幌拢诨A(chǔ)的數(shù)據(jù)結(jié)構(gòu)中，tree結(jié)構(gòu)和Hash結(jié)構(gòu)可謂數(shù)據(jù)處理的兩大法寶：一個(gè)保證數(shù)據(jù)有序方便實(shí)現(xiàn)區(qū)間搜索，一個(gè)通過(guò)hash函數(shù)實(shí)現(xiàn)精準(zhǔn)命中點(diǎn)對(duì)點(diǎn)查詢效率高。

在這樣的背景下，通過(guò)將小表Hash化，實(shí)現(xiàn)Join的想法也就不足為奇了。

（來(lái)源：Join Processing in Relational Databases）

而且即使一張表在單機(jī)環(huán)境生成Hash內(nèi)存消耗過(guò)大，還可以利用Hash將數(shù)據(jù)進(jìn)行切分，實(shí)現(xiàn)分布式能力。所以，在Presto中Join算法通常會(huì)選擇Hash Join，該算法的時(shí)間復(fù)雜度為O(m+n)。

通過(guò)相關(guān)資料的學(xué)習(xí)，可以發(fā)現(xiàn)Join算法的實(shí)現(xiàn)原理還是相當(dāng)簡(jiǎn)單的，排序和Hash是數(shù)據(jù)結(jié)構(gòu)最為基礎(chǔ)的內(nèi)容。了解了Join的基本思想，如何落地實(shí)踐出來(lái)呢？畢竟talk is cheap。在項(xiàng)目中實(shí)現(xiàn)Join之前，需要一些鋪墊知識(shí)。通常來(lái)說(shuō)核心算法是皇冠上的明珠，但是僅有明珠是不夠的還需要皇冠作為底座。

三、Join工程化前置條件

3.1 SQL處理架構(gòu)-火山模型

在將Join算法落地前，需要先了解一下數(shù)據(jù)庫(kù)處理數(shù)據(jù)的基本架構(gòu)。在理解架構(gòu)的基礎(chǔ)上，才能將Join算法放置到合適的位置。在前面系列文章中探討了基于antlr實(shí)現(xiàn)SQL語(yǔ)句的解析?？梢园l(fā)現(xiàn)SQL語(yǔ)法支持的操作類型非常豐富：查詢表(TableScan)，過(guò)濾數(shù)據(jù)(Filter)，排序(Order)，限制(Limit)，字段進(jìn)行運(yùn)算(Project)， 聚合(Group)，關(guān)聯(lián)(Join)等。為了實(shí)現(xiàn)上述的能力，需要一個(gè)具備并行化能力且可擴(kuò)展的架構(gòu)。

1994年Goetz Graefe在論文《Volcano-An Extensible and Parallel Query Evaluation System》提出了一個(gè)架構(gòu)設(shè)計(jì)思想，這就是大名鼎鼎的火山模型，也稱為迭代模型?；鹕侥Ｐ推鋵?shí)包含了文件系統(tǒng)和查詢處理兩個(gè)部分，這里我們重點(diǎn)關(guān)注查詢處理的設(shè)計(jì)思想。架構(gòu)圖如下：

(來(lái)源:《Balancing vectorized execution with bandwidth-optimized storage》)

簡(jiǎn)單解讀一下：

職責(zé)分離：將不同操作獨(dú)立成一個(gè)的Operator，Operator采用open-next-close的迭代器模式。例如對(duì)于SQL 。

SELECT Id, Name, Age, (Age - 30) * 50 AS Bonus
FROM People
WHERE Age > 30

對(duì)應(yīng)到Scan， Select, Project三個(gè)Operator，數(shù)據(jù)交互通過(guò)next()函數(shù)實(shí)現(xiàn)。上述的理論在Presto中可以對(duì)應(yīng)起來(lái)，例如Presto中幾個(gè)常用的Operator, 基本上是見(jiàn)名知意：

動(dòng)態(tài)組裝：Operator基于SQL語(yǔ)句的解析實(shí)現(xiàn)動(dòng)態(tài)組裝，多個(gè)Operator形成一個(gè)管道(pipeline)。例如：print和predicate兩個(gè)operator形成一個(gè)管道：

(來(lái)源: 《Volcano-An Extensible and Parallel Query Evaluation System》)

在火山模型的基礎(chǔ)上，Presto吸收了數(shù)據(jù)庫(kù)領(lǐng)域的其他思想，對(duì)基礎(chǔ)的火山模型進(jìn)行了優(yōu)化改造，主要體現(xiàn)在如下幾點(diǎn)：

• Operator數(shù)據(jù)處理優(yōu)化成一次一個(gè)Page，而不是一次行(也稱為tuple)。
• Page的存儲(chǔ)采用列式結(jié)構(gòu)。即相同的列封裝到一個(gè)Block中。

批量處理結(jié)合列式存儲(chǔ)奠定了向量化計(jì)算的基礎(chǔ)。這也是數(shù)據(jù)庫(kù)領(lǐng)域的優(yōu)化方向。

3.2 批量處理和列式存儲(chǔ)

在研讀Presto源碼時(shí)，幾乎到處都可以看到Page/Block的身影。所以理解Page/Block背后的思想是理解Presto實(shí)現(xiàn)機(jī)制的基礎(chǔ)。有相關(guān)書(shū)籍和文檔講解Page/Block的概念，但是由于這些概念是跟其他概念混在一起呈現(xiàn)，導(dǎo)致一時(shí)間不容易理解。

筆者認(rèn)為Type-Block-Page三者放在一起，更容易理解。我們使用數(shù)據(jù)庫(kù)，通常需要定義表，字段名稱，字段類型。在傳統(tǒng)的DBMS中，通常是按行存儲(chǔ)數(shù)據(jù)，通常結(jié)構(gòu)如下：

(來(lái)源：《數(shù)據(jù)庫(kù)系統(tǒng)實(shí)現(xiàn)》)

但是通常OLAP場(chǎng)景不需要讀取所有的字段，基于這樣的場(chǎng)景，就衍生出來(lái)了列式存儲(chǔ)。就是我們看到的如下結(jié)構(gòu)：

(來(lái)源：《Presto技術(shù)內(nèi)幕》)

即每個(gè)字段對(duì)應(yīng)一個(gè)Block， 多個(gè)Block的切面才是一條記錄，也就是所謂的行，在一些論文中稱為tuple。通過(guò)對(duì)比可以清楚看出Presto中，Page就是典型了列式存儲(chǔ)的實(shí)現(xiàn)。所以在Presto中，每個(gè)Type必然會(huì)關(guān)聯(lián)到一種Block。例如：bigint類型就對(duì)應(yīng)著LongArrayBlockBuilder，varchar類型對(duì)應(yīng)著VariableWidthBlock。

理解了原理，操作Page/Block就變得非常簡(jiǎn)單了，簡(jiǎn)單的demo代碼如下：

import com.facebook.presto.common.Page;
import com.facebook.presto.common.PageBuilder;
import com.facebook.presto.common.block.Block;
import com.facebook.presto.common.block.BlockBuilder;
import com.facebook.presto.common.type.BigintType;
import com.facebook.presto.common.type.Type;
import com.facebook.presto.common.type.VarcharType;
import com.google.common.collect.Lists;
import io.airlift.slice.Slice;
import java.util.List;
import static io.airlift.slice.Slices.utf8Slice;
/**
? PageBlockDemo
? 
? @version 1.0
? @since 2021/6/22 19:26
*/
public class PageBlockDemo {
private static Page buildPage(List types,List<Object[]> dataSet){
PageBuilder pageBuilder = new PageBuilder(types);
// 封裝成Page
for(Object[] row:dataSet){
// 完成一行
pageBuilder.declarePosition();
for (int column = 0; column < types.size(); column++) {
BlockBuilder out =  pageBuilder.getBlockBuilder(column);
          Object colVal = row[column];
         if(colVal == null){
             out.appendNull();
         }else{
             Type type = types.get(column);
             Class<?> javaType = type.getJavaType();
             if(javaType == long.class){
                 type.writeLong(out,(long)colVal);
             }else if(javaType == Slice.class){
                 type.writeSlice(out, utf8Slice((String)colVal));
             }else{
                 throw new UnsupportedOperationException("not implemented");
             }
         }
     }
 }
 // 生成Page
 Page page = pageBuilder.build();
 pageBuilder.reset();
 return page; 
  }
private static void readColumn(List types,Page page){
// 從Page中讀取列
for(int column=0;column<types.size();column++){
Block block = page.getBlock(column);
Type type = types.get(column);
Class<?> javaType = type.getJavaType();
       System.out.print("column["+type.getDisplayName()+"]>>");
     List<Object> colList = Lists.newArrayList();
     for(int pos=0;pos<block.getPositionCount();pos++){
         if(javaType == long.class){
             colList.add(block.getLong(pos));
         }else if(javaType == Slice.class){
             colList.add(block.getSlice(pos,0,block.getSliceLength(pos)).toStringUtf8());
         }else{
             throw new UnsupportedOperationException("not implemented");
         }
     }
     System.out.println(colList);
 }
  }
public static void main(String[] args) {
/**
* 假設(shè)有兩個(gè)字段，一個(gè)字段類型為int, 一個(gè)字段類型為varchar
*/
List types = Lists.newArrayList(BigintType.BIGINT, VarcharType.VARCHAR);
  }
public static void main(String[] args) {
/**
* 假設(shè)有兩個(gè)字段，一個(gè)字段類型為int, 一個(gè)字段類型為varchar
*/
List types = Lists.newArrayList(BigintType.BIGINT, VarcharType.VARCHAR);
   // 按行存儲(chǔ)
 List<Object[]> dataSet = Lists.newArrayList(
         new Object[]{1L,"aa"},
         new Object[]{2L,"ba"},
         new Object[]{3L,"cc"},
         new Object[]{4L,"dd"});

 Page page = buildPage(types, dataSet);
 readColumn(types,page);
  }
}
// 運(yùn)行結(jié)果：
//column[bigint]>>[1, 2, 3, 4]
//column[varchar]>>[aa, ba, cc, dd]

將數(shù)據(jù)封裝成Page在各個(gè)Operator中流轉(zhuǎn)，一方面避免了對(duì)象的序列化和反序列化成本，另一方面相比tuple的方式降低了函數(shù)調(diào)用的開(kāi)銷。這跟集裝箱運(yùn)貨降低運(yùn)輸成本的思想是類似的。

四、Join算法的工程實(shí)踐

理解了Join的核心算法和基礎(chǔ)架構(gòu)，結(jié)合前文中對(duì)antlr實(shí)現(xiàn)SQL表達(dá)式的解析以及實(shí)現(xiàn)where條件過(guò)濾，我們已經(jīng)具備了實(shí)現(xiàn)Join的基礎(chǔ)條件。接下來(lái)簡(jiǎn)單講述一下Join算法的落地流程。首先在語(yǔ)法層面需要支持Join的語(yǔ)法，由于本文目的在于研究算法實(shí)現(xiàn)流程，而不在于實(shí)現(xiàn)完整的Join功能，因此我們暫且先考慮支持兩張表單字段的等值Join語(yǔ)法。

首先在語(yǔ)法上需要支持Join, 基于antlr語(yǔ)法的定義關(guān)鍵點(diǎn)如下：

querySpecification
: SELECT  selectItem (',' selectItem)*
(FROM relation (',' relation)*)?
(WHERE where=booleanExpression)?
;
selectItem
: expression  #selectSingle
;
relation
: left=relation
(
joinType JOIN rightRelation=relation joinCriteria
)                                           #joinRelation
| sampledRelation                             #relationDefault
;
joinType
: INNER?
;
joinCriteria
: ON booleanExpression
;

上述的語(yǔ)法定義將Join的關(guān)鍵要素拆解得非常清晰：Join的左表， Join的類型，Join關(guān)鍵詞， Join的右表， Join的關(guān)聯(lián)條件。例如，通常我們最簡(jiǎn)單的Join語(yǔ)句用例如下(借用presto的tpch數(shù)據(jù)源)：

select t2.custkey, t2.phone, t1.orderkey from orders t1 inner join customer t2 on t1.custkey=t2.custkey limit 10;

對(duì)應(yīng)著語(yǔ)法和SQL語(yǔ)句用例，可以看到在將Join算法落地，還需要考慮如下細(xì)節(jié)點(diǎn)：

• 檢測(cè)SQL語(yǔ)句，確保SQL語(yǔ)句符合語(yǔ)法要求。
• 梳理表的別名和字段的對(duì)應(yīng)關(guān)系，確保查詢的字段和表能夠?qū)?yīng)起來(lái)，Join條件的字段類型能夠匹配。
• Join算法的選取，是HashJoin還是NestedLoopJoin還是SortMergeJoin?  
• 哪個(gè)表是build表，哪個(gè)表是probe表？
• Join條件的判斷如何實(shí)現(xiàn)？
• 整個(gè)查詢涉及到Operator如何組裝，以實(shí)現(xiàn)最終結(jié)果的輸出？

我們回顧一下SQL執(zhí)行的關(guān)鍵流程：

(來(lái)源: Query Execution Flow Architecture (SQL Server))

基于上面的流程，問(wèn)題其實(shí)已經(jīng)有了答案。

• Parser:  借助antlr的能力即可實(shí)現(xiàn)SQL語(yǔ)法的檢測(cè)。
• Binding: 基于SQL語(yǔ)句生成AST，利用元數(shù)據(jù)檢測(cè)字段和表的映射關(guān)系以及Join條件的字段類型。
• Planner: 基于AST生成查詢計(jì)劃。  
• Executor: 基于查詢計(jì)劃生成對(duì)應(yīng)的Operator并執(zhí)行。

以NestedLoop Join算法為例，了解一下Presto的實(shí)現(xiàn)思路。對(duì)于NestedLoopJoin Join算法的落地，在Presto中其實(shí)是拆解為兩個(gè)階段：組合階段和過(guò)濾階段。在實(shí)現(xiàn)JoinOperator時(shí)，只需負(fù)責(zé)兩個(gè)表數(shù)據(jù)的笛卡爾積組合即可。核心代碼如下：

// NestedLoopPageBuilder中實(shí)現(xiàn)兩個(gè)Page計(jì)算笛卡爾積的處理邏輯，這里RunLengthEncodedBlock用于一個(gè)元素復(fù)制，典型地笛卡爾積計(jì)算中需要將一列元素從1行復(fù)制成多行。
@Override
public Page next()
{
if (!hasNext()) {
throw new NoSuchElementException();
}
if (noColumnShortcutResult >= 0) {
    rowIndex = maxRowIndex;
    return new Page(noColumnShortcutResult);
}
rowIndex++;

// Create an array of blocks for all columns in both pages.
Block[] blocks = new Block[numberOfProbeColumns + numberOfBuildColumns];
// Make sure we always put the probe data on the left and build data on the right.
int indexForRleBlocks = buildPageLarger ? 0 : numberOfProbeColumns;
int indexForPageBlocks = buildPageLarger ? numberOfProbeColumns : 0;
// For the page with less rows, create RLE blocks and add them to the blocks array
for (int i = 0; i < smallPage.getChannelCount(); i++) {
    Block block = smallPage.getBlock(i).getSingleValueBlock(rowIndex);
    blocks[indexForRleBlocks] = new RunLengthEncodedBlock(block, largePage.getPositionCount());
    indexForRleBlocks++;
}
// Put the page with more rows in the blocks array
for (int i = 0; i < largePage.getChannelCount(); i++) {
    blocks[indexForPageBlocks + i] = largePage.getBlock(i);
}
return new Page(largePage.getPositionCount(), blocks);
}

五、小結(jié)

本文簡(jiǎn)單梳理了Join的基本算法以及在Presto中實(shí)現(xiàn)的基本框架，并以NestedLoop Join算法為例，演示了在Presto中的實(shí)現(xiàn)核心點(diǎn)?？梢钥闯鱿啾仍嫉乃惴枋?，Presto的工程落地是截然不同: 不僅支持了所有的Join語(yǔ)義，而且實(shí)現(xiàn)了分布式能力。這其中有架構(gòu)層面的思考，也有性能層面的思考，非常值得探索跟研究。就Join算法，可以探索的點(diǎn)還有很多，比如多表Join的順序選取，大表跟小表Join的算法優(yōu)化，Semi Join的算法優(yōu)化，Join算法數(shù)據(jù)傾斜的問(wèn)題等等，可謂路漫漫其修遠(yuǎn)兮，將在后續(xù)系列文章中繼續(xù)分析探索。