檢查 orphaned parts#

1
if (!getStoragePolicy()->isDefaultPolicy() && !skip_sanity_checks ...)
2
{
3
    // 確保所有 parts 都落在定義的 storage policy 的磁碟上
4
    // 如果發現 part 在未知磁碟，直接丟 Exception
5
}

防止「資料實體還在，但 metadata 已經指不到」，確保資料一致性。

掃描磁碟，收集 parts#

1
runner([&expected_parts, &unexpected_disk_parts, &disk_parts, this, disk_ptr]()
2
{
3
    for (auto it = disk_ptr->iterateDirectory(relative_data_path); it->isValid(); it->next())
4
    {
5
        if (auto part_info = MergeTreePartInfo::tryParsePartName(it->name(), format_version))
6
        {
7
            if (expected_parts && !expected_parts->contains(it->name()))
8
                unexpected_disk_parts.emplace_back(...);
9
            else
10
                disk_parts.emplace_back(...);
11
        }
12
    }
13
});

• 跳過 tmp*、format_version.txt、detached/ 這些特殊目錄。
• 判斷目錄名稱能否 parse 出 MergeTreePartInfo（不符合規則就當垃圾忽略）。
• 把 parts 分為 預期內（正常）與 unexpected（意外找到的）。

建立 PartLoadingTree#

1
auto loading_tree = PartLoadingTree::build(std::move(parts_to_load));

• parts 之間可能有「包含」關係（例如新 part 覆蓋了老 part 的範圍）。
• PartLoadingTree 幫助找出「最上層、最完整」的 parts 作為 active parts。

載入 active parts#

1
auto loaded_parts = loadDataPartsFromDisk(active_parts);

對於每個 active part：

• 檢查是否 broken（檔案缺失、壓縮錯誤等）
• 檢查是否 duplicate
• 判斷 index granularity（adaptive vs non-adaptive）
• 記錄是否帶有 lightweight delete、transaction metadata

這裡是把磁碟上的目錄轉換為 DataPart 物件的關鍵。

處理異常情況#

• broken parts：移動到 detached/broken-on-start/ 目錄
• duplicate parts：刪除（除非 static storage）
• unexpected parts：記錄下來，開 background task 去處理
• outdated parts：同樣丟到 background task async 載入

防呆檢查#

1
if (have_non_adaptive_parts && have_adaptive_parts && !enable_mixed_granularity_parts)
2
    throw Exception(...);

如果表裡同時有 舊的非 adaptive granularity parts 和 新的 adaptive parts，預設是不允許混用的（除非開 enable_mixed_granularity_parts 設定）。

紀錄監控 & 任務註冊#

1
LOG_DEBUG(log, "Loaded data parts ({} items) took {} seconds", data_parts_indexes.size(), watch.elapsedSeconds());
2
ProfileEvents::increment(ProfileEvents::LoadedDataParts, data_parts_indexes.size());

• 用 log 與 ProfileEvents 記錄 parts 載入耗時與數量。
• 如果 disk 全是 Readonly（例如所有磁碟都是 cold storage），會啟動一個定期 refresh parts 的任務。

INSERT 流程#

1. 使用者做一個 INSERT INTO table VALUES (...)
2. ClickHouse 把資料寫成一個 臨時 part（名字以 tmp_ 開頭），確保寫入中途崩潰不會污染active parts。
3. 寫完後呼叫 renameTempPartAndReplace(...)：
   • 把 tmp_xxx rename 成正式 part 名稱
   • 更新 in-memory 索引結構
   • 移除被它覆蓋的舊 parts
4. INSERT commit 成功 → 新 part 對查詢可見

可以達到以下的功能

1. Temp file Protection：
   • 所有 INSERT 都先寫成 tmp_ part，只有最後 rename 成功才「生效」，確保 Atomicity。
2. Optimistic Merge：
   • 插入新 part 時順便覆蓋舊的 overlapped parts，避免資料重疊，保持狀態乾淨。
3. Transaction Consistency：
   • 如果用 rename_in_transaction，就能確保多個操作同時 commit 或 rollback ，支持更複雜的 ALTER/REPLACE 操作。

鎖保護#

1
std::lock_guard lock(mutex);

保護 pending 任務隊列，避免多執行緒同時修改。

檢查當前任務數量是否超限#

1
auto & value = CurrentMetrics::values[metric];
2
if (value.load() >= static_cast<int64_t>(max_tasks_count))
3
    return false;

• 透過 CurrentMetrics 監控目前這個 executor 的 task 數量。
• 如果超過 max_tasks_count（例如限制同時執行的合併數量），就拒絕排程。

把任務放進 queue 中#

1
pending.push(std::make_shared<TaskRuntimeData>(std::move(task), metric));

• 把傳入的 task 包裝成 TaskRuntimeData 放入 pending queue。
• TaskRuntimeData 附帶 metric，用來追蹤這個 task 的執行狀態。

通知 worker thread#

1
has_tasks.notify_one();

• 喚醒一個等待中的 worker thread，讓它開始處理任務。

整理#

這個函式做了非阻塞調度、資源控制、監控 等作用，避免同時合併過多 parts，導致磁碟 I/O 打爆、所有任務數量都掛在 CurrentMetrics 上，方便監控系統看到目前 Merge/Mutation 壓力

參數#

1
FutureMergedMutatedPartPtr future_part,
2
StorageMetadataPtr metadata_snapshot,
3
MergeList::Entry * merge_entry,
4
std::unique_ptr<MergeListElement> projection_merge_list_element,
5
TableLockHolder & holder,
6
time_t time_of_merge,
7
ContextPtr context,
8
ReservationSharedPtr space_reservation,
9
bool deduplicate,
10
const Names & deduplicate_by_columns,
11
bool cleanup,
12
MergeTreeData::MergingParams merging_params,
13
MergeTreeTransactionPtr txn,
14
bool need_prefix,
15
IMergeTreeDataPart * parent_part,
16
const String & suffix

• future_part
  • 將要產生的新 part（描述來源 parts 與輸出名稱）。
  • 是合併計畫的 blueprint。
• metadata_snapshot
  • 表的 metadata 快照，包含欄位定義、索引、TTL 等資訊。
• merge_entry / projection_merge_list_element
  • 追蹤 merge 在 system.merges（或 system.part_log）的紀錄，方便監控。
• holder
  • 表鎖，保證合併過程不與 DDL 衝突。
• space_reservation
  • 磁碟空間預留，確保合併過程不會因空間不足失敗。
• deduplicate / deduplicate_by_columns
  • 是否需要做去重（適用於 ReplacingMergeTree、OPTIMIZE TABLE ... DEDUPLICATE）。
• cleanup
  • 是否在合併時執行清理（例如 TTL）。
• merging_params
  • MergeTree 的合併參數（普通、Collapsing、VersionedCollapsing、Summing、Aggregating 等）。
• txn
  • 與 transaction 結合，支援多語句 Transaction 或 MVCC。
• need_prefix / parent_part / suffix
  • 控制新 part 的命名方式，特別是在 mutation、projection merge 的情境下。

判斷式#

1
if (future_part->isResultPatch())
2
{
3
    merging_params = MergeTreeData::getMergingParamsForPatchParts();
4
    metadata_snapshot = future_part->parts.front()->getMetadataSnapshot();
5
}

• 如果這是一個 patch merge（例如 mutation 補丁），就調整 merging_params，並從來源 part 重新抓 metadata。

回傳值#

1
return std::make_shared<MergeTask>(...);

• 最後建立 MergeTask，包含所有合併需要的上下文：
  • 要合併哪些 parts (future_part)
  • 怎麼合併（merging_params、deduplicate、cleanup）
  • 合併過程需要的資源（space_reservation、holder）
  • log tracing（merge_entry、projection_merge_list_element）

之後 background thread pool 會呼叫 MergeTask::execute()，真正讀取來源 parts → 做合併 → 輸出新 part。

所以 mergePartsToTemporaryPart(...) 的角色：

1. 不是直接合併資料，而是 建立一個 MergeTask。
2. MergeTask 帶齊所有 context：parts、metadata、transaction、磁碟空間、dedup、TTL、log。
3. Background Executor 會排程並執行這個 MergeTask，最後產生新的 臨時 part。
4. 臨時 part 成功後，透過 renameTempPartAndReplace(...) 變成正式 part。

取出當前階段#

1
chassert(stages_iterator != stages.end());
2
const auto & current_stage = *stages_iterator;

• MergeTask 持有一個 stages 向量（pipeline 概念）。
• stages_iterator 指向當前要執行的 Stage。

嘗試執行當前 Stage#

1
if (current_stage->execute())
2
    return true;

• 如果當前 Stage 還沒做完，回傳 true → 表示下次還要繼續跑這個 Stage。
• 注意：Stage 本身可能是耗時操作（例如讀取、壓縮、寫入），會分批執行。

Stage 結束 → 記錄耗時#

1
UInt64 current_elapsed_ms = global_ctx->merge_list_element_ptr->watch.elapsedMilliseconds();
2
UInt64 stage_elapsed_ms = current_elapsed_ms - global_ctx->prev_elapsed_ms;
3
global_ctx->prev_elapsed_ms = current_elapsed_ms;

• 記錄從上次到現在為止的時間差 = 本階段耗時。
• global_ctx 是 merge 任務的全域上下文，裡面有 ProfileEvents 追蹤器。

更新 ProfileEvents#

1
if (global_ctx->parent_part == nullptr)
2
{
3
    ProfileEvents::increment(current_stage->getTotalTimeProfileEvent(), stage_elapsed_ms);
4
    ProfileEvents::increment(ProfileEvents::MergeTotalMilliseconds, stage_elapsed_ms);
5
}

• 如果不是 projection 合併，更新總耗時統計。
• 每個 Stage 都有對應的 ProfileEvent，例如 MergeReadBlocks、MergeWriteBlocks。

移動到下一個 Stage#

1
++stages_iterator;
2
if (stages_iterator == stages.end())
3
    return false;

• 移動迭代器，進入下一階段。
• 如果已經到 stages.end() → 合併全部完成，回傳 false。

初始化下一個 Stage#

1
(*stages_iterator)->setRuntimeContext(std::move(next_stage_context), global_ctx);
2
return true;

• 把上一個 Stage 的輸出 context 傳給下一個 Stage（類似 pipeline）。
• 回傳 true → 表示還有 Stage 要跑。

例外處理#

1
catch (...)
2
{
3
    merge_failures.withLabels({String(ErrorCodes::getName(getCurrentExceptionCode()))}).increment();
4
    throw;
5
}

• 如果合併失敗（I/O 錯誤、磁碟滿、資料壞掉），紀錄失敗次數並拋出 Exception。
• merge_failures 是一個 metrics counter，用來統計失敗的類型。