1.3.2 查詢引擎

Doris的查詢引擎是基于MPP框架的火山模型，是從早期的Apache Impala演化而來的。Doris會基于SQL語句先生成一個邏輯執行計劃，然后根據數據的分布，形成一個物理執行計劃。物理執行計劃涉及多個Fragment，Fragment之間的數據傳輸則是由Exchange模塊完成的。通過Exchange模塊，Doris在執行查詢任務時就有了數據重分布（Reshuffle）能力，讓查詢不再局限于數據存儲節點，從而更好地利用多節點資源并行進行數據處理。基于MPP框架的查詢引擎執行流程示意圖如圖1-12所示。

圖1-12 基于MPP框架的查詢引擎執行流程示意圖

邏輯執行計劃的Agg階段分為先重分布數據再匯總兩個步驟，這個過程和Hadoop類似，都是按照相同的Key進行數據重分布。

除了通過并行設計來提高查詢效率外，Doris還對很多具體的查詢算子進行了優化，比如圖1-13中的聚合算子。

圖1-13 聚合算子

在Doris中，聚合算子被拆分成兩級聚合：第一級聚合是在數據所在節點，以減少第二級聚合的數據；而第二級聚合將Key相同的數據匯聚到同一個節點，進行最終的聚合。

在此基礎上，Doris還實現了自適應聚合。首先我們要知道，聚合算子是一種阻塞型算子，需要等全部數據處理完后，才會將數據發送給上層節點。而自適應聚合是在第一級聚合中，如果發現聚合效果很差，即使聚合后也無法有效減少需要傳輸的數據，則會自動停止第一級聚合，轉換為非阻塞的流式算子，直接將讀取的數據發送到上層節點，從而避免不必要的阻塞等待。

針對Join算子，Doris也進行了大量優化，其中Runtime Filter是一種很重要的優化方式。在兩個表的Join操作中，我們通常將右表稱為BuildTable，將左表稱為ProbeTable。在實現上，通常首先讀取右表中的數據，在內存中構建一個HashTable，然后開始讀取左表中的每一行數據，并在HashTable中進行連接匹配，返回符合連接條件的數據。通常來說，左表的數據量會大于右表的數據量。

Runtime Filter的設計思路是在右表內存中構建HashTable的同時，為連接列生成一個過濾器，之后把這個過濾器推給左表。這樣，左表就可以利用過濾器對數據進行過濾，從而減少Probe節點需要傳輸和比對的數據。這種過濾器被稱為Runtime Filter。針對不同的數據，Doris設計了不同類型的過濾器，例如In Predicate、Bloom Filter和Min Max。用戶可以根據不同場景選擇不同的過濾器。Runtime Filter實現邏輯示意圖如圖1-14所示。

Runtime Filter適用于大部分Join場景，包括節點的自動穿透，可將過濾器下推到最底層的掃描節點，例如分布式Shuffle Join中，可先將多個節點產生的過濾器進行合并，再下推到數據讀取節點。

圖1-14 Runtime Filter實現邏輯示意圖