[​DuckDB] 多核運算元並行的源碼解析

来源:https://www.cnblogs.com/happenlee/archive/2023/02/12/17113749.html
-Advertisement-
Play Games

DuckDB 是近年來頗受關註的OLAP資料庫，號稱是OLAP領域的SQLite，以精巧簡單，性能優異而著稱。筆者前段時間在調研Doris的Pipeline的運算元並行方案，而DuckDB基於論文《Morsel-Driven Parallelism: A NUMA-Aware Query Evalua ...


DuckDB 是近年來頗受關註的OLAP資料庫，號稱是OLAP領域的SQLite，以精巧簡單，性能優異而著稱。筆者前段時間在調研Doris的Pipeline的運算元並行方案，而DuckDB基於論文《Morsel-Driven Parallelism: A NUMA-Aware Query Evaluation Framework for the Many-Core Age》實現SQL運算元的高效並行化的Pipeline執行引擎，所以筆者花了一些時間進行了學習和總結，這裡結合了Mark Raasveldt進行的分享和原始代碼來一一剖析DuckDB在執行運算元並行上的具體實現。

1. 基礎知識

問題1：並行task的數目由什麼決定 ？

image.png

Pipeline的核心是：Morsel-Driven，數據是拆分成了小部分的數據。所以並行Task的核心是：能夠利用多線程來處理數據，每一個數據拆分為小部分，所以拆分並行的數目由Source決定。

DuckDB在GlobalSource上實現了一個虛函數MaxThread來決定task數目：

image.png

每一個運算元的GlobalSource抽象了自己的並行度：

image.png

問題2：並行task的怎麼樣進行多線程同步：

  • 多線程的競爭只會發生在SinkOperator上，也就是Pipeline的尾端。
  • parallelism-aware的演算法需要實現在Sink端
  • 其他的非Sink operators (比如：Hash Join Probe, Projection, Filter等)， 不需要感知多線程同步的問題

image.png

問題3：DuckDB的是如何抽象介面的：

Sink的Opeartor 定義了兩種類型：GlobalState, LocalState

  1. GlobalState: 每個查詢的Operator全局只有一個GlobalSinkState，記錄全局部分的信息
class PhysicalOperator {
public:
	unique_ptr<GlobalSinkState> sink_state;
  1. LocalState: 每個查詢的PipelineExecutor都有一個LocalSinkState，都是局部私有
//! The Pipeline class represents an execution pipeline
class PipelineExecutor {
private:
	//! The local sink state (if any)
	unique_ptr<LocalSinkState> local_sink_state;

後續會詳細解析不同的sink之間的LocalState和GlobalState如何配合的，核心部分如下：

image.png

Sink ：處理LocalState的數據

Combine：合併LocalState到GlobalState之中

2. 核心運算元的並行

這部分進行各個運算元的源碼剖析，筆者在源碼的關鍵部分加上了中文註釋，以方便大家的理解

Sort運算元

  • Sink介面：這裡需要註意的是DuckDB排序是進行了列轉行的工作的，後續讀取時需要行轉列。Sink這部分相當於實現了部分數據的排序工作。
SinkResultType PhysicalOrder::Sink(ExecutionContext &context, GlobalSinkState &gstate_p, LocalSinkState &lstate_p,
                                   DataChunk &input) const {
	auto &lstate = (OrderLocalSinkState &)lstate_p;
        
      // keys 是排序的列block，payload是輸出的排序後數據，這裡調用LocalState的SinkChunk，進行數據的轉行，
	local_sort_state.SinkChunk(keys, payload);

	// 數據達到記憶體閾值的時候進行基數排序處理，排序之後的結果存入LocalState的本地的SortedBlock中
	if (local_sort_state.SizeInBytes() >= gstate.memory_per_thread) {
		local_sort_state.Sort(global_sort_state, true);
	}
	return SinkResultType::NEED_MORE_INPUT;
}
  • Combine介面： 加鎖，拷貝sorted block到Global State
void PhysicalOrder::Combine(ExecutionContext &context, GlobalSinkState &gstate_p, LocalSinkState &lstate_p) const {
	auto &gstate = (OrderGlobalSinkState &)gstate_p;
	auto &lstate = (OrderLocalSinkState &)lstate_p;
        // 排序剩餘記憶體中不滿的數據
	local_sort_state.Sort(*this, external || !local_sort_state.sorted_blocks.empty());

	// Append local state sorted data to this global state
	lock_guard<mutex> append_guard(lock);
	for (auto &sb : local_sort_state.sorted_blocks) {
		sorted_blocks.push_back(move(sb));
	}
}
  • MergeTask：啟動核數相同的task來進行Merge (這裡可以看出DuckDB對於多線程的使用是很激進的)， 這裡是通過Event的機制實現的
void Schedule() override {
		auto &context = pipeline->GetClientContext();
		idx_t num_threads = ts.NumberOfThreads();

		vector<unique_ptr<Task>> merge_tasks;
		for (idx_t tnum = 0; tnum < num_threads; tnum++) {
			merge_tasks.push_back(make_unique<PhysicalOrderMergeTask>(shared_from_this(), context, gstate));
		}
		SetTasks(move(merge_tasks));
	}

class PhysicalOrderMergeTask : public ExecutorTask {
public:
	TaskExecutionResult ExecuteTask(TaskExecutionMode mode) override {
		// Initialize merge sorted and iterate until done
		auto &global_sort_state = state.global_sort_state;
		MergeSorter merge_sorter(global_sort_state, BufferManager::GetBufferManager(context));
		
        // 加鎖，獲取兩路，不斷進行兩路歸併，最終完成全局排序。
	while (true) {
		{
			lock_guard<mutex> pair_guard(state.lock);
			if (state.pair_idx == state.num_pairs) {
				break;
			}
			GetNextPartition();
		}
		MergePartition();
	}
		event->FinishTask();
		return TaskExecutionResult::TASK_FINISHED;
	}

聚合運算元(這裡分析的是Prefetch Agg Operator運算元)

  • Sink介面：和Sort運算元一樣，這裡拆分為Group Chunk和Aggregate Input Chunk，可以理解為代表聚合時的key與value列。註意此時Sink介面上的聚合是在LocalSinkState上完成的。
SinkResultType PhysicalPerfectHashAggregate::Sink(ExecutionContext &context, GlobalSinkState &state,
                                                  LocalSinkState &lstate_p, DataChunk &input) const {
	lstate.ht->AddChunk(group_chunk, aggregate_input_chunk);
}


void PerfectAggregateHashTable::AddChunk(DataChunk &groups, DataChunk &payload) {
	auto address_data = FlatVector::GetData<uintptr_t>(addresses);
	memset(address_data, 0, groups.size() * sizeof(uintptr_t));
	D_ASSERT(groups.ColumnCount() == group_minima.size());

	// 計算group key列對應的entry的位置
	idx_t current_shift = total_required_bits;
	for (idx_t i = 0; i < groups.ColumnCount(); i++) {
		current_shift -= required_bits[i];
		ComputeGroupLocation(groups.data[i], group_minima[i], address_data, current_shift, groups.size());
	}

	// 通過data加上面的entry位置 + tuple的偏移量，計算出對應的記憶體地址，併進行init
	idx_t needs_init = 0;
	for (idx_t i = 0; i < groups.size(); i++) {
		D_ASSERT(address_data[i] < total_groups);
		const auto group = address_data[i];
		address_data[i] = uintptr_t(data) + address_data[i] * tuple_size;
	}
	RowOperations::InitializeStates(layout, addresses, sel, needs_init);

	// after finding the group location we update the aggregates
	idx_t payload_idx = 0;
	auto &aggregates = layout.GetAggregates();
	for (idx_t aggr_idx = 0; aggr_idx < aggregates.size(); aggr_idx++) {
		auto &aggregate = aggregates[aggr_idx];
		auto input_count = (idx_t)aggregate.child_count;
                // 進行聚合的Update操作
		RowOperations::UpdateStates(aggregate, addresses, payload, payload_idx, payload.size());
	}
}
  • Combine介面： 加鎖，merge local hash table 與 global hash table
void PhysicalPerfectHashAggregate::Combine(ExecutionContext &context, GlobalSinkState &gstate_p,
                                           LocalSinkState &lstate_p) const {
	auto &lstate = (PerfectHashAggregateLocalState &)lstate_p;
	auto &gstate = (PerfectHashAggregateGlobalState &)gstate_p;

	lock_guard<mutex> l(gstate.lock);
	gstate.ht->Combine(*lstate.ht);
}

        // local state的地址vector
	Vector source_addresses(LogicalType::POINTER);
       // global state的地址vector
	Vector target_addresses(LogicalType::POINTER);
	auto source_addresses_ptr = FlatVector::GetData<data_ptr_t>(source_addresses);
	auto target_addresses_ptr = FlatVector::GetData<data_ptr_t>(target_addresses);

	// 遍歷所有hash table的表，然後進行合併對應能夠合併的key
	data_ptr_t source_ptr = other.data;
	data_ptr_t target_ptr = data;
	idx_t combine_count = 0;
	idx_t reinit_count = 0;
	const auto &reinit_sel = *FlatVector::IncrementalSelectionVector();
	for (idx_t i = 0; i < total_groups; i++) {
		auto has_entry_source = other.group_is_set[i];
		// we only have any work to do if the source has an entry for this group
		if (has_entry_source) {
			auto has_entry_target = group_is_set[i];
			if (has_entry_target) {
				// both source and target have an entry: need to combine
				source_addresses_ptr[combine_count] = source_ptr;
				target_addresses_ptr[combine_count] = target_ptr;
				combine_count++;
				if (combine_count == STANDARD_VECTOR_SIZE) {
					RowOperations::CombineStates(layout, source_addresses, target_addresses, combine_count);
					combine_count = 0;
				}
			} else {
				group_is_set[i] = true;
				// only source has an entry for this group: we can just memcpy it over
				memcpy(target_ptr, source_ptr, tuple_size);
				// we clear this entry in the other HT as we "consume" the entry here
				other.group_is_set[i] = false;
			}
		}
		source_ptr += tuple_size;
		target_ptr += tuple_size;
	}

        // 做對應的merge操作
	RowOperations::CombineStates(layout, source_addresses, target_addresses, combine_count);

Join運算元

  • Sink介面：和Sort運算元一樣，註意此時Sink介面上的hash 表是在LocalSinkState上完成的。
SinkResultType PhysicalHashJoin::Sink(ExecutionContext &context, GlobalSinkState &gstate_p, LocalSinkState &lstate_p,
                                      DataChunk &input) const {
	auto &gstate = (HashJoinGlobalSinkState &)gstate_p;
	auto &lstate = (HashJoinLocalSinkState &)lstate_p;

	lstate.join_keys.Reset();
	lstate.build_executor.Execute(input, lstate.join_keys);
	// build the HT
	auto &ht = *lstate.hash_table;
	if (!right_projection_map.empty()) {
		// there is a projection map: fill the build chunk with the projected columns
		lstate.build_chunk.Reset();
		lstate.build_chunk.SetCardinality(input);
		for (idx_t i = 0; i < right_projection_map.size(); i++) {
			lstate.build_chunk.data[i].Reference(input.data[right_projection_map[i]]);
		}
                // 構建local state的hash 表
		ht.Build(lstate.join_keys, lstate.build_chunk)

	return SinkResultType::NEED_MORE_INPUT;
}
  • Combine介面： 加鎖，拷貝local state的hash表到global state
void PhysicalHashJoin::Combine(ExecutionContext &context, GlobalSinkState &gstate_p, LocalSinkState &lstate_p) const {
	auto &gstate = (HashJoinGlobalSinkState &)gstate_p;
	auto &lstate = (HashJoinLocalSinkState &)lstate_p;
	if (lstate.hash_table) {
		lock_guard<mutex> local_ht_lock(gstate.lock);
		gstate.local_hash_tables.push_back(move(lstate.hash_table));
	}
}
  • MergeTask：啟動核數相同的task來進行Hash table的Merge (這裡可以看出DuckDB對於多線程的使用是很激進的)， 每個任務merge一部分Block(DuckDB之中的行數據，落盤使用）
void Schedule() override {
		auto &context = pipeline->GetClientContext();

		vector<unique_ptr<Task>> finalize_tasks;
		auto &ht = *sink.hash_table;
		const auto &block_collection = ht.GetBlockCollection();
		const auto &blocks = block_collection.blocks;
		const auto num_blocks = blocks.size();
		if (block_collection.count < PARALLEL_CONSTRUCT_THRESHOLD && !context.config.verify_parallelism) {
			// Single-threaded finalize
			finalize_tasks.push_back(
			    make_unique<HashJoinFinalizeTask>(shared_from_this(), context, sink, 0, num_blocks, false));
		} else {
			// Parallel finalize
			idx_t num_threads = TaskScheduler::GetScheduler(context).NumberOfThreads();
			auto blocks_per_thread = MaxValue<idx_t>((num_blocks + num_threads - 1) / num_threads, 1);

			idx_t block_idx = 0;
			for (idx_t thread_idx = 0; thread_idx < num_threads; thread_idx++) {
				auto block_idx_start = block_idx;
				auto block_idx_end = MinValue<idx_t>(block_idx_start + blocks_per_thread, num_blocks);
				finalize_tasks.push_back(make_unique<HashJoinFinalizeTask>(shared_from_this(), context, sink,
				                                                           block_idx_start, block_idx_end, true));
				block_idx = block_idx_end;
				if (block_idx == num_blocks) {
					break;
				}
			}
		}
		SetTasks(move(finalize_tasks));
	}

template <bool PARALLEL>
static inline void InsertHashesLoop(atomic<data_ptr_t> pointers[], const hash_t indices[], const idx_t count,
                                    const data_ptr_t key_locations[], const idx_t pointer_offset) {
	for (idx_t i = 0; i < count; i++) {
		auto index = indices[i];
		if (PARALLEL) {
			data_ptr_t head;
			do {
				head = pointers[index];
				Store<data_ptr_t>(head, key_locations[i] + pointer_offset);
			} while (!std::atomic_compare_exchange_weak(&pointers[index], &head, key_locations[i]));
		} else {
			// set prev in current key to the value (NOTE: this will be nullptr if there is none)
			Store<data_ptr_t>(pointers[index], key_locations[i] + pointer_offset);

			// set pointer to current tuple
			pointers[index] = key_locations[i];
		}
	}
}
  • 並行掃描hash表，進行outer數據的處理：
void PhysicalHashJoin::GetData(ExecutionContext &context, DataChunk &chunk, GlobalSourceState &gstate_p,
                               LocalSourceState &lstate_p) const {
	auto &sink = (HashJoinGlobalSinkState &)*sink_state;
	auto &gstate = (HashJoinGlobalSourceState &)gstate_p;
	auto &lstate = (HashJoinLocalSourceState &)lstate_p;
	sink.scanned_data = true;

	if (!sink.external) {
		if (IsRightOuterJoin(join_type)) {
			{
				lock_guard<mutex> guard(gstate.lock);
                                // 拆解掃描部分hash表的數據
				lstate.ScanFullOuter(sink, gstate);
			}
                        // 掃描hash表讀取數據
			sink.hash_table->GatherFullOuter(chunk, lstate.addresses, lstate.full_outer_found_entries);
		}
		return;
	}
}


void HashJoinLocalSourceState::ScanFullOuter(HashJoinGlobalSinkState &sink, HashJoinGlobalSourceState &gstate) {
	auto &fo_ss = gstate.full_outer_scan;
	idx_t scan_index_before = fo_ss.scan_index;
	full_outer_found_entries = sink.hash_table->ScanFullOuter(fo_ss, addresses);
	idx_t scanned = fo_ss.scan_index - scan_index_before;
	full_outer_in_progress = scanned;
}

小結

  • DuckDB在多線程同步，核心就是在Combine的時候：加鎖，併發是通過原子變數的方式實現併發寫入hash表的操作
  • 通過local/global 拆分私有記憶體和公共記憶體，併發的基礎是在私有記憶體上進行運算，同步的部分主要在公有記憶體的更新

3. Spill To Disk的實現

DuckDB並沒有如筆者預期的實現非同步IO， 所以任意的執行線程是有可能Stall在系統的I/O調度上的，我想大概率是DuckDB本身的定位對於高併發場景的支持不是那麼敏感所導致的。這裡他們也作為了後續TODO的計劃之一。

image.png

4. 參考資料

DuckDB源碼

Push-Based Execution in DuckDB


您的分享是我們最大的動力!

-Advertisement-
Play Games
更多相關文章
  • 乾貨總結｜快速構造String對象及訪問其內部成員的技巧
    FASTJSON2項目使用了上面的技巧，其中JDKUtils和UnsafeUtils有上面技巧的實現： JDKUtils：https://github.com/alibaba/fastjson2/blob/fastcode_demo_20221218/core/src/main/java/com/... ...
  • ASP.NET6 + Mongo + OData
    準備工作 Docker環境 Mongo資料庫 配置Mongo資料庫 ASP.NET6 集成Mongo 安裝MongoDB.Driver { "Logging": { "LogLevel": { "Default": "Information", "Microsoft.AspNetCore": "Wa ...
  • ASP.NET Core - 入口文件
    最近遇到需求需要把控制台程式做成 windows服務 的形式運行。改成 windows服務 後，程式無法訪問共用網路文件，訪問提示“Access to the path '網路路徑' is denied.“，拒絕訪問，原因是當前服務的用戶憑證無許可權訪問，於是需要修改當前服務的用戶憑證，改為可訪問共用 ...
  • T4模版引擎
    簡單說下什麼是t4模版T4，即4個T開頭的英文字母組合：Text Template Transformation Toolkit。 T4(Text Template Transformation Toolkit)是微軟官方在VisualStudio 2008中開始使用的代碼生成引擎。在 Visual ...
  • 《痞子衡嵌入式半月刊》 第 71 期
    痞子衡嵌入式半月刊： 第 71 期 這裡分享嵌入式領域有用有趣的項目/工具以及一些熱點新聞，農曆年分二十四節氣，希望在每個交節之日準時發佈一期。 本期刊是開源項目（GitHub: JayHeng/pzh-mcu-bi-weekly），歡迎提交 issue，投稿或推薦你知道的嵌入式那些事兒。 上期回顧 ...
  • mysql 運維篇
    日誌 錯誤日誌 錯誤日誌是mysql中最重要的日誌之一，它記錄了當mysqld啟動和停止時，以及伺服器在運行過程中發生任何嚴重錯誤時的相關信息，當資料庫出現任何故障導致無法正常使用時，建議首先查看此日誌。 該日誌是預設開啟的，預設存放目錄：/var/log/,預設的日誌文件名為mysql.log。查 ...
  • 資料庫索引原理及優化
    摘要： 本文內容主要來源於互聯網上主流文章，只是按照個人理解稍作整合，後面附有參考鏈接。 本文內容主要來源於互聯網上主流文章，只是按照個人理解稍作整合，後面附有參考鏈接。 一、摘要 本文以MySQL資料庫為研究對象，討論與資料庫索引相關的一些話題。特別需要說明的是，MySQL支持諸多存儲引擎，而各種 ...
  • SSMS建表時，出現報錯，無法建表
    出現的提示信息 This backend version is not supported to design database diagrams or tables. (MS Visual Database Tools 問題發生的原因 SSMS的版本低於SQL Server的版本 ———————— ...
一周排行
    -Advertisement-
    Play Games
  • 移動開發（一）：使用.NET MAUI開發第一個安卓APP
    移動開發（一）：使用.NET MAUI開發第一個安卓APP 對於工作多年的C#程式員來說，近來想嘗試開發一款安卓APP，考慮了很久最終選擇使用.NET MAUI這個微軟官方的框架來嘗試體驗開發安卓APP，畢竟是使用Visual Studio開發工具，使用起來也比較的順手，結合微軟官方的教程進行了安卓 ...
  • wpf ToggleButton選中效果和一個登錄界面
    前言 QuestPDF 是一個開源 .NET 庫，用於生成 PDF 文檔。使用了C# Fluent API方式可簡化開發、減少錯誤並提高工作效率。利用它可以輕鬆生成 PDF 報告、發票、導出文件等。 項目介紹 QuestPDF 是一個革命性的開源 .NET 庫，它徹底改變了我們生成 PDF 文檔的方 ...
  • 彈幕樹洞項目功能新增篇
    項目地址 項目後端地址： https://github.com/ZyPLJ/ZYTteeHole 項目前端頁面地址： ZyPLJ/TreeHoleVue (github.com) https://github.com/ZyPLJ/TreeHoleVue 目前項目測試訪問地址： http://tree ...
  • 第27篇 sqlserver2022詳細安裝步驟
    話不多說，直接開乾 一.下載 1.官方鏈接下載： https://www.microsoft.com/zh-cn/sql-server/sql-server-downloads 2.在下載目錄中找到下麵這個小的安裝包 SQL2022-SSEI-Dev.exe，運行開始下載SQL server； 二. ...
  • .NET 開源高性能 MQTT 類庫
    前言 隨著物聯網（IoT）技術的迅猛發展，MQTT（消息隊列遙測傳輸）協議憑藉其輕量級和高效性，已成為眾多物聯網應用的首選通信標準。 MQTTnet 作為一個高性能的 .NET 開源庫，為 .NET 平臺上的 MQTT 客戶端與伺服器開發提供了強大的支持。 本文將全面介紹 MQTTnet 的核心功能 ...
  • Serilog文檔翻譯系列（六） - 可用的接收器、增強器、格式化輸出
    Serilog支持多種接收器用於日誌存儲，增強器用於添加屬性，LogContext管理動態屬性，支持多種輸出格式包括純文本、JSON及ExpressionTemplate。還提供了自定義格式化選項，適用於不同需求。 ...
  • 警惕 Visual Studio 屬性求值副作用導致邏輯不符合預期
    目錄簡介獲取 HTML 文檔解析 HTML 文檔測試參考文章 簡介 動態內容網站使用 JavaScript 腳本動態檢索和渲染數據，爬取信息時需要模擬瀏覽器行為，否則獲取到的源碼基本是空的。 本文使用的爬取步驟如下： 使用 Selenium 獲取渲染後的 HTML 文檔 使用 HtmlAgility ...
  • [使用目前最新版]HybridCLR6.9.0+YooAsset2.2.4實現純C# Unity熱更新方案 (一)
    1.前言 什麼是熱更新 游戲或者軟體更新時，無需重新下載客戶端進行安裝，而是在應用程式啟動的情況下，在內部進行資源或者代碼更新 Unity目前常用熱更新解決方案 HybridCLR，Xlua，ILRuntime等 Unity目前常用資源管理解決方案 AssetBundles，Addressable， ...
  • 在 ASP.NET Core Web API 中使用操作篩選器統一處理通用操作
    本文章主要是在C# ASP.NET Core Web API框架實現向手機發送驗證碼簡訊功能。這裡我選擇是一個互億無線簡訊驗證碼平臺，其實像阿裡雲，騰訊雲上面也可以。 首先我們先去 互億無線 https://www.ihuyi.com/api/sms.html 去註冊一個賬號 註冊完成賬號後，它會送 ...
  • 第28篇 如何.net中實現高效可靠數據同步api
    通過以下方式可以高效，並保證數據同步的可靠性 1.API設計 使用RESTful設計，確保API端點明確，並使用適當的HTTP方法（如POST用於創建，PUT用於更新）。 設計清晰的請求和響應模型，以確保客戶端能夠理解預期格式。 2.數據驗證 在伺服器端進行嚴格的數據驗證，確保接收到的數據符合預期格 ...
所有分類