Hologres（中文名交互式分析）是阿里雲自研的一站式實時數倉，這個雲原生系統融合了實時服務和分析大數據的場景，全面兼容PostgreSQL協議並與大數據生態無縫打通，能用同一套數據架構同時支持實時寫入實時查詢以及實時離線聯邦分析。它的出現簡化了業務的架構，與此同時為業務提供實時決策的能力，讓大數據發揮出更大的商業價值。從阿里集團誕生到雲上商業化，隨著業務的發展和技術的演進，Hologres也在持續不斷優化核心技術競爭力，為了讓大家更加了解Hologres，我們計劃持續推出Hologers底層技術原理揭祕系列，從高性能存儲引擎到高效率查詢引擎，高吞吐寫入到高QPS查詢等，全方位解讀Hologers，請大家持續關注！
往期精彩內容：
- 2020年VLDB的論文《Alibaba Hologres: A cloud-Native Service for Hybrid Serving/Analytical Processing》
- Hologres揭祕：首次公開！阿里巴巴雲原生實時數倉核心技術揭祕
- Hologres揭祕：首次揭祕雲原生Hologres存儲引擎
本期我們將帶來Hologers高效率分佈式查詢引擎的技術原理解析。
Hologres作為HSAP服務分析一體化的落地最佳實踐，其查詢引擎是一個完全自研的執行引擎，它的核心設計目標是支持所有類型的分佈式分析和服務查詢，並做到極致查詢性能。為了做到這一點，我們借鑑了各種分佈式查詢系統，包括分析型數據庫，實時數倉等，吸取了各方面的優勢從零開始打造出一個全新的執行引擎。
為什麼要選擇從零開始做一個新的查詢引擎？開源的分佈式分析查詢系統主要有兩大類：
- 一類是傳統的 Massively Parallel Processing 系統，能夠支持通用的 SQL 查詢，但是對實時場景支持不夠好，性能不夠理想。
- 一類是 Apache Druid 和 ClickHouse這些實時數倉，是專門為實時場景設計和優化的，能夠比較好地支持一些常見的單表實時查詢，但是複雜查詢的性能比較差。
- 另外大數據生態圈基於 MapReduce 的引擎比較適合批處理 ETL，一般不太適合在線服務和多維分析的場景，性能也差不少。
Hologres 執行引擎是在一個能支持複雜查詢和上述高性能實時服務查詢的通用架構，先首先實現了常用的實時數倉場景，深入優化並用內部 Benchmark 驗證了性能和穩定性超過包括專用實時數倉的其它競品之後，再擴展到其它複雜查詢的支持。擴展的過程中，在不可避免地系統變得越來越複雜的同時，也用 Benchmark 幫助保持簡單實時查詢的性能沒有回退。如果在已有的查詢引擎上做改進，因為很多架構和設計上的選擇已經定型，牽一髮而動全身，就很難達到這樣的效果。
Hologres執行引擎從開發到落地實踐面臨了非常多的挑戰，但也給我們提供了機會把這個領域的各種新進展都結合利用起來，並超越已有系統做到對各種查詢類型的高性能處理，其背後主要是基於以下特點：
- 分佈式執行模型：一個和存儲計算分離架構配合的分佈式執行模型。執行計劃由異步算子組成的執行圖 DAG（有向無環圖） 表示，可以表達各種複雜查詢，並且完美適配 Hologres 的數據存儲模型，方便對接查詢優化器，利用業界各種查詢優化技術。
- 全異步執行：端到端的全異步處理框架，可以避免高併發系統的瓶頸，充分利用資源，並且最大可能地避免存儲計算分離系統帶來的讀數據延遲的影響。
- 向量化和列處理：算子內部處理數據時最大可能地使用向量化執行，和存儲引擎的深度集成，通過靈活的執行模型，充分利用各種索引，並且最大化地延遲向量物化和延遲計算，避免不必要的讀數據和計算。
- 自適應增量處理：對常見實時數據應用查詢模式的自適應增量處理。
- 特定查詢深度優化：對一些查詢模式的獨特優化
下面將會對各個模塊一一介紹。
分佈式執行模型
Hologres 是能夠彈性無限水平擴展數據量和計算能力的系統，需要能夠支持高效的分佈式查詢。
Hologres 查詢引擎執行的是由優化器生成的分佈式執行計劃。執行計劃由算子組成。因為 Hologres 的一個表的數據會根據 Distribution Key 分佈在多個 Shard 上，每個 Shard 內又可以包含很多 Segment，執行計劃也會反映這樣的結構，並分佈到數據所在的節點去執行。每個Table Shard 會被加載到一個計算節點，數據會被緩存到這個節點的內存和本地存儲。因為是存儲計算分離的架構，如果一個節點出錯，其服務的 Shard 可以被重新加載到任意一個計算節點，只是相當於清空了緩存。
例如一個比較簡單的查詢。
select key, count(value) as total from table1 group by key order by total desc limit 100。
如果是單機數據庫，可以用這樣的執行計劃。如果數據和計算分佈在多個節點上，就需要更復雜的執行計劃。
在分佈式表上，為了更高效地執行，儘量減少數據傳輸，可以把執行計劃分為不同片段（Fragment）分佈到相應節點執行，並且把一些操作下推來減少 Fragment 輸出的數據，可能就變成這樣的執行計劃：
根據數據的特性，優化器可能會生成不同的計劃。例如在某一個局部聚合並沒有顯著減少數據量的時候，可以省略這個算子。又例如在 Key 就是 Distribution key 的時候，可以優化為：
從這些例子可以看出，Hologres 的執行計劃根據數據的特性切分為不同的片段之後分佈式併發執行。片段之間通過 Exchange 算子進行數據交換。更復雜的比如多表關聯（Join）查詢，會有更多的片段和更復雜的數據交換模式。
比如以下SQL
select user_name, sum(value) as total from t1 join t2 on t1.user_id = t2.user_id where … group by user_name order by total limit 100
在Hologres中可以是這樣的執行計劃
如果 Join key 和 Distribution Key 一致，可以優化為如下執行計劃，減少遠程數據傳輸。根據需要的查詢合理地設置 Distribution Key，可能顯著提高查詢性能。
根據過濾條件和統計信息等等，優化器還可能生成不同的優化執行計劃，比如包含動態過濾，局部聚合等等。
這樣的分佈式執行計劃足夠通用，可以表達所有的 SQL 查詢和一些其它查詢。執行計劃和大部分 Massively Parallel Processing (MPP) 系統也比較類似，方便借鑑和集成業界的一些適用的優化。稍微獨特一些的地方是很多查詢計劃片段的實例是和 Hologres 的存儲結構對齊的，能夠進行高效的分區裁剪和文件裁剪。
同時，Hologres 實現了 PostgreSQL 的 Explain 和 Explain Analyze 系列語句，可以展示文本格式的執行計劃和相應的執行信息，方便用戶自助瞭解執行計劃，並針對性做出SQL優化調整。
全異步執行
高併發系統，特別是有大量 I/O 的系統，頻繁地等待或者任務切換是常見的系統瓶頸。異步處理是一種已經被證明行之有效的避免這些瓶頸，並把高併發系統性能推到極致的方法。
Hologres 的整個後端，包括執行引擎、存儲引擎和其它組件，統一使用 HOS（Hologres Operation System） 組件提供的異步無鎖編程框架，能夠最大化異步執行的效果。每個 Fragment 的實例使用 HOS 的一個 EC （邏輯調度單位），使得一個 Fragment 裡的所有算子和存儲引擎可以異步執行並且無鎖安全訪問絕大多數資源。
算子和 Fragment 都是類似這樣的接口：
future<> Open(const SeekParameters& parameters, ...) future<RecordBatchPtr, bool> GetNext(...) future<> Close(...)
除了一般異步處理的好處外，異步算子接口較好地規避了存儲計算分離架構下相對較高的讀數據延遲對查詢性能的影響，並且對分佈式查詢的執行模型本身也有獨特的好處。
DAG 執行引擎一般可以分為拉數據的模性（比如火山模型）和推的模型（比如很多大數據的分階段執行模型），各有其優缺點。而 Hologres採用的異步的拉模型能夠取得兩種模型的好處並且避免其缺點（已經申請了專利）。舉一個常見的 Hash Join 來說明：
火山模型可以簡單做到先拉完 b 的數據構建 hash table，然後流式處理 a 的數據不用全放在內存裡。但是當 a 或者 b 需要讀數據的時候，簡單的實現需要等待不能把 CPU 打滿，需要通過提高 Fragment 的併發數或者引入複雜的 pre-fetch 機制來充分利用資源，而這些又會引入別的性能問題。
推數據的模型，比較容易做到併發讀數據請求並在完成的時候觸發下游處理，但是上述 Join算子的實現會比較複雜。比如 a 處理完一批數據推到 Join 算子而 b 的 hash table 還沒有構建完成，這批數據就需要暫存到內存裡或者盤上，或者引入反壓機制。在 Fragment 的邊界也會有類似問題，造成一些在拉數據模型下不需要的數據緩存。
Hologres 的算子和 Fragment 的異步拉數據模型，可以像火山模型一樣簡單做到按需從上游獲取數據，而同時又可以像推數據模型一樣簡單做到讀數據併發，只要向上遊發出多個異步 GetNext，上游處理完成時會自然觸發後續處理。異步 GetNext 的數目和時機，可以看做是天然的流控機制，可以有效做到提高 CPU 利用率並且避免不必要的數據暫存。
Hologres 已經用這個異步模型實現了一個完整的查詢引擎，可以支持所有 PostgreSQL 的查詢。
列處理和向量化
按列處理和向量化執行都是分析查詢引擎常用的優化機制，可以大幅度提高數據處理的效率。Hologres 也不例外，在能使用向量處理的時候儘量使用。
Hologres 在內存裡也採用列式存儲。在內存裡按列存儲數據能夠使用更多的向量處理。列式組織數據還有一個好處，就是對延遲計算比較友好。比如 select … where a = 1 and b = 2 …，對一批數據（一般對應存儲的一個 row group），Hologres的 scan 算子輸出的 a 和 b 可以是延遲讀取的 a 和 b 的信息，在處理 a = 1 的時候會讀取這一批的 a。如果 a=1 對這一批的所有行都不滿足，這一批的 b 這一列就根本不會被讀取。
但是對某些按行處理的算子，比如 Join，按列存儲的數據可能會造成更多的 CPU cache miss ，帶來較大的性能問題。很多查詢引擎會在不同的點引入按列存儲和按行存儲的轉換，但是頻繁的轉換本身會帶來不小的開銷，而且列轉行會造成上述延遲讀取列被不必要地讀取，還有一些其它的性能問題。
自適應增量處理
很多實時數據應用經常會對一個查詢用不同的時間段反覆執行。比如一個監控指標頁面打開後，會定期執行 select avg(v1) from metrics where d1 = x and d2 = y and ts >= '2020-11-11 00:00:00' and ts < '2020-11-11 03:01:05' and … group by d3 … 這樣的查詢，下一次會改成 ts < '2020-11-11 00:03:10'，再下一次 ts < '2020-11-11 00:03:15'。
流計算或者增量計算可以對這種查詢進行非常高效的處理。但是對這種用戶可以隨意生成的交互式查詢，通常不可能對所有組合都配置流計算或者增量計算任務。如果每次都簡單執行查詢，又可能有大量的重複計算造成資源浪費和性能不理想。
Hologres充分利用存儲引擎和計算引擎的深度集成和列式存儲大部分數據在只讀文件中的特性，在能提供包含最新寫入數據的查詢結果的同時儘量避免重複計算，對這種類型的查詢能夠顯著提升性能和減少資源使用。
針對特定查詢模式的深度優化
Hologres 對一些特定查詢模式有獨特的優化。這裡以Filter Aggregate 優化為例子。
很多數據應用都有開放列的需求，相當於可以動態添加邏輯列而不用改 Table Schema。比如有一列是多值列 tags（Postgres 可以用 Array 類型）裡面存了'{c1:v1, c2:u1}' 這樣的多個邏輯列的值。查詢的時候，如果使用普通列，一類常見的查詢是
-- Q1: select c1, sum(x) from t1 where c1 in (v1, v2, v3) and name = 'abc' group by c1
使用開放列後，這樣的查詢會轉變為
-- Q2: select unnest(tags), sum(x) from t1 where name = 'abc' and tags && ARRAY['c1:v1', 'c1:v2', c1:v3'] group by unnest(tags) having unnest(tags) in ('c1:v1', 'c1:v2', c1:v3')
這種查詢，Hologres 可以利用位圖索引快速計算過濾條件得到相關的行，但是之後從多值列裡面取出相關數據操作不能使用向量處理，性能不能達到最優。經過調研，可以把查詢的執行轉換為
這樣每個 UNION ALL 分支可以只讀取 name 和 tags 的位圖索引計算過濾條件，然後用 x 列的數據和過濾條件進行向量計算 SUM_IF 即可得出想要的結果。這樣的問題是，每個分支都要過一遍 t1，讀取 x 列以及 name 列的位圖索引，帶來重複計算。最後引入了一個 filter aggregate 的特殊算子來把這類常用查詢優化到極致性能，可以只過一遍 t1 並且去掉重複操作，只用向量計算即可得到結果，不需要讀取 tags 列的數據。在一個幾十 TB的表上實測性能提升 3 倍以上。
類似的優化，Hologres 的執行引擎都會盡量抽象為比較通用的算子，可以適用於更多場景。Filter Aggregate 算子也是 Hologres 申請的專利之一。
總結
Hologres 執行引擎在一個架構裡集中了相關分佈式查詢系統的幾乎所有最高效的優化方式（包括各種類型的索引）並作出了特有的改進。通過和存儲引擎深度整合，能充分發揮異步模型的優勢，並高效利用各種類型的索引來加速查詢。所有這些加起來，帶來了超越已有系統的性能，並在阿里巴巴雙 11 的數據規模下通過了實戰的考驗，（2020年雙11頂住了5.96億/秒的實時數據洪峰，基於萬億級數據對外提供多維分析和服務，99.99%的查詢可以在80ms以內返回結果），對外高併發高性能地提供分佈式 HSAP 查詢服務。
後續我們將會陸續推出有關Hologres的技術底層原理揭祕系列，具體規劃如下，敬請持續關注！
- Hologres揭祕：首次公開！阿里巴巴雲原生實時數倉核心技術揭祕
- Hologres揭祕：首次揭祕雲原生Hologres存儲引擎
- Hologres揭祕：深度解析高效率分佈式查詢引擎（本文）
- Hologres揭祕：透明加速MaxCompute查詢核心原理
- Hologres揭祕：如何實現MaxCompute與Hologres數據同步速度快百倍
- Hologres揭祕：如何支持高吞吐Upsert
- Hologres揭祕：如何支持在線服務場景的超高QPS
- Hologres揭祕：如何支持高併發查詢
- Hologres揭祕：如何支持高可用架構
- Hologres揭祕：如何支持資源隔離，支持多種負載
- Hologres揭祕：向量檢索引擎Proxima原理與使用實踐
- Hologres揭祕：讀懂執行計劃，查詢性能翻十倍
- Hologres揭祕：分佈式系統如何設計Shard與Table Group
- Hologres揭祕：如何支持更多Postgres生態擴展包
- Hologres揭祕：高吞吐寫入Hologres的N種姿勢
- ......