引言
ClickHouse核心分析系列文章，本文將為大家深度解讀ClickHouse當前的MPP計算模型、使用者資源隔離、查詢限流機制，在此基礎上為大家介紹阿里巴巴雲資料庫ClickHouse在八月份即將推出的自研彈性資源佇列功能。ClickHouse開源版本當前還沒有資源佇列相關的規劃，自研彈性資源佇列的初衷是更好地解決隔離和資源利用率的問題。下文將從ClickHouse的MPP計算模型、現有的資源隔離方案展開來看ClickHouse當前在資源隔離上的痛點，最後為大家介紹我們的自研彈性資源佇列功能。
MPP計算模型在深入到資源隔離之前，這裡有必要簡單介紹一下ClickHouse社群純自研的MPP計算模型，因為ClickHouse的MPP計算模型和成熟的開源MPP計算引擎（例如：Presto、HAWQ、Impala）存在著較大的差異（que xian），這使得ClickHouse的資源隔離也有一些獨特的要求，同時希望這部分內容能指導使用者更好地對ClickHouse查詢進行調優。
ClickHouse的MPP計算模型最大的特點是：它壓根沒有分散式執行計劃，只能透過遞迴子查詢和廣播表來解決多表關聯查詢，這給分散式多表關聯查詢帶來的問題是資料shuffle爆炸。另外ClickHouse的執行計劃生成過程中，僅有一些簡單的filter push down，column prune規則，完全沒有join reorder能力。對使用者來說就是"所寫即所得"的模式，要求人人都是DBA，下面將結合簡單的查詢例子來介紹一下ClickHouse計算模型最大的幾個原則。
遞迴子查詢
在閱讀原始碼的過程中，我可以感受到ClickHouse前期是一個完全受母公司Yandex搜尋分析業務驅動成長起來的資料庫。而搜尋業務場景下的Metric分析（uv / pv ...），對分散式多表關聯分析的並沒有很高的需求，絕大部分業務場景都可以透過簡單的資料分表分析然後聚合結果（資料建模比較簡單），所以從一開始ClickHouse就註定不擅長處理複雜的分散式多表關聯查詢，ClickHouse的核心只是把單機（單表）分析做到了效能極致。但是任何一個業務場景下都不能完全避免分散式關聯分析的需求，ClickHouse採用了一套簡單的Rule來處理多表關聯分析的查詢。
對ClickHouse有所瞭解的同學應該知道ClickHouse採用的是簡單的節點對等架構，同時不提供任何分散式的語義保證，ClickHouse的節點中存在著兩種型別的表：本地表（真實存放資料的表引擎），分散式表（代理了多個節點上的本地表，相當於"分庫分表"的Proxy）。當ClickHouse的節點收到兩表的Join關聯分析時，問題比較收斂，無非是以下幾種情況：本地表 Join 分散式表 、本地表 Join 本地表、 分散式表 Join 分散式表、分散式表 Join 本地表，這四種情況會如何執行這裡先放一下，等下一小節再介紹。
接下來問題複雜化，如何解決多個Join的關聯查詢？ClickHouse採用遞迴子查詢來解決這個問題，如下面的簡單例子所示ClickHouse會自動把多個Join的關聯查詢改寫成子查詢進行巢狀, 規則非常簡單：1）Join的左右表必須是本地表、分散式表或者子查詢；2）傾向把Join的左側變成子查詢；3）從最後一個Join開始遞迴改寫成子查詢；4）對Join order不做任何改動；5）可以自動根據where條件改寫Cross Join到Inner Join。下面是兩個具體的例子幫助大家理解：
例1
select * from local_tabA join (select * from dist_tabB join local_tabC on dist_tabB.key2 = local_tabC.key2) as sub_Q1on local_tabA.key1 = sub_Q1.key1join dist_tabD on local_tabA.key1 = dist_tabD.key1;=============>select * from (select * from local_tabA join (select * from dist_tabB join local_tabC on dist_tabB.key2 = local_tabC.key2) as sub_Q1on local_tabA.key1 = sub_Q1.key1) as sub_Q2join dist_tabD on sub_Q2.key1 = dist_tabD.key1;例2
select * from local_tabA join (select * from dist_tabB join local_tabC on dist_tabB.key2 = local_tabC.key2) as sub_Q1on local_tabA.key1 = sub_Q1.key1join dist_tabD on local_tabA.key1 = dist_tabD.key1;=============>select * from (select * from local_tabA join (select * from dist_tabB join local_tabC on dist_tabB.key2 = local_tabC.key2) as sub_Q1on local_tabA.key1 = sub_Q1.key1) as sub_Q2join dist_tabD on sub_Q2.key1 = dist_tabD.key1;Join關聯中的子查詢在計算引擎裡就相關於是一個本地的"臨時表"，只不過這個臨時表的Input Stream對接的是一個子查詢的Output Stream。所以在處理多個Join的關聯查詢時，ClickHouse會把查詢拆成遞迴的子查詢，每一次遞迴只處理一個Join關聯，單個Join關聯中，左右表輸入有可能是本地表、分散式表、子查詢，這樣問題就簡化了。
這種簡單的遞迴子查詢解決方案存在最致命的缺陷是：
（1）系統沒有自動最佳化能力，Join reorder是最佳化器的重要課題，但是ClickHouse完全不提供這個能力，對核心不夠了解的使用者基本無法寫出效能最佳的關聯查詢，但是對經驗老道的工程師來說這是另一種體驗：可以完全掌控SQL的執行計劃。
（2）無法完全發揮分散式計算的能力，ClickHouse在兩表的Join關聯中能否利用分散式算力進行join計算取決於左表是否是分散式表，只有當左表是分散式表時才有可能利用上Cluster的計算能力，也就是左表是本地表或者子查詢時Join計算過程只在一個節點進行。
（3）多個大表的Join關聯容易引起節點的OOM，ClickHouse中的Hash Join運算元目前不支援spill（落盤），遞迴子查詢需要節點在記憶體中同時維護多個完整的Hash Table來完成最後的Join關聯。
兩表Join規則
上一節介紹了ClickHouse如何利用遞迴子查詢來解決多個Join的關聯分析，最終系統只會focus在單個Join的關聯分析上。除了常規的Join方式修飾詞以外，ClickHouse還引入了另外一個Join流程修飾詞"Global"，它會影響整個Join的執行計劃。節點真正採用Global Join進行關聯的前提條件是左表必須是分散式表，Global Join會構建一個記憶體臨時表來儲存Join右測的資料，然後把左表的Join計算任務分發給所有代理的儲存節點，收到Join計算任務的儲存節點會跨節點複製記憶體臨時表的資料，用以構建Hash Table。
下面依次介紹所有可能出現的單個Join關聯分析場景：
（1）（本地表/子查詢）Join（本地表/子查詢）：常規本地Join，Global Join不生效
（2）（本地表/子查詢）Join（分散式表）：分散式表資料全部讀到當前節點進行Hash Table構建，Global Join不生效
（3）（分散式表）Join（本地表/子查詢）：Join計算任務分發到分散式表的所有儲存節點上，儲存節點上收到的Join右表取決於是否採用Global Join策略，如果不是Global Join則把右測的（本地表名/子查詢）直接轉給所有儲存節點。如果是Global Join則當前節點會構建Join右測資料的記憶體表，收到Join計算任務的節點會來拉取這個記憶體表資料。
（4）（分散式表）Join（分散式表）：Join計算任務分發到分散式表的所有儲存節點上，儲存節點上收到的Join右表取決於是否採用Global Join策略，如果不是Global Join則把右測的分散式表名直接轉給所有儲存節點。如果是Global Join則當前節點會把右測分散式表的資料全部收集起來構建記憶體表，收到Join計算任務的節點會來拉取這個記憶體表資料。
從上面可以看出只有分散式表的Join關聯是可以進行分散式計算的，Global Join可以提前計算Join右測的結果資料構建記憶體表，當Join右測是帶過濾條件的分散式表或者子查詢時，降低了Join右測資料重複計算的次數，還有一種場景是Join右表只在當前節點存在則此時必須使用Global Join把它替換成記憶體臨時表，因為直接把右表名轉給其他節點一定會報錯。
ClickHouse中還有一個開關和Join關聯分析的行為有關：distributed_product_mode，它只是一個簡單的查詢改寫Rule用來改寫兩個分散式表的Join行為。當set distributed_product_mode = 'LOCAL'時，它會把右表改寫成代理的儲存表名，這要求左右表的資料分割槽對齊，否則Join結果就出錯了，當set distributed_product_mode = 'GLOBAL'時，它會把自動改寫Join到Global Join。但是這個改寫Rule只針對左右表都是分散式表的case，複雜的多表關聯分析場景下對SQL的最佳化作用比較小，還是不要去依賴這個自動改寫的能力。
ClickHouse的分散式Join關聯分析中還有另外一個特點是它並不會對左表的資料進行re-sharding，每一個收到Join任務的節點都會要全量的右表資料來構建Hash Table。在一些場景下，如果使用者確定Join左右表的資料是都是按照某個Join key分割槽的，則可以使用（分散式表）Join（本地表）的方式來緩解一下這個問題。但是ClickHouse的分散式表Sharding設計並不保證Cluster在調整節點後資料能完全分割槽對齊，這是使用者需要注意的。
小結
總結一下上面兩節的分析，ClickHouse當前的MPP計算模型並不擅長做多表關聯分析，主要存在的問題：1）節點間資料shuffle膨脹，Join關聯時沒有資料re-sharding能力，每個計算節點都需要shuffle全量右表資料；2）Join記憶體膨脹，原因同上；3）非Global Join下可能引起計算風暴，計算節點重複執行子查詢；4）沒有Join reorder最佳化。其中的1和3還會隨著節點數量增長變得更加明顯。在多表關聯分析的場景下，使用者應該儘可能為小表構建Dictionary，並使用dictGet內建函式來代替Join，針對無法避免的多表關聯分析應該直接寫成巢狀子查詢的方式，並根據真實的查詢執行情況嘗試調整Join order尋找最優的執行計劃。當前ClickHouse的MPP計算模型下，仍然存在不少查詢最佳化的小"bug"可能導致效能不如預期，例如列裁剪沒有下推，過濾條件沒有下推，partial agg沒有下推等等，不過這些小問題都是可以修復。
資源隔離現狀當前的ClickHouse開源版本在系統的資源管理方面已經做了很多的feature，我把它們總結為三個方面：全鏈路（執行緒-》查詢-》使用者）的資源使用追蹤、查詢&使用者級別資源隔離、資源使用限流。對於ClickHouse的資深DBA來說，這些資源追蹤、隔離、限流功能已經可以解決非常多的問題。接下來我將展開介紹一下ClickHouse在這三個方面的功能設計實現。
trace & profile
ClickHouse的資源使用都是從查詢thread級別就開始進行追蹤，主要的相關程式碼在 ThreadStatus 類中。每個查詢執行緒都會有一個thread local的ThreadStatus物件，ThreadStatus物件中包含了對記憶體使用追蹤的 MemoryTracker、profile cpu time的埋點物件 ProfileEvents、以及監控thread 熱點執行緒棧的 QueryProfiler。
1.MemoryTracker
ClickHouse中有很多不同level的MemoryTracker，包括執行緒級別、查詢級別、使用者級別、server級別，這些MemoryTracker會透過parent指標組織成一個樹形結構，把記憶體申請釋放資訊層層反饋上去。
MemoryTrack中還有額外的峰值資訊（peak）統計，記憶體上限檢查，一旦某個查詢執行緒的申請記憶體請求在上層（查詢級別、使用者級別、server級別）MemoryTracker遇到超過限制錯誤，查詢執行緒就會丟擲OOM異常導致查詢退出。同時查詢執行緒的MemoryTracker每申請一定量的記憶體都會統計出當前的工作棧，非常方便排查記憶體OOM的原因。
ClickHouse的MPP計算引擎中每個查詢的主執行緒都會有一個ThreadGroup物件，每個MPP引擎worker執行緒在啟動時必須要attach到ThreadGroup上，線上程退出時detach，這保證了整個資源追蹤鏈路的完整傳遞。最後一個問題是如何把CurrentThread::MemoryTracker hook到系統的記憶體申請釋放上去？ClickHouse首先是過載了c++的new_delete operator，其次針對需要使用malloc的一些場景封裝了特殊的Allocator同步記憶體申請釋放。為了解決記憶體追蹤的效能問題，每個執行緒的記憶體申請釋放會在thread local變數上進行積攢，最後以大塊記憶體的形式同步給MemoryTracker。
class MemoryTracker{ std::atomic<Int64> amount {0}; std::atomic<Int64> peak {0}; std::atomic<Int64> hard_limit {0}; std::atomic<Int64> profiler_limit {0}; Int64 profiler_step = 0; /// Singly-linked list. All information will be passed to subsequent memory trackers also (it allows to implement trackers hierarchy). /// In terms of tree nodes it is the list of parents. Lifetime of these trackers should "include" lifetime of current tracker. std::atomic<MemoryTracker *> parent {}; /// You could specify custom metric to track memory usage. CurrentMetrics::Metric metric = CurrentMetrics::end(); ...}2.ProfileEvents:
ProfileEvents顧名思義，是監控系統的profile資訊，覆蓋的資訊非常廣，所有資訊都是透過程式碼埋點進行收集統計。它的追蹤鏈路和MemoryTracker一樣，也是透過樹狀結構組織層層追蹤。其中和cpu time相關的核心指標包括以下：
///Total (wall clock) time spent in processing thread.RealTimeMicroseconds; ///Total time spent in processing thread executing CPU instructions in user space.///This include time CPU pipeline was stalled due to cache misses, branch mispredictions, hyper-threading, etc.UserTimeMicroseconds; ///Total time spent in processing thread executing CPU instructions in OS kernel space.///This include time CPU pipeline was stalled due to cache misses, branch mispredictions, hyper-threading, etc.SystemTimeMicroseconds; SoftPageFaults;HardPageFaults;///Total time a thread spent waiting for a result of IO operation, from the OS point of view.///This is real IO that doesn't include page cache.OSIOWaitMicroseconds;///Total time a thread was ready for execution but waiting to be scheduled by OS, from the OS point of view.OSCPUWaitMicroseconds; ///CPU time spent seen by OS. Does not include involuntary waits due to virtualization.OSCPUVirtualTimeMicroseconds;///Number of bytes read from disks or block devices.///Doesn't include bytes read from page cache. May include excessive data due to block size, readahead, etc.OSReadBytes; ///Number of bytes written to disks or block devices.///Doesn't include bytes that are in page cache dirty pages. May not include data that was written by OS asynchronouslyOSWriteBytes; ///Number of bytes read from filesystem, including page cacheOSReadChars; ///Number of bytes written to filesystem, including page cacheOSWriteChars;以上這些資訊都是從linux系統中直接採集，參考 sys/resource.h 和 linux/taskstats.h。採集沒有固定的頻率，系統在查詢計算的過程中每處理完一個Block的資料就會依據距離上次採集的時間間隔決定是否採集最新資料。
3.QueryProfiler:
QueryProfiler的核心功能是抓取查詢執行緒的熱點棧，ClickHouse透過對執行緒設定timer_create和自定義的signal_handler讓worker執行緒定時收到SIGUSR訊號量記錄自己當前所處的棧，這種方法是可以抓到所有被lock block或者sleep的執行緒棧的。
除了以上三種執行緒級別的trace&profile機制，ClickHouse還有一套server級別的Metrics統計，也是透過程式碼埋點記錄系統中所有Metrics的瞬時值。ClickHouse底層的這套trace&profile手段保障了使用者可以很方便地從系統硬體層面去定位查詢的效能瓶頸點或者OOM原因，所有的metrics, trace, profile資訊都有物件的system_log系統表可以追溯歷史。
資源隔離
資源隔離需要關注的點包括記憶體、CPU、IO，目前ClickHouse在這三個方面都做了不同程度功能：
1.記憶體隔離
當前使用者可以透過max_memory_usage（查詢記憶體限制），max_memory_usage_for_user（使用者的記憶體限制），max_memory_usage_for_all_queries（server的記憶體限制），max_concurrent_queries_for_user（使用者併發限制），max_concurrent_queries（server併發限制）這一套引數去規劃系統的記憶體資源使用做到使用者級別的隔離。但是當用戶進行多表關聯分析時，系統派發的子查詢會突破使用者的資源規劃，所有的子查詢都屬於default使用者，可能引起使用者查詢的記憶體超用。
2.CPU隔離
ClickHouse提供了Query級別的CPU優先順序設定，當然也可以為不同使用者的查詢設定不同的優先順序，有以下兩種優先順序引數：
///Priority of the query.///1 - higher value - lower priority; 0 - do not use priorities.///Allows to freeze query execution if at least one query of higher priority is executed.priority;///If non zero - set corresponding 'nice' value for query processing threads.///Can be used to adjust query priority for OS scheduler.os_thread_priority;3.IO隔離
ClickHouse目前在IO上沒有做任何隔離限制，但是針對非同步merge和查詢都做了各自的IO限制，儘量避免IO打滿。隨著非同步merge task數量增多，系統會開始限制後續單個merge task涉及到的Data Parts的disk size。在查詢並行讀取MergeTree data的時候，系統也會統計每個執行緒當前的IO吞吐，如果吞吐不達標則會反壓讀取執行緒，降低讀取執行緒數緩解系統的IO壓力，以上這些限制措施都是從區域性來緩解問題的一個手段。
Quota限流
除了靜態的資源隔離限制，ClickHouse內部還有一套時序資源使用限流機制--Quota。使用者可以根據查詢的使用者或者Client IP對查詢進行分組限流。限流和資源隔離不同，它是約束查詢執行的"速率"，當前主要包括以下幾種"速率"：
QUERIES; /// Number of queries.ERRORS; /// Number of queries with exceptions.RESULT_ROWS; /// Number of rows returned as result.RESULT_BYTES; /// Number of bytes returned as result.READ_ROWS; /// Number of rows read from tables.READ_BYTES; /// Number of bytes read from tables.EXECUTION_TIME; /// Total amount of query execution time in nanoseconds.使用者可以自定義規劃自己的限流策略，防止系統的負載（IO、網路、CPU）被打爆，Quota限流可以認為是系統自我保護的手段。系統會根據查詢的使用者名稱、IP地址或者Quota Key Hint來為查詢繫結對應的限流策略。計算引擎在運算元之間傳遞Block時會檢查當前Quota組內的流速是否過載，進而透過sleep查詢執行緒來降低系統負載。
小結
總結一下ClickHouse在資源隔離/trace層面的優缺點：ClickHouse為使用者提供了非常多的工具元件，但是欠缺整體性的解決方案。以trace & profile為例，ClickHouse在自身系統裡集成了非常完善的trace / profile / metrics日誌和瞬時狀態系統表，在排查效能問題的過程中它的鏈路是完備的。但問題是這個鏈路太複雜了，對一般使用者來說排查非常困難，尤其是碰上遞迴子查詢的多表關聯分析時，需要從使用者查詢到一層子查詢到二層子查詢步步深入分析。當前的資源隔離方案呈現給使用者的更加是一堆配置，根本不是一個完整的功能。Quota限流雖然是一個完整的功能，但是卻不容易使用，因為使用者不知道如何量化合理的"速率"。
彈性資源佇列第一章為大家介紹了ClickHouse的MPP計算模型，核心想闡述的點是ClickHouse這種簡單的遞迴子查詢計算模型在資源利用上是非常粗暴的，如果沒有很好的資源隔離和系統過載保護，節點很容易就會因為bad sql變得不穩定。第二章介紹ClickHouse當前的資源使用trace profile功能、資源隔離功能、Quota過載保護。但是ClickHouse目前在這三個方面做得都不夠完美，還需要深度打磨來提升系統的穩定性和資源利用率。我認為主要從三個方面進行加強：效能診斷鏈路自動化使使用者可以一鍵診斷，資源佇列功能加強，Quota（負載限流）做成自動化並拉通來看查詢、寫入、非同步merge任務對系統的負載，防止過載。
阿里雲資料庫ClickHouse在ClickHouse開源版本上即將推出使用者自定義的彈性資源佇列功能，資源佇列DDL定義如下：
CREATE RESOURCE QUEUE [IF NOT EXISTS | OR REPLACE] test_queue [ON CLUSTER cluster]memory=10240000000, ///資源佇列的總記憶體限制concurrency=8, ///資源佇列的查詢併發控制isolate=0, ///資源佇列的記憶體搶佔隔離級別priority=high ///資源佇列的cpu優先順序和記憶體搶佔優先順序TO {role [,...] | ALL | ALL EXCEPT role [,...]};我認為資源佇列的核心問題是要在保障使用者查詢穩定性的基礎上最大化系統的資源利用率和查詢吞吐。傳統的MPP資料庫類似GreenPlum的資源佇列設計思想是佇列之間的記憶體資源完全隔離，透過最佳化器去評估每一個查詢的複雜度加上佇列的預設併發度來決定查詢在佇列中可佔用的記憶體大小，在查詢真實開始執行之前已經限定了它可使用的記憶體，加上GreenPlum強大的計算引擎所有運算元都可以落盤，使得資源佇列可以保障系統內的查詢穩定執行，但是吞吐並不一定是最大化的。因為GreenPlum資源佇列之間的記憶體不是彈性的，有佇列空閒下來它的記憶體資源也不能給其他佇列使用。拋開資源佇列間的彈性問題，其要想做到單個資源佇列內的查詢穩定高效執行離不開Greenplum的兩個核心能力：CBO最佳化器智慧評估出查詢需要佔用的記憶體，全運算元可落盤的計算引擎。
ClickHouse目前的現狀是：1）沒有最佳化器幫助評估查詢的複雜度，2）整個計算引擎的落盤能力比較弱，在限定記憶體的情況下無法保障query順利執行。因此我們結合ClickHouse計算引擎的特色，設計了一套彈性資源佇列模型，其中核心的彈性記憶體搶佔原則包括以下幾個：
對資源佇列內的查詢不設記憶體限制佇列中的查詢在申請記憶體時如果遇到記憶體不足，則嘗試從優先順序更低的佇列中搶佔式申請記憶體在2）中記憶體搶佔過程中，如果搶佔申請失敗，則檢查自己所屬的資源佇列是否被其他查詢搶佔記憶體，嘗試kill搶佔記憶體最多的查詢回收記憶體資源如果在3）中嘗試回收被搶佔記憶體資源失敗，則當前查詢報OOM Exception每個資源佇列預留一定比例的記憶體不可搶佔，當資源佇列中的查詢負載到達一定水位時，記憶體就變成完全不可被搶佔。同時使用者在定義資源佇列時，isolate=0的佇列是允許被搶佔的，isolate=1的佇列不允許被搶佔，isolate=2的佇列不允許被搶佔也不允許搶佔其他隊列當資源佇列中有查詢OOM失敗，或者因為搶佔記憶體被kill，則把當前資源佇列的併發數臨時下調，等系統恢復後再逐步上調。ClickHouse彈性資源佇列的設計原則就是允許記憶體資源搶佔來達到資源利用率的最大化，同時動態調整資源佇列的併發限制來防止bad query出現時導致使用者的查詢大面積失敗。由於計算引擎的約束限制，目前無法保障查詢完全沒有OOM，但是使用者端可以透過錯誤資訊來判斷查詢是否屬於bad sql，同時對誤殺的查詢進行retry。