Hologres（中文名交互式分析）是阿里云自研的一站式實(shí)時(shí)數(shù)倉，這個(gè)云原生系統(tǒng)融合了實(shí)時(shí)服務(wù)和分析大數(shù)據(jù)的場(chǎng)景，全面兼容PostgreSQL協(xié)議并與大數(shù)據(jù)生態(tài)無縫打通，能用同一套數(shù)據(jù)架構(gòu)同時(shí)支持實(shí)時(shí)寫入實(shí)時(shí)查詢以及實(shí)時(shí)離線聯(lián)邦分析。它的出現(xiàn)簡化了業(yè)務(wù)的架構(gòu)，與此同時(shí)為業(yè)務(wù)提供實(shí)時(shí)決策的能力，讓大數(shù)據(jù)發(fā)揮出更大的商業(yè)價(jià)值。從阿里集團(tuán)誕生到云上商業(yè)化，隨著業(yè)務(wù)的發(fā)展和技術(shù)的演進(jìn)，Hologres也在持續(xù)不斷優(yōu)化核心技術(shù)競(jìng)爭力，為了讓大家更加了解Hologres，我們計(jì)劃持續(xù)推出Hologers底層技術(shù)原理揭秘系列，從高性能存儲(chǔ)引擎到高效率查詢引擎，高吞吐寫入到高QPS查詢等，全方位解讀Hologers，請(qǐng)大家持續(xù)關(guān)注！

本期我們將帶來Hologers高效率分布式查詢引擎的技術(shù)原理解析。

Hologres作為HSAP服務(wù)分析一體化的落地最佳實(shí)踐，其查詢引擎是一個(gè)完全自研的執(zhí)行引擎，它的核心設(shè)計(jì)目標(biāo)是支持所有類型的分布式分析和服務(wù)查詢，并做到極致查詢性能。為了做到這一點(diǎn)，我們借鑒了各種分布式查詢系統(tǒng)，包括分析型數(shù)據(jù)庫，實(shí)時(shí)數(shù)倉等，吸取了各方面的優(yōu)勢(shì)從零開始打造出一個(gè)全新的執(zhí)行引擎。

為什么要選擇從零開始做一個(gè)新的查詢引擎？開源的分布式分析查詢系統(tǒng)主要有兩大類：

一類是傳統(tǒng)的 Massively Parallel Processing 系統(tǒng)，能夠支持通用的 SQL 查詢，但是對(duì)實(shí)時(shí)場(chǎng)景支持不夠好，性能不夠理想。
一類是 Apache Druid 和 ClickHouse這些實(shí)時(shí)數(shù)倉，是專門為實(shí)時(shí)場(chǎng)景設(shè)計(jì)和優(yōu)化的，能夠比較好地支持一些常見的單表實(shí)時(shí)查詢，但是復(fù)雜查詢的性能比較差。
另外大數(shù)據(jù)生態(tài)圈基于 MapReduce 的引擎比較適合批處理 ETL，一般不太適合在線服務(wù)和多維分析的場(chǎng)景，性能也差不少。
Hologres 執(zhí)行引擎是在一個(gè)能支持復(fù)雜查詢和上述高性能實(shí)時(shí)服務(wù)查詢的通用架構(gòu)，先首先實(shí)現(xiàn)了常用的實(shí)時(shí)數(shù)倉場(chǎng)景，深入優(yōu)化并用內(nèi)部 Benchmark 驗(yàn)證了性能和穩(wěn)定性超過包括專用實(shí)時(shí)數(shù)倉的其它競(jìng)品之后，再擴(kuò)展到其它復(fù)雜查詢的支持。擴(kuò)展的過程中，在不可避免地系統(tǒng)變得越來越復(fù)雜的同時(shí)，也用 Benchmark 幫助保持簡單實(shí)時(shí)查詢的性能沒有回退。如果在已有的查詢引擎上做改進(jìn)，因?yàn)楹芏嗉軜?gòu)和設(shè)計(jì)上的選擇已經(jīng)定型，牽一發(fā)而動(dòng)全身，就很難達(dá)到這樣的效果。

Hologres執(zhí)行引擎從開發(fā)到落地實(shí)踐面臨了非常多的挑戰(zhàn)，但也給我們提供了機(jī)會(huì)把這個(gè)領(lǐng)域的各種新進(jìn)展都結(jié)合利用起來，并超越已有系統(tǒng)做到對(duì)各種查詢類型的高性能處理，其背后主要是基于以下特點(diǎn)：

分布式執(zhí)行模型：一個(gè)和存儲(chǔ)計(jì)算分離架構(gòu)配合的分布式執(zhí)行模型。執(zhí)行計(jì)劃由異步算子組成的執(zhí)行圖 DAG（有向無環(huán)圖） 表示，可以表達(dá)各種復(fù)雜查詢，并且完美適配 Hologres 的數(shù)據(jù)存儲(chǔ)模型，方便對(duì)接查詢優(yōu)化器，利用業(yè)界各種查詢優(yōu)化技術(shù)。
全異步執(zhí)行：端到端的全異步處理框架，可以避免高并發(fā)系統(tǒng)的瓶頸，充分利用資源，并且最大可能地避免存儲(chǔ)計(jì)算分離系統(tǒng)帶來的讀數(shù)據(jù)延遲的影響。
向量化和列處理：算子內(nèi)部處理數(shù)據(jù)時(shí)最大可能地使用向量化執(zhí)行，和存儲(chǔ)引擎的深度集成，通過靈活的執(zhí)行模型，充分利用各種索引，并且最大化地延遲向量物化和延遲計(jì)算，避免不必要的讀數(shù)據(jù)和計(jì)算。
自適應(yīng)增量處理：對(duì)常見實(shí)時(shí)數(shù)據(jù)應(yīng)用查詢模式的自適應(yīng)增量處理。
特定查詢深度優(yōu)化：對(duì)一些查詢模式的獨(dú)特優(yōu)化
下面將會(huì)對(duì)各個(gè)模塊一一介紹。

分布式執(zhí)行模型
Hologres 是能夠彈性無限水平擴(kuò)展數(shù)據(jù)量和計(jì)算能力的系統(tǒng)，需要能夠支持高效的分布式查詢。

Hologres 查詢引擎執(zhí)行的是由優(yōu)化器生成的分布式執(zhí)行計(jì)劃。執(zhí)行計(jì)劃由算子組成。因?yàn)?Hologres 的一個(gè)表的數(shù)據(jù)會(huì)根據(jù) Distribution Key 分布在多個(gè) Shard 上，每個(gè) Shard 內(nèi)又可以包含很多 Segment，執(zhí)行計(jì)劃也會(huì)反映這樣的結(jié)構(gòu)，并分布到數(shù)據(jù)所在的節(jié)點(diǎn)去執(zhí)行。每個(gè)Table Shard 會(huì)被加載到一個(gè)計(jì)算節(jié)點(diǎn)，數(shù)據(jù)會(huì)被緩存到這個(gè)節(jié)點(diǎn)的內(nèi)存和本地存儲(chǔ)。因?yàn)槭谴鎯?chǔ)計(jì)算分離的架構(gòu)，如果一個(gè)節(jié)點(diǎn)出錯(cuò)，其服務(wù)的 Shard 可以被重新加載到任意一個(gè)計(jì)算節(jié)點(diǎn)，只是相當(dāng)于清空了緩存。

例如一個(gè)比較簡單的查詢。

select key, count(value) as total from table1 group by key order by total desc limit 100。 

如果是單機(jī)數(shù)據(jù)庫，可以用這樣的執(zhí)行計(jì)劃。如果數(shù)據(jù)和計(jì)算分布在多個(gè)節(jié)點(diǎn)上，就需要更復(fù)雜的執(zhí)行計(jì)劃。

在分布式表上，為了更高效地執(zhí)行，盡量減少數(shù)據(jù)傳輸，可以把執(zhí)行計(jì)劃分為不同片段（Fragment）分布到相應(yīng)節(jié)點(diǎn)執(zhí)行，并且把一些操作下推來減少 Fragment 輸出的數(shù)據(jù)，可能就變成這樣的執(zhí)行計(jì)劃：

根據(jù)數(shù)據(jù)的特性，優(yōu)化器可能會(huì)生成不同的計(jì)劃。例如在某一個(gè)局部聚合并沒有顯著減少數(shù)據(jù)量的時(shí)候，可以省略這個(gè)算子。又例如在 Key 就是 Distribution key 的時(shí)候，可以優(yōu)化為：

從這些例子可以看出，Hologres 的執(zhí)行計(jì)劃根據(jù)數(shù)據(jù)的特性切分為不同的片段之后分布式并發(fā)執(zhí)行。片段之間通過 Exchange 算子進(jìn)行數(shù)據(jù)交換。更復(fù)雜的比如多表關(guān)聯(lián)（Join）查詢，會(huì)有更多的片段和更復(fù)雜的數(shù)據(jù)交換模式。

比如以下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中可以是這樣的執(zhí)行計(jì)劃

如果 Join key 和 Distribution Key 一致，可以優(yōu)化為如下執(zhí)行計(jì)劃，減少遠(yuǎn)程數(shù)據(jù)傳輸。根據(jù)需要的查詢合理地設(shè)置 Distribution Key，可能顯著提高查詢性能。

根據(jù)過濾條件和統(tǒng)計(jì)信息等等，優(yōu)化器還可能生成不同的優(yōu)化執(zhí)行計(jì)劃，比如包含動(dòng)態(tài)過濾，局部聚合等等。

這樣的分布式執(zhí)行計(jì)劃足夠通用，可以表達(dá)所有的 SQL 查詢和一些其它查詢。執(zhí)行計(jì)劃和大部分 Massively Parallel Processing (MPP) 系統(tǒng)也比較類似，方便借鑒和集成業(yè)界的一些適用的優(yōu)化。稍微獨(dú)特一些的地方是很多查詢計(jì)劃片段的實(shí)例是和 Hologres 的存儲(chǔ)結(jié)構(gòu)對(duì)齊的，能夠進(jìn)行高效的分區(qū)裁剪和文件裁剪。

同時(shí)，Hologres 實(shí)現(xiàn)了 PostgreSQL 的 Explain 和 Explain Analyze 系列語句，可以展示文本格式的執(zhí)行計(jì)劃和相應(yīng)的執(zhí)行信息，方便用戶自助了解執(zhí)行計(jì)劃，并針對(duì)性做出SQL優(yōu)化調(diào)整。

全異步執(zhí)行

高并發(fā)系統(tǒng)，特別是有大量 I/O 的系統(tǒng)，頻繁地等待或者任務(wù)切換是常見的系統(tǒng)瓶頸。異步處理是一種已經(jīng)被證明行之有效的避免這些瓶頸，并把高并發(fā)系統(tǒng)性能推到極致的方法。

Hologres 的整個(gè)后端，包括執(zhí)行引擎、存儲(chǔ)引擎和其它組件，統(tǒng)一使用 HOS（Hologres Operation System） 組件提供的異步無鎖編程框架，能夠最大化異步執(zhí)行的效果。每個(gè) Fragment 的實(shí)例使用 HOS 的一個(gè) EC （邏輯調(diào)度單位），使得一個(gè) Fragment 里的所有算子和存儲(chǔ)引擎可以異步執(zhí)行并且無鎖安全訪問絕大多數(shù)資源。

算子和 Fragment 都是類似這樣的接口：

future<> Open(const SeekParameters& parameters, ...)future<RecordBatchPtr, bool> GetNext(...)future<> Close(...) 

除了一般異步處理的好處外，異步算子接口較好地規(guī)避了存儲(chǔ)計(jì)算分離架構(gòu)下相對(duì)較高的讀數(shù)據(jù)延遲對(duì)查詢性能的影響，并且對(duì)分布式查詢的執(zhí)行模型本身也有獨(dú)特的好處。

DAG 執(zhí)行引擎一般可以分為拉數(shù)據(jù)的模性（比如火山模型）和推的模型（比如很多大數(shù)據(jù)的分階段執(zhí)行模型），各有其優(yōu)缺點(diǎn)。而 Hologres采用的異步的拉模型能夠取得兩種模型的好處并且避免其缺點(diǎn)（已經(jīng)申請(qǐng)了專利）。舉一個(gè)常見的 Hash Join 來說明：

火山模型可以簡單做到先拉完 b 的數(shù)據(jù)構(gòu)建 hash table，然后流式處理 a 的數(shù)據(jù)不用全放在內(nèi)存里。但是當(dāng) a 或者 b 需要讀數(shù)據(jù)的時(shí)候，簡單的實(shí)現(xiàn)需要等待不能把 CPU 打滿，需要通過提高 Fragment 的并發(fā)數(shù)或者引入復(fù)雜的 pre-fetch 機(jī)制來充分利用資源，而這些又會(huì)引入別的性能問題。

推數(shù)據(jù)的模型，比較容易做到并發(fā)讀數(shù)據(jù)請(qǐng)求并在完成的時(shí)候觸發(fā)下游處理，但是上述 Join算子的實(shí)現(xiàn)會(huì)比較復(fù)雜。比如 a 處理完一批數(shù)據(jù)推到 Join 算子而 b 的 hash table 還沒有構(gòu)建完成，這批數(shù)據(jù)就需要暫存到內(nèi)存里或者盤上，或者引入反壓機(jī)制。在 Fragment 的邊界也會(huì)有類似問題，造成一些在拉數(shù)據(jù)模型下不需要的數(shù)據(jù)緩存。

Hologres 的算子和 Fragment 的異步拉數(shù)據(jù)模型，可以像火山模型一樣簡單做到按需從上游獲取數(shù)據(jù)，而同時(shí)又可以像推數(shù)據(jù)模型一樣簡單做到讀數(shù)據(jù)并發(fā)，只要向上游發(fā)出多個(gè)異步 GetNext，上游處理完成時(shí)會(huì)自然觸發(fā)后續(xù)處理。異步 GetNext 的數(shù)目和時(shí)機(jī)，可以看做是天然的流控機(jī)制，可以有效做到提高 CPU 利用率并且避免不必要的數(shù)據(jù)暫存。

Hologres 已經(jīng)用這個(gè)異步模型實(shí)現(xiàn)了一個(gè)完整的查詢引擎，可以支持所有 PostgreSQL 的查詢。

列處理和向量化

按列處理和向量化執(zhí)行都是分析查詢引擎常用的優(yōu)化機(jī)制，可以大幅度提高數(shù)據(jù)處理的效率。Hologres 也不例外，在能使用向量處理的時(shí)候盡量使用。

Hologres 在內(nèi)存里也采用列式存儲(chǔ)。在內(nèi)存里按列存儲(chǔ)數(shù)據(jù)能夠使用更多的向量處理。列式組織數(shù)據(jù)還有一個(gè)好處，就是對(duì)延遲計(jì)算比較友好。比如 select … where a = 1 and b = 2 …，對(duì)一批數(shù)據(jù)（一般對(duì)應(yīng)存儲(chǔ)的一個(gè) row group），Hologres的 scan 算子輸出的 a 和 b 可以是延遲讀取的 a 和 b 的信息，在處理 a = 1 的時(shí)候會(huì)讀取這一批的 a。如果 a=1 對(duì)這一批的所有行都不滿足，這一批的 b 這一列就根本不會(huì)被讀取。

但是對(duì)某些按行處理的算子，比如 Join，按列存儲(chǔ)的數(shù)據(jù)可能會(huì)造成更多的 CPU cache miss ，帶來較大的性能問題。很多查詢引擎會(huì)在不同的點(diǎn)引入按列存儲(chǔ)和按行存儲(chǔ)的轉(zhuǎn)換，但是頻繁的轉(zhuǎn)換本身會(huì)帶來不小的開銷，而且列轉(zhuǎn)行會(huì)造成上述延遲讀取列被不必要地讀取，還有一些其它的性能問題。

自適應(yīng)增量處理

很多實(shí)時(shí)數(shù)據(jù)應(yīng)用經(jīng)常會(huì)對(duì)一個(gè)查詢用不同的時(shí)間段反復(fù)執(zhí)行。比如一個(gè)監(jiān)控指標(biāo)頁面打開后，會(huì)定期執(zhí)行 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 … 這樣的查詢，下一次會(huì)改成 ts < '2020-11-11 00:03:10'，再下一次 ts < '2020-11-11 00:03:15'。

流計(jì)算或者增量計(jì)算可以對(duì)這種查詢進(jìn)行非常高效的處理。但是對(duì)這種用戶可以隨意生成的交互式查詢，通常不可能對(duì)所有組合都配置流計(jì)算或者增量計(jì)算任務(wù)。如果每次都簡單執(zhí)行查詢，又可能有大量的重復(fù)計(jì)算造成資源浪費(fèi)和性能不理想。

Hologres充分利用存儲(chǔ)引擎和計(jì)算引擎的深度集成和列式存儲(chǔ)大部分?jǐn)?shù)據(jù)在只讀文件中的特性，在能提供包含最新寫入數(shù)據(jù)的查詢結(jié)果的同時(shí)盡量避免重復(fù)計(jì)算，對(duì)這種類型的查詢能夠顯著提升性能和減少資源使用。

針對(duì)特定查詢模式的深度優(yōu)化

Hologres 對(duì)一些特定查詢模式有獨(dú)特的優(yōu)化。這里以Filter Aggregate 優(yōu)化為例子。

很多數(shù)據(jù)應(yīng)用都有開放列的需求，相當(dāng)于可以動(dòng)態(tài)添加邏輯列而不用改 Table Schema。比如有一列是多值列 tags（Postgres 可以用 Array 類型）里面存了'{c1:v1, c2:u1}' 這樣的多個(gè)邏輯列的值。查詢的時(shí)候，如果使用普通列，一類常見的查詢是

-- Q1: select c1, sum(x) from t1 where c1 in (v1, v2, v3) and name = 'abc' group by c1 

使用開放列后，這樣的查詢會(huì)轉(zhuǎn)變?yōu)?/p>

-- 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 可以利用位圖索引快速計(jì)算過濾條件得到相關(guān)的行，但是之后從多值列里面取出相關(guān)數(shù)據(jù)操作不能使用向量處理，性能不能達(dá)到最優(yōu)。經(jīng)過調(diào)研，可以把查詢的執(zhí)行轉(zhuǎn)換為

Q3: select 'c1:v1', sum(x) from t1 where tags && ARRAY['c1:v1'] UNION ALL select 'c1:v2', sum(x) from t1 where tags && ARRAY['c1:v2'] UNION ALL… 

這樣每個(gè) UNION ALL 分支可以只讀取 name 和 tags 的位圖索引計(jì)算過濾條件，然后用 x 列的數(shù)據(jù)和過濾條件進(jìn)行向量計(jì)算 SUM_IF 即可得出想要的結(jié)果。這樣的問題是，每個(gè)分支都要過一遍 t1，讀取 x 列以及 name 列的位圖索引，帶來重復(fù)計(jì)算。最后引入了一個(gè) filter aggregate 的特殊算子來把這類常用查詢優(yōu)化到極致性能，可以只過一遍 t1 并且去掉重復(fù)操作，只用向量計(jì)算即可得到結(jié)果，不需要讀取 tags 列的數(shù)據(jù)。在一個(gè)幾十 TB的表上實(shí)測(cè)性能提升 3 倍以上。

類似的優(yōu)化，Hologres 的執(zhí)行引擎都會(huì)盡量抽象為比較通用的算子，可以適用于更多場(chǎng)景。Filter Aggregate 算子也是 Hologres 申請(qǐng)的專利之一。

總結(jié)

Hologres 執(zhí)行引擎在一個(gè)架構(gòu)里集中了相關(guān)分布式查詢系統(tǒng)的幾乎所有最高效的優(yōu)化方式（包括各種類型的索引）并作出了特有的改進(jìn)。通過和存儲(chǔ)引擎深度整合，能充分發(fā)揮異步模型的優(yōu)勢(shì)，并高效利用各種類型的索引來加速查詢。所有這些加起來，帶來了超越已有系統(tǒng)的性能，并在阿里巴巴雙 11 的數(shù)據(jù)規(guī)模下通過了實(shí)戰(zhàn)的考驗(yàn)，（2020年雙11頂住了5.96億/秒的實(shí)時(shí)數(shù)據(jù)洪峰，基于萬億級(jí)數(shù)據(jù)對(duì)外提供多維分析和服務(wù)，99.99%的查詢可以在80ms以內(nèi)返回結(jié)果），對(duì)外高并發(fā)高性能地提供分布式 HSAP 查詢服務(wù)。