汪細勖
小米雲平台工程師
負責小米Apache Doris項目的開發和運維
專注于OLAP計算引擎的SQL解析和優化
導讀
本文主要介紹了Doris SQL解析的原理。
重點講述了生成單機邏輯計劃，生成分布式邏輯計劃，生成分布式實體計劃的過程。對應于代碼實作是Analyze，SinglePlan，DistributedPlan，Schedule四個部分。
Analyze負責對AST進行前期的一些處理，SinglePlan根據AST進行優化生成單機查詢計劃，DistributedPlan将單機的查詢計劃拆成分布式的查詢計劃，Schedule階段負責決定查詢計劃下發到哪些機器上執行。
由于SQL類型有很多，本文側重介紹查詢SQL的解析，從算法原理和代碼實作上深入講解了Doris的SQL解析原理。
1 Doris簡介
Doris是基于MPP架構的互動式SQL資料倉庫，主要用于解決近實時的報表和多元分析。
Doris分成兩部分FE和BE，FE 負責存儲以及維護叢集中繼資料、接收、解析、查詢、設計規劃整體查詢流程，BE 負責資料存儲和具體的實施過程。
在 Doris 的存儲引擎中，使用者資料被水準劃分為若幹個資料分片（Tablet，也稱作資料分桶）。每個 Tablet 包含若幹資料行。多個 Tablet 在邏輯上歸屬于不同的分區Partition。一個 Tablet 隻屬于一個 Partition。而一個 Partition 包含若幹個 Tablet。Tablet 是資料移動、複制等操作的最小實體存儲單元。
2 SQL解析簡介
SQL解析在這篇文章中指的是将一條sql語句經過一系列的解析最後生成一個完整的實體執行計劃的過程。
這個過程包括以下四個步驟：詞法分析，文法分析，生成邏輯計劃，生成實體計劃。
如圖1所示：
圖 1 SQL解析的流程
2.1 詞法分析
詞法分析主要負責将字元串形式的sql識别成一個個token，為文法分析做準備。
select ...... from ...... where ....... group by ..... order by ......
SQL 的 Token 可以分為如下幾類：
￮ 關鍵字（select、from、where）
￮ 操作符（+、-、>=）
￮ 開閉合标志（(、CASE）
￮ 占位符（?）
￮ 注釋
￮ 空格
......
2.2 文法分析
文法分析主要負責根據文法規則，将詞法分析生成的token轉成抽象文法樹（Abstract Syntax Tree），如圖2所示。
圖 2 抽象文法樹示例
2.3 邏輯計劃
邏輯計劃負責将抽象文法樹轉成代數關系。代數關系是一棵算子樹，每個節點代表一種對資料的計算方式，整棵樹代表了資料的計算方式以及流動方向，如圖3所示。
圖3 關系代數示例
2.4 實體計劃
實體計劃是在邏輯計劃的基礎上，根據機器的分布，資料的分布，決定去哪些機器上執行哪些計算操作。
Doris系統的SQL解析也是采用這些步驟，隻不過根據Doris系統結構的特點和資料的存儲方式，進行了細化和優化，最大化發揮機器的計算能力。
3 設計目标
Doris SQL解析架構的設計有以下目标：
- 最大化計算的并行性
- 最小化資料的網絡傳輸
- 最大化減少需要掃描的資料
4 總體架構
Doris SQL解析具體包括了五個步驟：詞法分析，文法分析，生成單機邏輯計劃，生成分布式邏輯計劃，生成實體執行計劃。
具體代碼實作上包含以下五個步驟：Parse, Analyze, SinglePlan, DistributedPlan, Schedule。
圖4 系統總體架構圖
如圖4所示，Parse階段本文不詳細講，Analyze負責對AST進行前期的一些處理，SinglePlan根據AST進行優化生成單機查詢計劃，DistributedPlan将單機的查詢計劃拆成分布式的查詢計劃，Schedule階段負責決定查詢計劃下發到哪些機器上執行。
由于SQL類型有很多，本文側重介紹查詢SQL的解析。
圖5展示了一個簡單的查詢SQL在Doris的解析實作。
圖 5 查詢sql在Doris中的解析過程
5 Parse階段
詞法分析采用jflex技術，文法分析采用java cup parser技術，最後生成抽象文法樹（Abstract Syntax Tree）AST，這些都是現有的、成熟的技術，在這裡不進行詳細介紹。
AST是一種樹狀結構，代表着一條SQL。不同類型的查詢select, insert, show, set, alter table, create table等經過Parse階段後生成不同的資料結構（SelectStmt, InsertStmt, ShowStmt, SetStmt, AlterStmt, AlterTableStmt, CreateTableStmt等），但他們都繼承自Statement，并根據自己的文法規則進行一些特定的處理。例如：對于select類型的sql， Parse之後生成了SelectStmt結構。
SelectStmt結構包含了SelectList，FromClause，WhereClause，GroupByClause，SortInfo等結構。這些結構又包含了更基礎的一些資料結構，如WhereClause包含了BetweenPredicate（between表達式）, BinaryPredicate（二進制表達式）， CompoundPredicate（and or組合表達式）, InPredicate（in表達式）等。
AST中所有結構都是由基本結構表達式Expr通過多種組合而成，如圖6所示。
圖 6 Doris中抽象文法樹AST的實作
6 Analyze階段
Analyze主要是對Parse階段生成的抽象文法樹AST進行一些前期的處理和語義分析，為生成單機邏輯計劃做準備。
抽象文法樹是由StatementBase這個抽象類表示。這個抽象類包含一個最重要的成員函數analyze()，用來執行Analyze階段要做的事。
不同類型的查詢select, insert, show, set, alter table, create table等經過Parse階段後生成不同的資料結構（SelectStmt, InsertStmt, ShowStmt, SetStmt, AlterStmt, AlterTableStmt, CreateTableStmt等），這些資料結構繼承自StatementBase，并實作analyze()函數，對特定類型的SQL進行特定的Analyze。
例如：select類型的查詢，會轉成對select sql的子語句SelectList, FromClause, GroupByClause, HavingClause, WhereClause, SortInfo等的analyze()。然後這些子語句再各自對自己的子結構進行進一步的analyze()，通過層層疊代，把各種類型的sql的各種情景都分析完畢。例如：WhereClause進一步分析其包含的BetweenPredicate（between表達式）, BinaryPredicate（二進制表達式）， CompoundPredicate（and or組合表達式）, InPredicate（in表達式）等。
對于查詢類型的SQL，包含以下幾項重要工作：
· 元資訊的識别和解析：識别和解析sql中涉及的 Cluster, Database, Table, Column 等元資訊，确定需要對哪個叢集的哪個資料庫的哪些表的哪些列進行計算。
· SQL 的合法性檢查：視窗函數不能 DISTINCT，投影列是否有歧義，where語句中不能含有grouping操作等。
· SQL 簡單重寫：比如将 select * 擴充成 select 所有列，count distinct轉成bitmap或者hll函數等。
· 函數處理：檢查sql中包含的函數和系統定義的函數是否一緻，包括參數類型，參數個數等。
· Table 和 Column 的别名處理
· 類型檢查和轉換：例如二進制表達式兩邊的類型不一緻時，需要對其中一個類型進行轉換（BIGINT 和 DECIMAL 比較，BIGINT 類型需要 Cast 成 DECIMAL）。
對AST 進行analyze後，會再進行一次rewrite操作，進行精簡或者是轉成統一的處理方式。目前rewrite的算法是基于規則的方式，針對AST的樹狀結構，自底向上，應用每一條規則進行重寫。如果重寫後，AST有變化，則再次進行analyze和rewrite，直到AST無變化為止。
例如：常量表達式的化簡：1 + 1 + 1 重寫成 3，1 > 2 重寫成 Flase 等。将一些語句轉成統一的處理方式，比如将 where in, where exists 重寫成 semi join, where not in, where not exists 重寫成 anti join。
7 生成單機邏輯Plan階段
這部分工作主要是根據AST抽象文法樹生成代數關系，也就是俗稱的算子數。樹上的每個節點都是一個算子，代表着一種操作。
如圖7所示，ScanNode代表着對一個表的掃描操作，将一個表的資料讀出來。HashJoinNode代表着join操作，小表在記憶體中建構哈希表，周遊大表找到連接配接鍵相同的值。Project表示投影操作，代表着最後需要輸出的列，圖7表示隻用輸出citycode這一列。
圖7 單機邏輯計劃示例
如果不進行優化，生成的關系代數下發到存儲中執行的代價非常高。
對于查詢：
select a.siteid, a.pv from table1 a join table2 b on a.siteid = b.siteid where a.citycode=122216 and b.username="test" order by a.pv limit 10
未優化的關系代數，如圖8所示，需要将所有列讀出來進行一系列的計算，在最後選擇輸出siteid, pv兩列，大量無用的列資料浪費了計算資源。
Doris在生成代數關系時，進行了大量的優化，将投影列和查詢條件盡可能放到掃描操作時執行。
圖8 未優化的關系代數
具體來說這個階段主要做了如下幾項工作：
· Slot 物化：指确定一個表達式對應的列需要 Scan 和計算，比如聚合節點的聚合函數表達式和 Group By 表達式需要進行物化。
· 投影下推：BE 在 Scan 時隻會 Scan 必須讀取的列。
· 謂詞下推：在滿足語義正确的前提下将過濾條件盡可能下推到 Scan 節點。
· 分區，分桶裁剪：根據過濾條件中的資訊，确定需要掃描哪些分區，哪些桶的tablet。
· Join Reorder：對于 Inner Join, Doris 會根據行數調整表的順序，将大表放在前面。
· Sort + Limit 優化成 TopN：對于order by limit語句會轉換成TopN的操作節點，友善統一處理。
· MaterializedView 選擇：會根據查詢需要的列，過濾，排序和 Join 的列，行數，列數等因素選擇最佳的物化視圖。
圖9展示了優化的示例，Doris是在生成關系代數的過程中優化，邊生成邊優化。
圖 9 單機查詢計劃優化的過程
8 生成分布式Plan階段
有了單機的PlanNode樹之後，就需要進一步根據分布式環境，拆成分布式PlanFragment樹（PlanFragment用來表示獨立的執行單元），畢竟一個表的資料分散地存儲在多台主機上，完全可以讓一些計算并行起來。
這個步驟的主要目标是最大化并行度和資料本地化。主要方法是将能夠并行執行的節點拆分出去單獨建立一個PlanFragment，用ExchangeNode代替被拆分出去的節點，用來接收資料。拆分出去的節點增加一個DataSinkNode，用來将計算之後的資料傳送到ExchangeNode中，做進一步的處理。
這一步采用遞歸的方法，自底向上，周遊整個PlanNode樹，然後給樹上的每個葉子節點建立一個PlanFragment，如果碰到父節點，則考慮将其中能夠并行執行的子節點拆分出去，父節點和保留下來的子節點組成一個parent PlanFragment。拆分出去的子節點增加一個父節點DataSinkNode組成一個child PlanFragment，child PlanFragment指向parent PlanFragment。這樣就确定了資料的流動方向。
對于查詢操作來說，join操作是最常見的一種操作。
Doris目前支援4種join算法：broadcast join，hash partition join，colocate join，bucket shuffle join。
broadcast join：将小表發送到大表所在的每台機器，然後進行hash join操作。當一個表掃描出的資料量較少時，計算broadcast join的cost，通過計算比較hash partition的cost，來選擇cost最小的方式。
hash partition join：當兩張表掃描出的資料都很大時，一般采用hash partition join。它周遊表中的所有資料，計算key的哈希值，然後對叢集數取模，選到哪台機器，就将資料發送到這台機器進行hash join操作。
colocate join：兩個表在建立的時候就指定了資料分布保持一緻，那麼當兩個表的join key與分桶的key一緻時，就會采用colocate join算法。由于兩個表的資料分布是一樣的，那麼hash join操作就相當于在本地，不涉及到資料的傳輸，極大提高查詢性能。
bucket shuffle join：當join key是分桶key，并且隻涉及到一個分區時，就會優先采用bucket shuffle join算法。由于分桶本身就代表了資料的一種切分方式，是以可以利用這一特點，隻需将右表對左表的分桶數hash取模，這樣隻需網絡傳輸一份右表資料，極大減少了資料的網絡傳輸，如圖10所示。
圖 10 bucket shuffle join示例
如圖11展示了帶有HashJoinNode的單機邏輯計劃建立分布式邏輯計劃的核心流程。
· 對PlanNode，自底向上建立PlanFragment。
· 如果是ScanNode，則直接建立一個PlanFragment，PlanFragment的RootPlanNode是這個ScanNode。
· 如果是HashJoinNode，則首先計算下broadcastCost，為選擇boracast join還是hash partition join提供參考。
· 根據不同的條件判斷選擇哪種Join算法
· 如果使用colocate join，由于join操作都在本地，就不需要拆分。設定HashJoinNode的左子節點為leftFragment的RootPlanNode，右子節點為rightFragment的RootPlanNode，與leftFragment共用一個PlanFragment，删除掉rightFragment。
· 如果使用bucket shuffle join，需要将右表的資料發送給左表。是以先建立了一個ExchangeNode，設定HashJoinNode的左子節點為leftFragment的RootPlanNode，右子節點為這個ExchangeNode，與leftFragment共用一個PlanFragment，并且指定rightFragment資料發送的目的地為這個ExchangeNode。
· 如果使用broadcast join，需要将右表的資料發送給左表。是以先建立了一個ExchangeNode，設定HashJoinNode的左子節點為leftFragment的RootPlanNode，右子節點為這個ExchangeNode，與leftFragment共用一個PlanFragment，并且指定rightFragment資料發送的目的地為這個ExchangeNode。
· 如果使用hash partition join，左表和右邊的資料都要切分，需要将左右節點都拆分出去，分别建立left ExchangeNode, right ExchangeNode，HashJoinNode指定左右節點為left ExchangeNode和 right ExchangeNode。單獨建立一個PlanFragment，指定RootPlanNode為這個HashJoinNode。最後指定leftFragment, rightFragment的資料發送目的地為left ExchangeNode, right ExchangeNode。
圖 11 HashJoinNode建立分布式邏輯計劃核心流程
圖12是兩個表的join操作轉換成PlanFragment樹之後的示例，一共生成了3個PlanFragment。最終資料的輸出通過ResultSinkNode節點。
圖 12 從單機計劃到分布式計劃
9 Schedule階段
這一步是根據分布式邏輯計劃，建立分布式實體計劃。主要解決以下問題：
- 哪個 BE 執行哪個 PlanFragment
- 每個 Tablet 選擇哪個副本去查詢
- 如何進行多執行個體并發
圖13展示了建立分布式實體計劃的核心流程：
a. prepare階段：給每個PlanFragment建立一個FragmentExecParams結構，用來表示PlanFragment執行時所需的所有參數；如果一個PlanFragment包含有DataSinkNode，則找到資料發送的目的PlanFragment，然後指定目的PlanFragment的FragmentExecParams的輸入為該PlanFragment的FragmentExecParams。
b. computeScanRangeAssignment階段：針對不同類型的join進行不同的處理。
- computeScanRangeAssignmentByColocate：針對colocate join進行處理，由于join的兩個表桶中的資料分布都是一樣的，他們是基于桶的join操作，是以在這裡是确定每個桶選擇哪個host。在給host配置設定桶時，盡量保證每個host配置設定到的桶基本平均。
- computeScanRangeAssignmentByBucket：針對bucket shuffle join進行處理，也隻是基于桶的操作，是以在這裡是确定每個桶選擇哪個host。在給host配置設定桶時，同樣需要盡量保證每個host配置設定到的桶基本平均。
- computeScanRangeAssignmentByScheduler：針對其他類型的join進行處理。确定每個scanNode讀取tablet哪個副本。一個scanNode會讀取多個tablet，每個tablet有多個副本。為了使scan操作盡可能分散到多台機器上執行，提高并發性能，減少IO壓力，Doris采用了Round-Robin算法，使tablet的掃描盡可能地分散到多台機器上去。例如100個tablet需要掃描，每個tablet 3個副本，一共10台機器，在配置設定時，保障每台機器掃描10個tablet。
c. computeFragmentExecParams階段：這個階段解決PlanFragment下發到哪個BE上執行，以及如何處理執行個體并發問題。确定了每個tablet的掃描位址之後，就可以以位址為次元，将FragmentExecParams生成多個執行個體，也就是FragmentExecParams中包含的位址有多個，就生成多個執行個體FInstanceExecParam。如果設定了并發度，那麼一個位址的執行執行個體再進一步的拆成多個FInstanceExecParam。針對bucket shuffle join和colocate join會有一些特殊處理，但是基本思想一樣。FInstanceExecParam建立完成後，會配置設定一個唯一的ID，友善追蹤資訊。如果FragmentExecParams中包含有ExchangeNode，需要計算有多少senders，以便知道需要接受多少個發送方的資料。最後FragmentExecParams确定destinations，并把目的位址填充上去。
d. create result receiver階段：result receiver是查詢完成後，最終資料需要輸出的地方。
e. to thrift階段：根據所有PlanFragment的FInstanceExecParam建立rpc請求，然後下發到BE端執行。這樣一個完整的SQL解析過程完成了。
圖 13 建立分布式實體計劃核心流程
如圖14所示是一個簡單示例，圖中的PlanFrament包含了一個ScanNode，ScanNode掃描3個tablet，每個tablet有2副本，叢集假設有2台host。
computeScanRangeAssignment階段确定了需要掃描replica 1,3,5,8,10,12，其中replica 1,3,5位于host1上，replica 8,10,12位于host2上。
如果全局并發度設定為1時，則建立2個執行個體FInstanceExecParam，下發到host1和host2上去執行，如果如果全局并發度設定為3，這個host1上建立3個執行個體FInstanceExecParam，host2上建立3個執行個體FInstanceExecParam，每個執行個體掃描一個replica，相當于發起6個rpc請求。
圖 14 生成實體計劃的過程
10 總結
本文首先簡單介紹了Doris，然後介紹SQL解析的通用流程：詞法分析，文法分析，生成邏輯計劃，生成實體計劃，接着從總體上介紹了Doris在SQL解析這塊的總體架構，最後詳細講解了Parse，Analyze，SinglePlan，DistributedPlan，Schedule等5個過程，從算法原理和代碼實作上進行了深入的講解。
Doris遵守了SQL解析的常用方法，但根據底層存儲架構，以及分布式的特點，在SQL解析這塊進行了大量的優化，實作了最大并行度和最小化網絡傳輸，給SQL執行層面減少很多負擔。
百度資料倉庫Palo
基于 Apache Doris 的企業級資料倉庫托管服務
全新UI支援，更有新使用者0元試用3個月優惠活動
登陸百度智能雲官網搜尋Palo，馬上試用！
????百度 Palo/Doris 團隊，誠邀對開源軟體、分布式資料庫感興趣的小夥伴們
我們虛位以待！
履歷發送至：[email protected]
歡迎掃碼關注：
Apache Doris(incubating)官方公衆号
相關連結：
Apache Doris官方網站：
http://doris.incubator.apache.org
Apache Doris Github：
https://github.com/apache/incubator-doris
Apache Doris 開發者郵件組：
[email protected]
識别下方二維碼，回複“峰會”，即可獲得下載下傳位址。感覺幹貨多，記得設為星标哦