ClickHouse已經(jīng)成為行業(yè)主流且熱門(mén)的開(kāi)源引擎。隨著業(yè)務(wù)數(shù)據(jù)量擴(kuò)大，場(chǎng)景覆蓋變廣泛，在復(fù)雜query場(chǎng)景下，ClickHouse容易存在查詢(xún)異常問(wèn)題，影響業(yè)務(wù)正常推進(jìn)。

?全文將圍繞以下幾方面展開(kāi)：

• ?項(xiàng)目背景
• 技術(shù)方案
• 優(yōu)化與診斷
• 效果及展望

01 項(xiàng)目背景

1. ClickHouse執(zhí)行模式

ClickHouse 的執(zhí)行模式相對(duì)比較簡(jiǎn)單，和Druid、ES 類(lèi)似，其基本查詢(xún)模式分為兩個(gè)階段：

第一階段，Coordinator 收到查詢(xún)后將請(qǐng)求發(fā)送給對(duì)應(yīng)的 worker 節(jié)點(diǎn)；

第二階段，Coordinator 收到各個(gè) worker 節(jié)點(diǎn)的結(jié)果后匯聚起來(lái)處理后返回。

以下面的SQL為例：?

Select name from student_distribute where id = 5?

①當(dāng) Coordinator 收到請(qǐng)求后，由于student_distribute是一個(gè)分布式表，因此需要將SQL 改寫(xiě)為對(duì)local表查詢(xún)，并轉(zhuǎn)發(fā)請(qǐng)求給每一個(gè)shard的worker；

②Worker收到請(qǐng)求后查詢(xún)本地的local表數(shù)據(jù)，返回結(jié)果給coordinator；

③Coordinator匯總每一個(gè)shard的數(shù)據(jù)并把結(jié)果返回給client。

Select name from student_local where id = 5

第二階段執(zhí)行的模式能夠高效地支持很多常見(jiàn)場(chǎng)景，比如常見(jiàn)的針對(duì)大寬表的各類(lèi)查詢(xún)，但是隨著業(yè)務(wù)場(chǎng)景的復(fù)雜化，也存在以下三點(diǎn)問(wèn)題：

其一，第一階段返回的數(shù)據(jù)比較多且第二階段的計(jì)算比較復(fù)雜時(shí)，對(duì)于Coordinator的壓力會(huì)比較大，容易成為query的瓶頸，且shard越多可能計(jì)算越慢，瓶頸越大。例如一些重計(jì)算的agg算子count distinct。如果我們使用hash表去重時(shí)，第二階段需要在coordinator單機(jī)上merge各個(gè)worker的hash表，計(jì)算量很重且不能并行；又比如說(shuō)group by基數(shù)比較大或者window計(jì)算。

其二，join是SQL的重要場(chǎng)景。由于不支持Shuffle操作，對(duì)于Join來(lái)說(shuō)右表必須是全量數(shù)據(jù)。無(wú)論是普通Join還是Global Join，當(dāng)Join的右表比較大時(shí)都放到內(nèi)存里容易OOM，而Spill到磁盤(pán)雖然解決內(nèi)存問(wèn)題，可能會(huì)因?yàn)橛写疟P(pán) io和序列化計(jì)算的開(kāi)銷(xiāo)影響性能。特別是當(dāng)Join為最常見(jiàn)的Hash Join 時(shí)，右表如果是大表構(gòu)建也比較慢。雖然社區(qū)最近也做了一些右表構(gòu)建的優(yōu)化，通過(guò)單機(jī)按照 join key split 來(lái)達(dá)到并行構(gòu)建hash table。但是額外的代價(jià)是左右表都增加了一次 split 操作。

其三，對(duì)于復(fù)雜查詢(xún)（如多表 Join、嵌套多個(gè)子查詢(xún)、window function等）的支持并不友好，由于不能通過(guò)shuffle來(lái)分散數(shù)據(jù)，生成的pipeline在一些case下不能充分并行，難以充分發(fā)揮集群的全部資源。

2. 其他MMP數(shù)據(jù)庫(kù)

目前主流的MPP數(shù)據(jù)庫(kù)基本都支持Stage執(zhí)行的方式。以Presto為例，如下圖所示，一個(gè)兩表join的agg sql可拆分為5個(gè) Stage。

其中 Stage3、Stage4分別對(duì)應(yīng)左右表數(shù)據(jù)讀取，Stage2完成兩表Join和partial agg 計(jì)算，Stage1完成final agg計(jì)算，Stage0收集Stage1的數(shù)據(jù)后匯總和輸出。在這個(gè)過(guò)程中，Stage 3、4、2、1可以在多個(gè)節(jié)點(diǎn)上并行執(zhí)行，單個(gè)復(fù)雜的query被拆分成若干Stage，從而實(shí)現(xiàn)了Stage之間，不同worker的數(shù)據(jù)傳輸。

3. 業(yè)務(wù)背景和目標(biāo)

隨著業(yè)務(wù)復(fù)雜程度提高，業(yè)務(wù)并不希望所有的數(shù)據(jù)都通過(guò)etl 產(chǎn)生大寬表；復(fù)雜查詢(xún)（特別是多輪分布式 Join和比較多的agg）的需求越來(lái)越強(qiáng)烈，而整體的數(shù)據(jù)量又在不斷增長(zhǎng)。在集群資源有限的情況下，我們希望能夠充分利用機(jī)器資源，基于ClickHouse 高效地支持復(fù)雜查詢(xún)。

ByteHouse是字節(jié)跳動(dòng)研發(fā)同學(xué)基于開(kāi)源ClickHouse 進(jìn)行了深度優(yōu)化和改造的版本，提供海量數(shù)據(jù)上更強(qiáng)的查詢(xún)服務(wù)和數(shù)據(jù)寫(xiě)入性能，支持多種應(yīng)用場(chǎng)景。如圖所示，ByteHouse在內(nèi)部多個(gè)場(chǎng)景如行為分析、畫(huà)像分析、智能營(yíng)銷(xiāo)分析、APP 日志分析上得到充分的驗(yàn)證和使用，并在多個(gè)方面進(jìn)行了增強(qiáng)，具備特有的能力。

02 技術(shù)方案

1. 設(shè)計(jì)思想

基于 ClickHouse 的復(fù)雜查詢(xún)的實(shí)現(xiàn)采用分Stage的方式，替換目前 ClickHouse的兩階段執(zhí)行方式。類(lèi)似其他分布式數(shù)據(jù)庫(kù)引擎（如 Presto、Impala 等），將一個(gè)復(fù)雜的Query按照數(shù)據(jù)交換情況切分成多個(gè)Stage，Stage和Stage之間通過(guò) exchange完成數(shù)據(jù)的交換，單個(gè)Stage內(nèi)不存在數(shù)據(jù)交換。Stage間的數(shù)據(jù)交換主要有以下三種形式：

①按照單（多）個(gè) key 進(jìn)行 Shuffle（shuffle）

②由1個(gè)或者多個(gè)節(jié)點(diǎn)匯聚到一個(gè)節(jié)點(diǎn)（我們稱(chēng)為 gather）

③同一份數(shù)據(jù)復(fù)制到多個(gè)節(jié)點(diǎn)（也稱(chēng)為 broadcast 或者說(shuō)廣播）?

按照不同的功能切分不同的模塊，設(shè)計(jì)目標(biāo)如下：

①各個(gè)模塊約定好接口，盡量減少彼此的依賴(lài)和耦合。一旦某個(gè)模塊有變動(dòng)不會(huì)影響別的模塊，例如Stage生成邏輯的調(diào)整不影響調(diào)度的邏輯。

②模塊采用插件的架構(gòu)，允許模塊根據(jù)配置靈活支持不同的策略。

2. 相關(guān)術(shù)語(yǔ)

• ExchangeNode 在語(yǔ)法樹(shù)中表示數(shù)據(jù)交換的節(jié)點(diǎn)
• PlanSegment 單個(gè) Stage 對(duì)應(yīng)的執(zhí)行的計(jì)劃片段
• ExchangeManager 管理數(shù)據(jù)的 exchange，負(fù)責(zé)不同 Stage 節(jié)點(diǎn)之間的數(shù)據(jù)交換
• SegmentScheduler 計(jì)劃片段調(diào)度器，負(fù)責(zé)下發(fā)計(jì)劃片段給 worker，由 Coordinator 節(jié)點(diǎn)調(diào)用
• InterpreterPlanSegment 計(jì)劃片段執(zhí)行器，執(zhí)行一個(gè)具體的計(jì)劃片段

3. 執(zhí)行流程

①Coordinator 接受復(fù)雜查詢(xún)后，在目前 ClickHouse 語(yǔ)法樹(shù)的基礎(chǔ)上，根據(jù)節(jié)點(diǎn)類(lèi)型和數(shù)據(jù)分布情況插入 Exchange 節(jié)點(diǎn)并生成分布式 Plan。

②Coordinator 根據(jù) Exchange Node 類(lèi)型，切分分布式 Plan 生成每個(gè) Stage 的執(zhí)行片段 PlanSegment。

③Coordinator 調(diào)用 SegmentScheduler 將各階段的 PlanSegment 發(fā)送到 Worker 節(jié)點(diǎn)。

④Worker 節(jié)點(diǎn)接受 PlanSegment 通過(guò) InterpreterPlanSegment 完成數(shù)據(jù)的讀取和執(zhí)行，通過(guò) ExchangeManager 完成數(shù)據(jù)的交互。

⑤Coordinator 從最后一輪 Stage 對(duì)應(yīng)節(jié)點(diǎn)的 ExchangeManager 讀取數(shù)據(jù)后處理后返回給 client。

4. Plan切分

下面是一個(gè)Plan切分的例子，這是1個(gè)2表Join的查詢(xún)場(chǎng)景，根據(jù)Exchange信息，將整個(gè)分布式 Plan切分成4個(gè)Stage。

5. 查詢(xún)片段調(diào)度器（SegmentScheduler）

查詢(xún)片段調(diào)度器SegmentScheduler 根據(jù)上下游依賴(lài)關(guān)系和數(shù)據(jù)分布，以及 Stage 并行度和worker 分布和狀態(tài)信息，按照一定的調(diào)度策略，將 PlanSemgent 發(fā)給不同的 Worker 節(jié)點(diǎn)。

目前支持的2種策略是：?

①依賴(lài)調(diào)度：根據(jù) Stage 依賴(lài)關(guān)系定義拓?fù)浣Y(jié)構(gòu)，產(chǎn)生 DAG 圖，根據(jù) DAG 圖調(diào)度 stage，類(lèi)似于拓?fù)渑判颍鹊揭蕾?lài)的 Stage 啟動(dòng)后再啟動(dòng)新的 Stage。例如剛才的兩表 join，會(huì)先調(diào)度左右表讀取 stage，再調(diào)度 join stage。

②AllAtOnce：類(lèi)似于Presto的AllAtOnce策略，會(huì)先計(jì)算每一個(gè) Stage 的相關(guān)信息，一次性調(diào)度所有的Stage。

相比而言，這兩種策略是在容錯(cuò)、資源使用和延時(shí)上做取舍。

第一種調(diào)度策略可以實(shí)現(xiàn)更好的容錯(cuò)，由于 ClickHouse 可以有多個(gè)副本，當(dāng)前一個(gè) Stage 部分節(jié)點(diǎn)連接失敗時(shí)可以嘗試切換到副本節(jié)點(diǎn)，對(duì)后續(xù)依賴(lài) stage 無(wú)感知。這里指的是讀數(shù)據(jù)的 Stage，我們稱(chēng)為 Source Stage，非 Source Stage 因?yàn)闆](méi)有數(shù)據(jù)依賴(lài)，容錯(cuò)能力會(huì)更強(qiáng)，只要保證并行度的節(jié)點(diǎn)數(shù)即可，甚至極端情況下可以降低 stage 并行度來(lái)支持更好的容錯(cuò)。缺點(diǎn)是調(diào)度有依賴(lài)，不能完全并行，會(huì)增加調(diào)度時(shí)長(zhǎng)，對(duì)于一些數(shù)據(jù)量和計(jì)算量小，但是 stage 多的節(jié)點(diǎn)調(diào)度延時(shí)可能會(huì)占 SQL 整體時(shí)間不小的比例。我們也做了一些針對(duì)性的優(yōu)化，對(duì)于無(wú)依賴(lài)關(guān)系的盡可能支持并行。

第二種調(diào)度策略通過(guò)并行可以極大降低調(diào)度延時(shí)，為防止大量網(wǎng)絡(luò) io 線(xiàn)程，我們通過(guò)異步化并且控制線(xiàn)程數(shù)目；這種策略的缺點(diǎn)是容錯(cuò)性沒(méi)有依賴(lài)調(diào)度好，因?yàn)槊恳粋€(gè) stage 的 worker 在調(diào)度前就已經(jīng)確定，如果有一個(gè) worker 出現(xiàn)連接異常則整個(gè)查詢(xún)會(huì)直接失敗。并且可能有一些 Stage 上游數(shù)據(jù)還沒(méi)有 Ready 就被調(diào)度執(zhí)行了，需要長(zhǎng)時(shí)間等數(shù)據(jù)。例如 final agg stage，需要等 partial agg 完成后才能拿到數(shù)據(jù)。雖然我們做了一些優(yōu)化，并不會(huì)長(zhǎng)時(shí)間空跑浪費(fèi) cpu 資源，但是畢竟也消耗了一部分資源，比如創(chuàng)建了執(zhí)行的線(xiàn)程。

6. 查詢(xún)片段執(zhí)行器（InterpreterPlanSegment）

下面介紹下計(jì)劃片段是如何執(zhí)行的，原本 ClickHouse的查詢(xún)和節(jié)點(diǎn)執(zhí)行主要是 SQL 形式，切分Stag后需要支持執(zhí)行一個(gè)單獨(dú)的PlanSemgent。因此 InterpreterPlanSegment 的主要功能就是接受一個(gè)序列化后的 PlanSemgent，能夠在 Worker 節(jié)點(diǎn)上運(yùn)行整個(gè) PlanSemgent 的邏輯。主要的步驟為：

①根據(jù) input 信息讀取數(shù)據(jù)，如果 input 是具體的 table，則從本地讀取數(shù)據(jù)；如果 input 是一個(gè) exchange input，則從對(duì)應(yīng)的 ExchangeManager 讀取數(shù)據(jù)；

②執(zhí)行 PlanSemgent 的邏輯；

③輸出處理后的結(jié)果數(shù)據(jù)，如果是 Coordinator 節(jié)點(diǎn)，就將數(shù)據(jù)發(fā)給 Client；如果是非Coordinator 節(jié)點(diǎn)，就按照數(shù)據(jù)的exchange方式寫(xiě)給本實(shí)例對(duì)應(yīng)的 ExchangeManager。

Interpreter部分我們盡量復(fù)用當(dāng)前ClickHouse的執(zhí)行邏輯，例如processor 執(zhí)行方式，process list管理等等。相比于InterpreterSelect邏輯要更簡(jiǎn)單一些，可以認(rèn)為1 個(gè)Stage只有1個(gè)階段。當(dāng)然我們也做了很多功能和性能的增強(qiáng)，例如我們支持1個(gè) stage處理多個(gè)join等，這樣可以減少stage數(shù)目和不必要的數(shù)據(jù)傳輸，在一張大表（通常情況下是事實(shí)表） join 多個(gè)維度表的場(chǎng)景有比較好的幫助。

InterpreterPlan Segment執(zhí)行完會(huì)向coordinator上報(bào)對(duì)應(yīng)的狀態(tài)信息。執(zhí)行異常的時(shí)候會(huì)將異常信息報(bào)告給查詢(xún)片段調(diào)度器，取消Query其他worker的執(zhí)行。

7. 數(shù)據(jù)交換（ExchangeManager）

ExchangeManager是PlanSegment數(shù)據(jù)交換的媒介，更是平衡數(shù)據(jù)上下游處理能力的重要組件。整體上采用 push 的方式，當(dāng)上游數(shù)據(jù) ready 時(shí)主動(dòng)推送給下游，并支持反壓。其架構(gòu)如下圖所示：

具體的流程如下：?

①下游PlanSegment執(zhí)行時(shí)，當(dāng)input為exchange input時(shí)，根據(jù)一定的 token 規(guī)則 （通常由 query_id+segment_id+index_id 等組成）和數(shù)據(jù) source 信息，向上游 ExchangeManager 注冊(cè)對(duì)應(yīng)的數(shù)據(jù)請(qǐng)求；

②上游ExchangeManager收到請(qǐng)求后，建立上下游數(shù)據(jù)通道，并將上游的數(shù)據(jù)推送到下游，如果通道一直建立不了會(huì) block 上游的執(zhí)行。

在這個(gè)過(guò)程中，上下游都會(huì)通過(guò)隊(duì)列來(lái)優(yōu)化發(fā)送和讀取，當(dāng)隊(duì)列飽和的時(shí)候通過(guò)反壓的機(jī)制控制上游的執(zhí)行速度。由于采用了 push 和隊(duì)列，這里我們要考慮一個(gè)特殊的場(chǎng)景，在某些 case 下下游的 Stage 并不需要讀取全部的上游數(shù)據(jù)，一個(gè)典型的場(chǎng)景是 limit。例如 limit 100，下游 stage 是需要讀取 100 條數(shù)據(jù)即可，而上游可能會(huì)輸出更大規(guī)模的數(shù)據(jù)，因此在這種情況下，當(dāng)下游 stage 讀到足夠的數(shù)據(jù)后，需要能主動(dòng)取消上游數(shù)據(jù)的執(zhí)行并清空隊(duì)列。這是一個(gè)特定場(chǎng)景的優(yōu)化，能夠大大加速查詢(xún)時(shí)間。

ExchangeManager 需要考慮和優(yōu)化的點(diǎn)還有：?

①細(xì)粒度的內(nèi)存控制，能夠按照實(shí)例、query、segment 多層次進(jìn)行內(nèi)存控制，避免 OOM，更長(zhǎng)期的考慮是支持 spill 到磁盤(pán)上，降低對(duì)內(nèi)存的使用。為了提升傳輸效率，小數(shù)據(jù)需要進(jìn)行 merge，大數(shù)據(jù)要 split。同時(shí)，網(wǎng)絡(luò)處理在某些場(chǎng)景要保證有序性，比如 sort 時(shí)，partial sort 和 merge sort 的網(wǎng)絡(luò)傳輸必須有序，否則數(shù)據(jù)可能是有問(wèn)題的。

②連接復(fù)用和網(wǎng)絡(luò)優(yōu)化，包括針對(duì)上下游在同一個(gè)節(jié)的場(chǎng)景下選擇走內(nèi)存的交換不走網(wǎng)絡(luò)，可以減少網(wǎng)絡(luò)的開(kāi)銷(xiāo)和減少數(shù)據(jù)序列化、反序列化的代價(jià)。另外，由于 ClickHouse 在計(jì)算方面做了非常充足的優(yōu)化，有些場(chǎng)景下甚至內(nèi)存帶寬成為瓶頸，我們?cè)贓xchangeManager的一些場(chǎng)景上也應(yīng)用zero copy等技術(shù)來(lái)減少內(nèi)存的拷貝。

③異常處理和監(jiān)控，相比于單機(jī)執(zhí)行，分布式情況下異常情況更復(fù)雜且不好感知。通過(guò)重試能避免一些節(jié)點(diǎn)的暫時(shí)高負(fù)載或者異常，以及出問(wèn)題時(shí)能夠快速感知、排查和做針對(duì)性解決和優(yōu)化。這里的工程實(shí)踐更多一些。

03 優(yōu)化與診斷

1. Join 多種實(shí)現(xiàn)

根據(jù)數(shù)據(jù)的規(guī)模和分布，我們支持了多種Join實(shí)現(xiàn)，目前已經(jīng)支持的有：

①Shuffle Join，最通用的 Join；

②Broadcast Join，針對(duì)大表Join小表的場(chǎng)景，通過(guò)把右表廣播到左表的所有 worker 節(jié)點(diǎn)來(lái)減少左表的傳輸；

③Colocate Join，針對(duì)左右表根據(jù)Join key保持相通分布的場(chǎng)景，減少左右表數(shù)據(jù)傳輸。

2. 網(wǎng)絡(luò)連接優(yōu)化

網(wǎng)絡(luò)連接的優(yōu)化的核心本質(zhì)就是減少連接的使用。特別是數(shù)據(jù)需要Shuffle 的時(shí)候，下一輪 Stage的每一個(gè)節(jié)點(diǎn)需要從上一輪Stage的每一個(gè)節(jié)點(diǎn)拉取數(shù)據(jù)。當(dāng)一個(gè)集群的節(jié)點(diǎn)比較多的時(shí)候，如果存在比較多的復(fù)雜 Query(Stage多，并行度（節(jié)點(diǎn)數(shù)）比較大)，集群的Worker節(jié)點(diǎn)會(huì)建立非常多的連接，如下圖所示，單節(jié)點(diǎn)建立的連接數(shù)與集群節(jié)點(diǎn)數(shù)、并發(fā)stage數(shù)成正比。

字節(jié)內(nèi)部的clickhouse集群規(guī)模非常大，最大的集群（單集群幾千臺(tái)規(guī)模）在目前 ClickHouse 的執(zhí)行模式下單機(jī)最大可能會(huì)建立上幾萬(wàn)個(gè)網(wǎng)絡(luò)連接。因此如果支持復(fù)雜 Query 執(zhí)行，由于stage變多了，需要優(yōu)化網(wǎng)絡(luò)連接，特別是支持連接復(fù)用。我們通過(guò)盡可能復(fù)用連接，在不同節(jié)點(diǎn)之間只會(huì)建立固定數(shù)目的連接，不同的查詢(xún)會(huì)復(fù)用這些連接，不隨 query 和 stage 的規(guī)模而增長(zhǎng)。

3. 網(wǎng)絡(luò)傳輸優(yōu)化

在數(shù)據(jù)中心領(lǐng)域，遠(yuǎn)程直接內(nèi)存訪(fǎng)問(wèn)（RDMA）是一種繞過(guò)遠(yuǎn)程主機(jī)操作系統(tǒng)內(nèi)核訪(fǎng)問(wèn)其內(nèi)存中數(shù)據(jù)的技術(shù)，由于不經(jīng)過(guò)操作系統(tǒng)，不僅節(jié)省了大量CPU資源，同樣也提高了系統(tǒng)吞吐量、降低了系統(tǒng)的網(wǎng)絡(luò)通信延遲，尤其適合在大規(guī)模并行計(jì)算機(jī)集群中有廣泛應(yīng)用。

由于ClickHouse在計(jì)算層面做了很多優(yōu)化，而網(wǎng)絡(luò)帶寬相比于內(nèi)存帶寬要小不少，在一些數(shù)據(jù)量傳輸特別大的場(chǎng)景，網(wǎng)絡(luò)傳輸會(huì)成為一定的瓶頸。為了提升網(wǎng)絡(luò)傳輸?shù)男屎吞嵘龜?shù)據(jù)exchange的吞吐，一方面我們引入壓縮來(lái)降低傳輸數(shù)據(jù)量，另一方面我們引入 RDMA 來(lái)減少一定的開(kāi)銷(xiāo)。經(jīng)過(guò)測(cè)試，在一些數(shù)據(jù)傳輸量大的場(chǎng)景，有不小的收益。

4. Runtime Filter

Join算子通常是OLAP引擎中最耗時(shí)的算子。如果想優(yōu)化 Join 算子，可以有兩種思路，一方面可以提升Join算子的性能，例如更好的Hash Table實(shí)現(xiàn)和Hash算法，以及更好的并行。另一方面可以盡可能減少參與Join計(jì)算的數(shù)據(jù)。

Runtime Filter在一些場(chǎng)景，特別是事實(shí)表join維度表的星型模型場(chǎng)景下會(huì)有比較大的效果。因?yàn)檫@種情況下通常事實(shí)表的規(guī)模比較大，而大部分過(guò)濾條件都在維度表上，事實(shí)表可能要全量join維度表。Runtime Filter的作用是通過(guò)在 Join 的 probe 端（就是左表）提前過(guò)濾掉那些不會(huì)命中Join的輸入數(shù)據(jù)來(lái)大幅減少 Join 中的數(shù)據(jù)傳輸和計(jì)算，從而減少整體的執(zhí)行時(shí)間。以下圖為例：

左表并沒(méi)有直接過(guò)濾條件，右表帶有過(guò)濾條件item.proce > 1000。當(dāng)完成右表查詢(xún)時(shí)，可以確定item.id 的范圍和集合，根據(jù)join類(lèi)型inner join和join條件sales.item_id=item.id可以推斷出sales.item的范圍和集合。我們可以把sales.item 的范圍和集合作為一個(gè)過(guò)濾條件，在join前過(guò)濾sales的數(shù)據(jù)。

我們?cè)趶?fù)雜查詢(xún)上支持了Runtime Filter，目前主要支持minmax和bloomfilter。

總體執(zhí)行流程如下：

①build plan segment worker（right table）會(huì)將生成的單節(jié)點(diǎn) runtime filter 發(fā)送到coordinator節(jié)點(diǎn)；

②coordinator 在等待各個(gè) worker的 runtime filter 都發(fā)送完成之后進(jìn)行一次merge操作，將合并好的 runtime filter 分發(fā)到各個(gè) execute plan segment worker（left table）節(jié)點(diǎn)中去；

③在 runtime filter 構(gòu)造期間，execute plan segment（left table） 需要等待一定的時(shí)間，在超時(shí)之前如果runtime filter已經(jīng)下發(fā)，則通過(guò) runtime filter 執(zhí)行過(guò)濾。

這里需要思考一個(gè)問(wèn)題，Runtime filter column 是否構(gòu)建索引（主鍵、skip index等）和命中prewhere？如果runtime filter的列（join column）構(gòu)建了索引是需要重新生成 pipeline 的。因?yàn)槊兴饕螅赡軙?huì)減少數(shù)據(jù)的讀取，pipeline并行度和對(duì)應(yīng)數(shù)據(jù)的處理range都可能發(fā)生變化。如果runtime filter的列跟索引無(wú)關(guān)，可以在計(jì)劃生成的時(shí)候預(yù)先帶上過(guò)濾條件，只不過(guò)一開(kāi)始作為占位是空的，runtime filter下發(fā)的時(shí)候把占位信息改成真正的過(guò)濾條件即可。這樣即使runtime filter 下發(fā)超時(shí)了，查詢(xún)片段已經(jīng)開(kāi)始執(zhí)行了，只要查詢(xún)片段沒(méi)有執(zhí)行完，之后的數(shù)據(jù)仍然可以進(jìn)行過(guò)濾。

需要注意的是，runtime filter 是一種特殊場(chǎng)景下的優(yōu)化，其針對(duì)的場(chǎng)景是右表數(shù)據(jù)量不大，且構(gòu)建的 runtime filter 對(duì)左表有比較強(qiáng)的過(guò)濾效果。如果右表數(shù)據(jù)量比較大，構(gòu)建runtime filter比較慢，或者對(duì)左表的數(shù)據(jù)過(guò)濾效果很差甚至沒(méi)有，那么 runtime filter 反而會(huì)增加查詢(xún)的耗時(shí)。因此，要根據(jù)數(shù)據(jù)的特征和規(guī)模來(lái)決定是否開(kāi)啟。

5. 診斷和分析

引入復(fù)雜查詢(xún)的多Stage 執(zhí)行模型后，SQL的執(zhí)行模式變得復(fù)雜了。特別是當(dāng)用戶(hù)查詢(xún)一些非常復(fù)雜的查詢(xún)，幾百行的sql生成的stage會(huì)非常多，把stage都看一遍并理解sql的含義要花比較長(zhǎng)的時(shí)間。題外話(huà)：我們很早之前就完整的跑通了所有的tpcds query，這里面就有一些sql可能會(huì)產(chǎn)生幾十個(gè) stage。那么在這種情況下，如何定位 SQL 的瓶頸并加以?xún)?yōu)化是一個(gè)難題。

我們做了如下兩點(diǎn)優(yōu)化：?

首先，最常見(jiàn)的做法是增加各類(lèi)完善的metrics，包括整個(gè)Query的執(zhí)行時(shí)間和不同Stage的執(zhí)行時(shí)間、IO數(shù)據(jù)量、算子處理數(shù)據(jù)和執(zhí)行情況、算子 metrics 和profile event等。

其次，我們記錄了反壓信息和上下游隊(duì)列長(zhǎng)度，以此來(lái)推斷 stage 執(zhí)行情況和瓶頸。

坦率地說(shuō)，SQL 場(chǎng)景包括萬(wàn)象，很多非常復(fù)雜的場(chǎng)景目前還是需要對(duì)引擎比較熟悉的同學(xué)才能診斷和分析SQL才能給出優(yōu)化建議。在不斷積累經(jīng)驗(yàn)的過(guò)程中，我們希望通過(guò)能夠不斷完善 metrics 和分析路徑，不斷減輕oncall的負(fù)擔(dān)，并且在某些場(chǎng)景下可以更智能的給出優(yōu)化提示，這對(duì)于使用同學(xué)來(lái)說(shuō)也是有好處的。

04 效果及展望

1. 復(fù)雜查詢(xún)效果

根據(jù)上面的執(zhí)行模型的三個(gè)缺點(diǎn)，分別測(cè)試如下三個(gè)場(chǎng)景：

①第二階段的計(jì)算比較復(fù)雜

②Hash Join 右表為大表

③多表 Join

以SSB 1T數(shù)據(jù)作為數(shù)據(jù)集，集群包含8個(gè)節(jié)點(diǎn)。

2. 第二階段的計(jì)算比較復(fù)雜

這個(gè)case SQL 如下圖所示

?

uniqExact是count distinct的默認(rèn)算法，采用hash table進(jìn)行數(shù)據(jù)去重。使用復(fù)雜查詢(xún)后，query 執(zhí)行時(shí)間從 8.514s=>2.198s，第二階段 agg uniqExact 算子的合并原本由 coordinator單點(diǎn)合并，現(xiàn)在通過(guò)按照group by key shuffle 后可以由多個(gè)節(jié)點(diǎn)并行完成。因此通過(guò)shuffle減輕了coordinator的merge agg 壓力。

3. Hash Join 右表為大表

這個(gè) case 演示了右表是一個(gè)大表的場(chǎng)景，由于 ClickHouse 對(duì)多表的優(yōu)化做的還不是很到位。這里采用子查詢(xún)來(lái)下推過(guò)濾的條件。

在這個(gè)case中，采用復(fù)雜查詢(xún)模式后，query 執(zhí)行時(shí)間從17.210=>1.749s。lineorder 是一張大表，通過(guò)shuffle可以將大表數(shù)據(jù)按照join key shuffle到每個(gè)worker節(jié)點(diǎn)，減少了右表構(gòu)建的壓力。

4. 多表 Join

這個(gè) case 是一個(gè) 5 表 join 的 case。

?

開(kāi)啟復(fù)雜查詢(xún)模式后，query 執(zhí)行時(shí)間從8.583s=>4.464s，所有的右表可同時(shí)開(kāi)始數(shù)據(jù)讀取和構(gòu)建。為了和現(xiàn)有模式對(duì)比，針對(duì)復(fù)雜查詢(xún)沒(méi)有開(kāi)啟 runtime filter，開(kāi)啟 runtime filter后效果會(huì)更快。

這里還要重點(diǎn)說(shuō)一下，今天的分享主要是從執(zhí)行模式上講解如何支持復(fù)雜查詢(xún)。實(shí)際上，優(yōu)化器對(duì)于復(fù)雜查詢(xún)的性能提升也非常大。通過(guò)一些rbo的規(guī)則，比如常見(jiàn)的謂詞下推、相關(guān)子查詢(xún)處理等。實(shí)際上這里的優(yōu)化規(guī)則非常多，可以極大的提升 SQL 的執(zhí)行效率。上面的 SQL 其實(shí)原本比較簡(jiǎn)單，5 表 join 和一些維表的過(guò)濾條件，這里寫(xiě)成子查詢(xún)是為了在 ClickHouse 現(xiàn)有模式下右表過(guò)濾條件更好下推。其實(shí)對(duì)于我們來(lái)說(shuō)，在復(fù)雜查詢(xún)的模式下，由于有優(yōu)化器的存在，用戶(hù)不用寫(xiě)的這么復(fù)雜，優(yōu)化器會(huì)自動(dòng)完成下推和rbo優(yōu)化。

上面是一些規(guī)則的優(yōu)化，實(shí)際上在復(fù)雜查詢(xún)中， cbo 的優(yōu)化也有很大作用。舉一個(gè)例子，在 ClickHouse 中，相同的兩個(gè)表，大表 join 小表的性能比小表 join 大表要好很多。前一個(gè)效果 2 中如果把表順序調(diào)整一下會(huì)快很多；另外，選用哪一種 join 的實(shí)現(xiàn)對(duì) join 性能影響比較大，如果滿(mǎn)足 join key 分布，colcate join 比 shuffle join 來(lái)說(shuō)完全減少了數(shù)據(jù)的 shuffle。多表 join 中，join 的順序和 join 的實(shí)現(xiàn)方式對(duì)執(zhí)行的時(shí)長(zhǎng)影響會(huì)比 2 表 join 影響更大。借助數(shù)據(jù)的統(tǒng)計(jì)信息，通過(guò)一些 cbo 優(yōu)化，可以得到一個(gè)比較優(yōu)的執(zhí)行模式。

有了優(yōu)化器，業(yè)務(wù)同學(xué)可以按照業(yè)務(wù)邏輯來(lái)寫(xiě)任何的 SQL，引擎自動(dòng)計(jì)算出相對(duì)最優(yōu)的 SQL 計(jì)劃并執(zhí)行，加速查詢(xún)的執(zhí)行。

5. 展望

CLickHouse 目前的模式其實(shí)在很多單表查詢(xún)的場(chǎng)景上表現(xiàn)優(yōu)異。我們主要是針對(duì)復(fù)雜的查詢(xún)場(chǎng)景做優(yōu)化，主要是實(shí)現(xiàn)多stage的執(zhí)行模式，并實(shí)現(xiàn)了stage之間數(shù)據(jù)傳輸。工程實(shí)踐上來(lái)說(shuō)，做了比較多的嘗試和優(yōu)化來(lái)提升執(zhí)行和網(wǎng)絡(luò)傳輸?shù)男阅埽⑶蚁Ｍㄟ^(guò)完善metrics和智能診斷來(lái)降低SQL分析和調(diào)優(yōu)的門(mén)檻，并減少oncall 的壓力。

目前的實(shí)現(xiàn)只是第一步，未來(lái)我們還有很多努力的方向。

首先，肯定是繼續(xù)提升執(zhí)行和 Exchange 的性能。這里不談?wù)撘鎴?zhí)行通用的優(yōu)化，比如更好的索引或者算子的優(yōu)化，主要是跟復(fù)雜查詢(xún)模式有關(guān)。

其次是Metrics 和智能診斷加強(qiáng)，就如同剛才提到的，SQL 的靈活度太高了，對(duì)于一些復(fù)雜的查詢(xún)沒(méi)有 metrics 幾乎難以診斷和調(diào)優(yōu)，這個(gè)我們會(huì)長(zhǎng)期持續(xù)的去做。