注：以下分析基于開源 v19.15.2.2-stable 版本進行

引言

ClickHouse內(nèi)核分析系列文章，繼上一篇文章?MergeTree查詢鏈路?之后，這次我將為大家介紹MergeTree存儲引擎的異步Merge和Mutation機制。建議讀者先補充上一篇文章的基礎(chǔ)知識，這樣會比較容易理解。

MergeTree Mutation功能介紹

在上一篇系列文章中，我已經(jīng)介紹過ClickHouse內(nèi)核中的MergeTree存儲一旦生成一個Data Part，這個Data Part就不可再更改了。所以從MergeTree存儲內(nèi)核層面，ClickHouse就不擅長做數(shù)據(jù)更新刪除操作。但是絕大部分用戶場景中，難免會出現(xiàn)需要手動訂正、修復(fù)數(shù)據(jù)的場景。所以ClickHouse為用戶設(shè)計了一套離線異步機制來支持低頻的Mutation(改、刪)操作。

Mutation命令執(zhí)行

ALTER TABLE [db.]table DELETE WHERE filter_expr; ALTER TABLE [db.]table UPDATE column1 = expr1 [, ...] WHERE filter_expr;

ClickHouse的方言把Delete和Update操作也加入到了Alter Table的范疇中，它并不支持裸的Delete或者Update操作。當用戶執(zhí)行一個如上的Mutation操作獲得返回時，ClickHouse內(nèi)核其實只做了兩件事情：

檢查Mutation操作是否合法；

保存Mutation命令到存儲文件中，喚醒一個異步處理merge和mutation的工作線程；

兩者的主體邏輯分別在MutationsInterpreter::validate函數(shù)和StorageMergeTree::mutate函數(shù)中。

MutationsInterpreter::validate函數(shù)dry run一個異步Mutation執(zhí)行的全過程，其中涉及到檢查Mutation是否合法的判斷原則是列值更新后記錄的分區(qū)鍵和排序鍵不能有變化。因為分區(qū)鍵和排序鍵一旦發(fā)生變化，就會導(dǎo)致多個Data Part之間之間Merge邏輯的復(fù)雜化。剩余的Mutation執(zhí)行過程可以看做是打開一個Data Part的BlockInputStream，在這個BlockStream的基礎(chǔ)上封裝刪除操作的FilterBlockInputStream，再加上更新操作的ExpressionBlockInputStream，最后把數(shù)據(jù)通過BlockOutputStream寫回到新的Data Part中。這里簡單介紹一下ClickHouse的計算層實現(xiàn)，整體上它是一個火山模型的計算引擎，數(shù)據(jù)的各種filer、投影、join、agg都是通過BlockStrem抽象實現(xiàn)，在BlockStream中數(shù)據(jù)是按照Block進行傳輸處理的，而Block中的數(shù)據(jù)又是按照列模式組織，這使得ClickHouse在單列的計算上可以批量化并使用一些SIMD指令加速。BlockOutputStream承擔了MergeTree Data Part列存寫入和索引構(gòu)建的全部工作，我會在后續(xù)的文章中會詳細展開介紹ClickHouse計算層中各類功能的BlockStream，以及BlockOutputStream中構(gòu)建索引的實現(xiàn)細節(jié)。

在Mutation命令的執(zhí)行過程中，我們可以看到MergeTree會把整條Alter命令保存到存儲文件夾下，然后創(chuàng)建一個MergeTreeMutationEntry對象保存到表的待修改狀態(tài)中，最后喚醒一個異步處理merge和 mutation的工作線程。這里有一個關(guān)鍵的問題，因為Mutation的實際操作是異步發(fā)生的，在用戶的Alter命令返回之后仍然會有數(shù)據(jù)寫入，系統(tǒng)如何在異步訂正的過程中排除掉Alter命令之后寫入的數(shù)據(jù)呢？下一節(jié)中我會介紹MergeTree中Data Part的Version機制，它可以在Data Part級別解決上面的問題。但是因為ClickHouse寫入鏈路的異步性，ClickHouse仍然無法保證Alter命令前Insert的每條紀錄都被更新，只能確保Alter命令前已經(jīng)存在的Data Part都會被訂正，推薦用戶只用來訂正T+1場景的離線數(shù)據(jù)。

異步Merge&Mutation

Batch Insert和Mutation的數(shù)據(jù)一致性

struct MergeTreePartInfo {String partition_id;Int64 min_block = 0;Int64 max_block = 0;UInt32 level = 0;Int64 mutation = 0; /// If the part has been mutated or contains mutated parts, is equal to mutation version number..../// Get block number that can be used to determine which mutations we still need to apply to this part/// (all mutations with version greater than this block number).Int64 getDataVersion() const { return mutation ? mutation : min_block; }... bool operator<(const MergeTreePartInfo & rhs) const{return std::forward_as_tuple(partition_id, min_block, max_block, level, mutation)< std::forward_as_tuple(rhs.partition_id, rhs.min_block, rhs.max_block, rhs.level, rhs.mutation);} }

在具體展開MergeTree的異步merge和mutation機制之前，先需要詳細介紹一下MergeTree中對Data Part的管理方式。每個Data Part都有一個MergeTreePartInfo對象來保存它的meta信息，MergeTreePartInfo類的結(jié)構(gòu)如上方代碼所示。

partition_id：表示所屬的數(shù)據(jù)分區(qū)id。

min_block、max_block：blockNumber是數(shù)據(jù)寫入的一個版本信息，在上一篇系列文章中講過，用戶每次批量寫入的數(shù)據(jù)都會生成一個Data Part。同一批寫入的數(shù)據(jù)會被assign一個唯一的blockNumber，而這個blockNumber是在MergeTree表級別自增的。以及MergeTree在merge多個Data Part的時候會準守一個原則：在同一個數(shù)據(jù)分區(qū)下選擇blockNumber區(qū)間相鄰的若干個Data Parts進行合并，不會出現(xiàn)在同一個數(shù)據(jù)分區(qū)下Data Parts之間的blockNumber區(qū)間出現(xiàn)重合。所以Data Part中的min_block和max_block可以表示當前Data Part中數(shù)據(jù)的版本范圍。

level：表示Data Part所在的層級，新寫入的Data Part都屬于level 0。異步merge多個Data Part的過程中，系統(tǒng)會選擇其中最大的level + 1作為新Data Part的level。這個信息可以一定程度反映出當前的Data Part是經(jīng)歷了多少次merge，但是不能準確表示，核心原因是MergeTree允許多個Data Part跨level進行merge的，為了最終一個數(shù)據(jù)分區(qū)內(nèi)的數(shù)據(jù)merge成一個Data Part。

mutation：和批量寫入數(shù)據(jù)的版本號機制類似，MergeTree表的mutation命令也會被assign一個唯一的blockNumber作為版本號，這個版本號信息會保存在MergeTreeMutationEntry中，所以通過版本號信息我們可以看出數(shù)據(jù)寫入和mutation命令之間的先后關(guān)系。Data Part中的這個mutation表示的則是當前這個Data Part已經(jīng)完成的mutation操作，對每個Data Part來說它是按照mutation的blockNumber順序依次完成所有的mutation。

解釋了MergeTreePartInfo類中的信息含義，我們就可以理解上一節(jié)中遺留的異步Mutation如何選擇哪些Data Parts需要訂正的問題。系統(tǒng)可以通過MergeTreePartInfo::getDataVersion() { return mutation ? mutation : min_block }函數(shù)來判斷當前Data Part是否需要進行某個mutation訂正，比較兩者version即可。

Merge&Mutation工作任務(wù)

ClickHouse內(nèi)核中異步merge、mutation工作由統(tǒng)一的工作線程池來完成，這個線程池的大小用戶可以通過參數(shù)background_pool_size進行設(shè)置。線程池中的線程Task總體邏輯如下，可以看出這個異步Task主要做三塊工作：清理殘留文件，merge Data Parts 和 mutate Data Part。

BackgroundProcessingPoolTaskResult StorageMergeTree::mergeMutateTask() {....try{/// Clear old parts. It is unnecessary to do it more than once a second.if (auto lock = time_after_previous_cleanup.compareAndRestartDeferred(1)){{/// TODO: Implement tryLockStructureForShare.auto lock_structure = lockStructureForShare(false, "");clearOldPartsFromFilesystem();clearOldTemporaryDirectories();}clearOldMutations();}///TODO: read deduplicate option from table configif (merge(false /*aggressive*/, {} /*partition_id*/, false /*final*/, false /*deduplicate*/))return BackgroundProcessingPoolTaskResult::SUCCESS;if (tryMutatePart())return BackgroundProcessingPoolTaskResult::SUCCESS;return BackgroundProcessingPoolTaskResult::ERROR;}... }

需要清理的殘留文件分為三部分：過期的Data Part，臨時文件夾，過期的Mutation命令文件。如下方代碼所示，MergeTree Data Part的生命周期包含多個階段，創(chuàng)建一個Data Part的時候分兩階段執(zhí)行Temporary->Precommitted->Commited，淘汰一個Data Part的時候也可能會先經(jīng)過一個Outdated狀態(tài)，再到Deleting狀態(tài)。在Outdated狀態(tài)下的Data Part仍然是可查的。異步Task在收集Outdated Data Part的時候會根據(jù)它的shared_ptr計數(shù)來判斷當前是否有查詢Context引用它，沒有的話才進行刪除。清理臨時文件的邏輯較為簡單，在數(shù)據(jù)文件夾中遍歷搜索"tmp_"開頭的文件夾，并判斷創(chuàng)建時長是否超過temporary_directories_lifetime。臨時文件夾主要在ClickHouse的兩階段提交過程可能造成殘留。最后是清理數(shù)據(jù)已經(jīng)全部訂正完成的過期Mutation命令文件。

enum class State{Temporary, /// the part is generating now, it is not in data_parts listPreCommitted, /// the part is in data_parts, but not used for SELECTsCommitted, /// active data part, used by current and upcoming SELECTsOutdated, /// not active data part, but could be used by only current SELECTs, could be deleted after SELECTs finishesDeleting, /// not active data part with identity refcounter, it is deleting right now by a cleanerDeleteOnDestroy, /// part was moved to another disk and should be deleted in own destructor};

Merge邏輯

StorageMergeTree::merge函數(shù)是MergeTree異步Merge的核心邏輯，Data Part Merge的工作除了通過后臺工作線程自動完成，用戶還可以通過Optimize命令來手動觸發(fā)。自動觸發(fā)的場景中，系統(tǒng)會根據(jù)后臺空閑線程的數(shù)據(jù)來啟發(fā)式地決定本次Merge最大可以處理的數(shù)據(jù)量大小，max_bytes_to_merge_at_min_space_in_pool和max_bytes_to_merge_at_max_space_in_pool參數(shù)分別決定當空閑線程數(shù)最大時可處理的數(shù)據(jù)量上限以及只剩下一個空閑線程時可處理的數(shù)據(jù)量上限。當用戶的寫入量非常大的時候，應(yīng)該適當調(diào)整工作線程池的大小和這兩個參數(shù)。當用戶手動觸發(fā)merge時，系統(tǒng)則是根據(jù)disk剩余容量來決定可處理的最大數(shù)據(jù)量。

接下來介紹merge過程中最核心的邏輯：如何選擇Data Parts進行merge？為了方便理解，這里先介紹一下Data Parts在MergeTree表引擎中的管理組織方式。上一節(jié)中提到的MergeTreePartInfo類中定義了比較操作符，MergeTree中的Data Parts就是按照這個比較操作符進行排序管理，排序鍵是(partition_id, min_block, max_block, level, mutation)，索引管理結(jié)構(gòu)如下圖所示：

自動Merge的處理邏輯，首先是通過MergeTreeDataMergerMutator::selectPartsToMerge函數(shù)篩選出本次merge要合并的Data Parts，這個篩選過程需要準守三個原則：

跨數(shù)據(jù)分區(qū)的Data Part之間不能合并；

合并的Data Parts之間必須是相鄰（在上圖的有序組織關(guān)系中相鄰），只能在排序鏈表中按段合并，不能跳躍；

合并的Data Parts之間的mutation狀態(tài)必須是一致的，如果Data Part A 后續(xù)還需要完成mutation-23而Data Part B后續(xù)不需要完成mutation-23(數(shù)據(jù)全部是在mutation命令之后寫入或者已經(jīng)完成mutation-23)，則A和B不能進行合并；

所以我們上面的Data Parts組織關(guān)系邏輯示意圖中，相同顏色的Data Parts是可以合并的。雖然圖中三個不同顏色的Data Parts序列都是可以合并的，但是合并工作線程每次只會挑選其中某個序列的一小段進行合并（如前文所述，系統(tǒng)會限定每次合并的Data Parts的數(shù)據(jù)量）。對于如何從這些序列中挑選出最佳的一段區(qū)間，ClickHouse抽象出了IMergeSelector類來實現(xiàn)不同的邏輯。當前主要有兩種不同的merge策略：TTL數(shù)據(jù)淘汰策略和常規(guī)策略。

• TTL數(shù)據(jù)淘汰策略：TTL數(shù)據(jù)淘汰策略啟用的條件比較苛刻，只有當某個Data Part中存在數(shù)據(jù)生命周期超時需要淘汰，并且距離上次使用TTL策略達到一定時間間隔（默認1小時）。TTL策略也非常簡單，首先挑選出TTL超時最嚴重Data Part，把這個Data Part所在的數(shù)據(jù)分區(qū)作為要進行數(shù)據(jù)合并的分區(qū)，最后會把這個TTL超時最嚴重的Data Part前后連續(xù)的所有存在TTL過期的Data Part都納入到merge的范圍中。這個策略簡單直接，每次保證優(yōu)先合并掉最老的存在過期數(shù)據(jù)的Data Part。
• 常規(guī)策略：這里的選舉策略就比較復(fù)雜，基本邏輯是枚舉每個可能合并的Data Parts區(qū)間，通過啟發(fā)式規(guī)則判斷是否滿足合并條件，再有啟發(fā)式規(guī)則進行算分，選取分數(shù)最好的區(qū)間。啟發(fā)式判斷是否滿足合并條件的算法在SimpleMergeSelector.cpp::allow函數(shù)中，其中的主要思想分為以下幾點：系統(tǒng)默認對合并的區(qū)間有一個Data Parts數(shù)量的限制要求（每5個Data Parts才能合并）；如果當前數(shù)據(jù)分區(qū)中的Data Parts出現(xiàn)了膨脹，則適量放寬合并數(shù)量限制要求（最低可以兩兩merge）；如果參與合并的Data Parts中有很久之前寫入的Data Part，也適量放寬合并數(shù)量限制要求，放寬的程度還取決于要合并的數(shù)據(jù)量。第一條規(guī)則是為了提升寫入性能，避免在高速寫入時兩兩merge這種低效的合并方式。最后一條規(guī)則則是為了保證隨著數(shù)據(jù)分區(qū)中的Data Part老化，老齡化的數(shù)據(jù)分區(qū)內(nèi)數(shù)據(jù)全部合并到一個Data Part。中間的規(guī)則更多是一種保護手段，防止因為寫入和頻繁mutation的極端情況下，Data Parts出現(xiàn)膨脹。啟發(fā)式算法的策略則是優(yōu)先選擇IO開銷最小的Data Parts區(qū)間完成合并，盡快合并掉小數(shù)據(jù)量的Data Parts是對在線查詢最有利的方式，數(shù)據(jù)量很大的Data Parts已經(jīng)有了很較好的數(shù)據(jù)壓縮和索引效率，合并操作對查詢帶來的性價比較低。

Mutation邏輯

StorageMergeTree::tryMutatePart函數(shù)是MergeTree異步mutation的核心邏輯，主體邏輯如下。系統(tǒng)每次都只會訂正一個Data Part，但是會聚合多個mutation任務(wù)批量完成，這點實現(xiàn)非常的棒。因為在用戶真實業(yè)務(wù)場景中一次數(shù)據(jù)訂正邏輯中可能會包含多個Mutation命令，把這多個mutation操作聚合到一起訂正效率上就非常高。系統(tǒng)每次選擇一個排序鍵最小的并且需要訂正Data Part進行操作，本意上就是把數(shù)據(jù)從前往后進行依次訂正。

Mutation功能是MergeTree表引擎最新推出一大功能，從我個人的角度看在實現(xiàn)完備度上還有一下兩點需要去優(yōu)化：

mutation沒有實時可見能力。我這里的實時可見并不是指在存儲上立即原地更新，而是給用戶提供一種途徑可以立即看到數(shù)據(jù)訂正后的最終視圖確保訂正無誤。類比在使用CollapsingMergeTree、SummingMergeTree等高級MergeTree引擎時，數(shù)據(jù)還沒有完全merge到一個Data Part之前，存儲層并沒有一個數(shù)據(jù)的最終視圖。但是用戶可以通過Final查詢模式，在計算引擎層實時聚合出數(shù)據(jù)的最終視圖。這個原理對mutation實時可見也同樣適用，在實時查詢中通過FilterBlockInputStream和ExpressionBlockInputStream完成用戶的mutation操作，給用戶提供一個最終視圖。

mutation和merge相互獨立執(zhí)行。看完本文前面的分析，大家應(yīng)該也注意到了目前Data Part的merge和mutation是相互獨立執(zhí)行的，Data Part在同一時刻只能是在merge或者mutation操作中。對于MergeTree這種存儲徹底Immutable的設(shè)計，數(shù)據(jù)頻繁merge、mutation會引入巨大的IO負載。實時上merge和mutation操作是可以合并到一起去考慮的，這樣可以省去數(shù)據(jù)一次讀寫盤的開銷。對數(shù)據(jù)寫入壓力很大又有頻繁mutation的場景，會有很大幫助。

for (const auto & part : getDataPartsVector()){...size_t current_ast_elements = 0;for (auto it = mutations_begin_it; it != mutations_end_it; ++it){MutationsInterpreter interpreter(shared_from_this(), it->second.commands, global_context);size_t commands_size = interpreter.evaluateCommandsSize();if (current_ast_elements + commands_size >= max_ast_elements)break;current_ast_elements += commands_size;commands.insert(commands.end(), it->second.commands.begin(), it->second.commands.end());}auto new_part_info = part->info;new_part_info.mutation = current_mutations_by_version.rbegin()->first;future_part.parts.push_back(part);future_part.part_info = new_part_info;future_part.name = part->getNewName(new_part_info);tagger.emplace(future_part, MergeTreeDataMergerMutator::estimateNeededDiskSpace({part}), *this, true);break;}

最后在經(jīng)過后臺工作線程一輪merge和mutation操作之后，上一節(jié)中展示的MergeTree表引擎中的Data Parts可能發(fā)生的變化如下圖所示，2020-05-10數(shù)據(jù)分區(qū)下的頭兩個Data Parts被merge到了一起，并且完成了Mutation 37和Mutation 39的數(shù)據(jù)訂正，新產(chǎn)生的Data Part如紅色所示：

Clickhouse產(chǎn)品鏈接:https://www.aliyun.com/product/clickhouse

ClickHouse內(nèi)核分析系列文章：

MergeTree查詢鏈路
希望通過內(nèi)核分析系列文章，讓大家更好地了解這款世界領(lǐng)先的列式存儲分析型數(shù)據(jù)庫。

原文鏈接
本文為云棲社區(qū)原創(chuàng)內(nèi)容，未經(jīng)允許不得轉(zhuǎn)載。

總結(jié)

以上是生活随笔為你收集整理的ClickHouse内核分析-MergeTree的Merge和Mutation机制的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。