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

本次帶來的是系列文章的第2篇，本文梳理了Join的原理，以及Join算法在Presto中的實現思路。通過理論和實踐的結合，可以在理解原理的基礎上，更加深入理解Join算法在OLAP場景下的工程落地技巧，比如火山模型，列式存儲，批量處理等思想的應用。

一、背景

在業務開發中使用數據庫，通常會有規范不允許過多表的Join。例如阿里巴巴開發手冊中，有如下的規定：

【強制】超過三個表禁止Join。需要Join的字段，數據類型必須絕對一致；多表關聯查詢時，保證被關聯的字段需要有索引。說明：即使雙表Join也要注意表索引、SQL性能。

在大數據數倉的建設中，盡管我們有星型結構和雪花結構，但是最終交付業務使用的大多是寬表。

可以看出業務使用數據庫中的一個矛盾點：我們需要Join來提供靈活的關聯操作，但是又要盡量避免多表和大表Join帶來的性能問題。這是為什么呢？

二、Join的基本原理

在數據庫中Join提供的語義是非常豐富的。簡單總結如下：

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

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]
*/

可以看出實現邏輯非常簡單，就是兩個For循環嵌套。

2.1 Nested Loop Join算法

在這個基礎上，實現Inner Join的第一個算法就順其自然了。非常直白的名稱：Nested Loop，，實現關鍵點如下：

（來源：Join Processing in Relational Databases）

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

相比笛卡爾積的實現思路，也就是添加了一層if條件的判斷用于過濾滿足條件的組合。

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

• 小表驅動大表，即數據量較大的集作為于for循環的內部循環。
• 一次處理一個數據塊，而不是一條記錄。也就是所謂的Block Nested Loop Join，通過分塊降低IO次數，提升緩存命中率。

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

2.2 Sort Merge Join算法

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

（來源：Join Processing in Relational Databases）

通過將JOIN操作拆分成Sort和Merge兩個階段實現Join操作的加速。對于Sort階段，是可以提前準備好可以復用的。這樣的思想對于MySQL這類關系型數據庫是非常友好的，這也能解釋阿里巴巴開發手冊中要求關聯的字段必須建立索引，因為索引保證了數據有序。該算法時間復雜度為排序開銷O(mlog(m)+nlog(n))+合并開銷O(m+n)。但是通常由于索引保證了數據有序，索引其時間復雜度為O(m+n)。

2.3 Hash Join算法

Sort Merge Join的思想在落地中有一定的限制。所謂成也蕭何敗蕭何，對于基于Hadoop的數倉而言，保證數據存儲的有序性這個點對于性能影響過大。在海量數據的背景下，維護索引成本是比較大的。而且索引還依賴于使用場景，不可能每個字段都建一個索引。在數據表關聯的場景是大表關聯小表時，比如：用戶表(大表)--當日訂單表(小表)；事實表(大表)–維度表(小表)，可以通過空間換時間。回想一下，在基礎的數據結構中，tree結構和Hash結構可謂數據處理的兩大法寶：一個保證數據有序方便實現區間搜索，一個通過hash函數實現精準命中點對點查詢效率高。

在這樣的背景下，通過將小表Hash化，實現Join的想法也就不足為奇了。

（來源：Join Processing in Relational Databases）

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

通過相關資料的學習，可以發現Join算法的實現原理還是相當簡單的，排序和Hash是數據結構最為基礎的內容。了解了Join的基本思想，如何落地實踐出來呢？畢竟talk is cheap。在項目中實現Join之前，需要一些鋪墊知識。通常來說核心算法是皇冠上的明珠，但是僅有明珠是不夠的還需要皇冠作為底座。

三、Join工程化前置條件

3.1 SQL處理架構-火山模型

在將Join算法落地前，需要先了解一下數據庫處理數據的基本架構。在理解架構的基礎上，才能將Join算法放置到合適的位置。在前面系列文章中探討了基于antlr實現SQL語句的解析。可以發現SQL語法支持的操作類型非常豐富：查詢表(TableScan)，過濾數據(Filter)，排序(Order)，限制(Limit)，字段進行運算(Project)， 聚合(Group)，關聯(Join)等。為了實現上述的能力，需要一個具備并行化能力且可擴展的架構。

1994年Goetz Graefe在論文《Volcano-An Extensible and Parallel Query Evaluation System》提出了一個架構設計思想，這就是大名鼎鼎的火山模型，也稱為迭代模型。火山模型其實包含了文件系統和查詢處理兩個部分，這里我們重點關注查詢處理的設計思想。架構圖如下：

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

簡單解讀一下：

職責分離：將不同操作獨立成一個的Operator，Operator采用open-next-close的迭代器模式。例如對于SQL 。

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

對應到Scan， Select, Project三個Operator，數據交互通過next()函數實現。上述的理論在Presto中可以對應起來，例如Presto中幾個常用的Operator, 基本上是見名知意：

動態組裝：Operator基于SQL語句的解析實現動態組裝，多個Operator形成一個管道(pipeline)。例如：print和predicate兩個operator形成一個管道：

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

在火山模型的基礎上，Presto吸收了數據庫領域的其他思想，對基礎的火山模型進行了優化改造，主要體現在如下幾點：

• Operator數據處理優化成一次一個Page，而不是一次行(也稱為tuple)。
• Page的存儲采用列式結構。即相同的列封裝到一個Block中。

批量處理結合列式存儲奠定了向量化計算的基礎。這也是數據庫領域的優化方向。

3.2 批量處理和列式存儲

在研讀Presto源碼時，幾乎到處都可以看到Page/Block的身影。所以理解Page/Block背后的思想是理解Presto實現機制的基礎。有相關書籍和文檔講解Page/Block的概念，但是由于這些概念是跟其他概念混在一起呈現，導致一時間不容易理解。

筆者認為Type-Block-Page三者放在一起，更容易理解。我們使用數據庫，通常需要定義表，字段名稱，字段類型。在傳統的DBMS中，通常是按行存儲數據，通常結構如下：

(來源：《數據庫系統實現》)

但是通常OLAP場景不需要讀取所有的字段，基于這樣的場景，就衍生出來了列式存儲。就是我們看到的如下結構：

(來源：《Presto技術內幕》)

即每個字段對應一個Block， 多個Block的切面才是一條記錄，也就是所謂的行，在一些論文中稱為tuple。通過對比可以清楚看出Presto中，Page就是典型了列式存儲的實現。所以在Presto中，每個Type必然會關聯到一種Block。例如：bigint類型就對應著LongArrayBlockBuilder，varchar類型對應著VariableWidthBlock。

理解了原理，操作Page/Block就變得非常簡單了，簡單的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) {
/**
* 假設有兩個字段，一個字段類型為int, 一個字段類型為varchar
*/
List types = Lists.newArrayList(BigintType.BIGINT, VarcharType.VARCHAR);
  }
public static void main(String[] args) {
/**
* 假設有兩個字段，一個字段類型為int, 一個字段類型為varchar
*/
List types = Lists.newArrayList(BigintType.BIGINT, VarcharType.VARCHAR);
   // 按行存儲
 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);
  }
}
// 運行結果：
//column[bigint]>>[1, 2, 3, 4]
//column[varchar]>>[aa, ba, cc, dd]

將數據封裝成Page在各個Operator中流轉，一方面避免了對象的序列化和反序列化成本，另一方面相比tuple的方式降低了函數調用的開銷。這跟集裝箱運貨降低運輸成本的思想是類似的。

四、Join算法的工程實踐

理解了Join的核心算法和基礎架構，結合前文中對antlr實現SQL表達式的解析以及實現where條件過濾，我們已經具備了實現Join的基礎條件。接下來簡單講述一下Join算法的落地流程。首先在語法層面需要支持Join的語法，由于本文目的在于研究算法實現流程，而不在于實現完整的Join功能，因此我們暫且先考慮支持兩張表單字段的等值Join語法。

首先在語法上需要支持Join, 基于antlr語法的定義關鍵點如下：

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
;

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

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

對應著語法和SQL語句用例，可以看到在將Join算法落地，還需要考慮如下細節點：

• 檢測SQL語句，確保SQL語句符合語法要求。
• 梳理表的別名和字段的對應關系，確保查詢的字段和表能夠對應起來，Join條件的字段類型能夠匹配。
• Join算法的選取，是HashJoin還是NestedLoopJoin還是SortMergeJoin?  
• 哪個表是build表，哪個表是probe表？
• Join條件的判斷如何實現？
• 整個查詢涉及到Operator如何組裝，以實現最終結果的輸出？

我們回顧一下SQL執行的關鍵流程：

(來源: Query Execution Flow Architecture (SQL Server))

基于上面的流程，問題其實已經有了答案。

• Parser:  借助antlr的能力即可實現SQL語法的檢測。
• Binding: 基于SQL語句生成AST，利用元數據檢測字段和表的映射關系以及Join條件的字段類型。
• Planner: 基于AST生成查詢計劃。  
• Executor: 基于查詢計劃生成對應的Operator并執行。

以NestedLoop Join算法為例，了解一下Presto的實現思路。對于NestedLoopJoin Join算法的落地，在Presto中其實是拆解為兩個階段：組合階段和過濾階段。在實現JoinOperator時，只需負責兩個表數據的笛卡爾積組合即可。核心代碼如下：

// NestedLoopPageBuilder中實現兩個Page計算笛卡爾積的處理邏輯，這里RunLengthEncodedBlock用于一個元素復制，典型地笛卡爾積計算中需要將一列元素從1行復制成多行。
@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);
}

五、小結

本文簡單梳理了Join的基本算法以及在Presto中實現的基本框架，并以NestedLoop Join算法為例，演示了在Presto中的實現核心點。可以看出相比原始的算法描述，Presto的工程落地是截然不同: 不僅支持了所有的Join語義，而且實現了分布式能力。這其中有架構層面的思考，也有性能層面的思考，非常值得探索跟研究。就Join算法，可以探索的點還有很多，比如多表Join的順序選取，大表跟小表Join的算法優化，Semi Join的算法優化，Join算法數據傾斜的問題等等，可謂路漫漫其修遠兮，將在后續系列文章中繼續分析探索。