dbms: added settings to fine tune reads from merge tree [#METR-2944].

dbms/include/DB/Interpreters/Settings.h

@@ -92,6 +92,18 @@ struct Settings
 	M(SettingUInt64, max_parallel_replicas, 1) \
 	M(SettingUInt64, parallel_replicas_count, 0) \
 	M(SettingUInt64, parallel_replica_offset, 0) \
+	\
+	/** Тонкие настройки для чтения из MergeTree */ \
+	\
+	/** Если из одного файла читается хотя бы столько строк, чтение можно распараллелить. */ \
+	M(SettingUInt64, merge_tree_min_rows_for_concurrent_read, (20 * 8192)) \
+	/** Можно пропускать чтение более чем стольки строк ценой одного seek по файлу. */ \
+	M(SettingUInt64, merge_tree_min_rows_for_seek, (5 * 8192)) \
+	/** Если отрезок индекса может содержать нужные ключи, делим его на столько частей и рекурсивно проверяем их. */ \
+	M(SettingUInt64, merge_tree_coarse_index_granularity, 8) \
+	/** Максимальное количество строк на запрос, для использования кэша разжатых данных. Если запрос большой - кэш не используется. \
+	  * (Чтобы большие запросы не вымывали кэш.) */ \
+	M(SettingUInt64, merge_tree_max_rows_to_use_cache, (1024 * 1024)) \
 
 	/// Всевозможные ограничения на выполнение запроса.
 	Limits limits;

dbms/include/DB/Storages/MergeTree/MergeTreeData.h

@@ -96,24 +96,9 @@ struct MergeTreeSettings
 	/// Сколько заданий на слияние кусков разрешено одновременно иметь в очереди ReplicatedMergeTree.
 	size_t max_replicated_merges_in_queue = 6;
 
-	/// Если из одного файла читается хотя бы столько строк, чтение можно распараллелить.
-	size_t min_rows_for_concurrent_read = 20 * 8192;
-
 	/// Через сколько секунд удалять ненужные куски.
 	time_t old_parts_lifetime = 8 * 60;
 
-	/** Настройки чтения и работы с индексом. */
-
-	/// Можно пропускать чтение более чем стольки строк ценой одного seek по файлу.
-	size_t min_rows_for_seek = 5 * 8192;
-
-	/// Если отрезок индекса может содержать нужные ключи, делим его на столько частей и рекурсивно проверяем их.
-	size_t coarse_index_granularity = 8;
-
-	/// Максимальное количество строк на запрос, для использования кэша разжатых данных. Если запрос большой - кэш не используется.
-	/// (Чтобы большие запросы не вымывали кэш.)
-	size_t max_rows_to_use_cache = 1024 * 1024;
-
 	/** Настройки вставок. */
 
 	/// Если в таблице хотя бы столько активных кусков, искусственно замедлять вставки в таблицу.

dbms/include/DB/Storages/MergeTree/MergeTreeDataSelectExecutor.h

@@ -48,10 +48,6 @@ private:
 
 	typedef std::vector<RangesInDataPart> RangesInDataParts;
 
-	size_t min_marks_for_seek;
-	size_t min_marks_for_concurrent_read;
-	size_t max_marks_to_use_cache;
-
 	BlockInputStreams spreadMarkRangesAmongThreads(
 		RangesInDataParts parts,
 		size_t threads,
@@ -60,7 +56,8 @@ private:
 		bool use_uncompressed_cache,
 		ExpressionActionsPtr prewhere_actions,
 		const String & prewhere_column,
-		const Names & virt_columns);
+		const Names & virt_columns,
+		const Settings & settings);
 
 	BlockInputStreams spreadMarkRangesAmongThreadsFinal(
 		RangesInDataParts parts,
@@ -70,12 +67,13 @@ private:
 		bool use_uncompressed_cache,
 		ExpressionActionsPtr prewhere_actions,
 		const String & prewhere_column,
-		const Names & virt_columns);
+		const Names & virt_columns,
+		const Settings & settings);
 
 	/// Создать выражение "Sign == 1".
 	void createPositiveSignCondition(ExpressionActionsPtr & out_expression, String & out_column);
 
-	MarkRanges markRangesFromPkRange(const MergeTreeData::DataPart::Index & index, PKCondition & key_condition);
+	MarkRanges markRangesFromPkRange(const MergeTreeData::DataPart::Index & index, PKCondition & key_condition, const Settings & settings);
 };
 
 }

dbms/src/Storages/MergeTree/MergeTreeDataSelectExecutor.cpp

@@ -14,13 +14,9 @@
 namespace DB
 {
 
-MergeTreeDataSelectExecutor::MergeTreeDataSelectExecutor(MergeTreeData & data_) : data(data_), log(&Logger::get(data.getLogName() + " (SelectExecutor)"))
+MergeTreeDataSelectExecutor::MergeTreeDataSelectExecutor(MergeTreeData & data_)
+	: data(data_), log(&Logger::get(data.getLogName() + " (SelectExecutor)"))
 {
-	min_marks_for_seek = (data.settings.min_rows_for_seek + data.index_granularity - 1) / data.index_granularity;
-	min_marks_for_concurrent_read = (data.settings.min_rows_for_concurrent_read + data.index_granularity - 1) / data.index_granularity;
-	max_marks_to_use_cache = (data.settings.max_rows_to_use_cache + data.index_granularity - 1) / data.index_granularity;
-
-
 }
 
 /// Построить блок состоящий только из возможных значений виртуальных столбцов
@@ -132,7 +128,7 @@ BlockInputStreams MergeTreeDataSelectExecutor::read(
 			for (size_t i = 0; i < parts.size(); ++i)
 			{
 				MergeTreeData::DataPartPtr & part = parts[i];
-				MarkRanges ranges = markRangesFromPkRange(part->index, key_condition);
+				MarkRanges ranges = markRangesFromPkRange(part->index, key_condition, settings);
 
 				for (size_t j = 0; j < ranges.size(); ++j)
 					total_count += ranges[j].end - ranges[j].begin;
@@ -264,7 +260,7 @@ BlockInputStreams MergeTreeDataSelectExecutor::read(
 	for (auto & part : parts)
 	{
 		RangesInDataPart ranges(part, (*part_index)++);
-		ranges.ranges = markRangesFromPkRange(part->index, key_condition);
+		ranges.ranges = markRangesFromPkRange(part->index, key_condition, settings);
 
 		if (!ranges.ranges.empty())
 		{
@@ -298,7 +294,8 @@ BlockInputStreams MergeTreeDataSelectExecutor::read(
 			settings.use_uncompressed_cache,
 			prewhere_actions,
 			prewhere_column,
-			virt_column_names);
+			virt_column_names,
+			settings);
 	}
 	else
 	{
@@ -310,7 +307,8 @@ BlockInputStreams MergeTreeDataSelectExecutor::read(
 			settings.use_uncompressed_cache,
 			prewhere_actions,
 			prewhere_column,
-			virt_column_names);
+			virt_column_names,
+			settings);
 	}
 
 	if (relative_sample_size != 0)
@@ -328,8 +326,12 @@ BlockInputStreams MergeTreeDataSelectExecutor::spreadMarkRangesAmongThreads(
 	bool use_uncompressed_cache,
 	ExpressionActionsPtr prewhere_actions,
 	const String & prewhere_column,
-	const Names & virt_columns)
+	const Names & virt_columns,
+	const Settings & settings)
 {
+	size_t min_marks_for_concurrent_read = (settings.merge_tree_min_rows_for_concurrent_read + data.index_granularity - 1) / data.index_granularity;
+	size_t max_marks_to_use_cache = (settings.merge_tree_max_rows_to_use_cache + data.index_granularity - 1) / data.index_granularity;
+
 	/// На всякий случай перемешаем куски.
 	std::random_shuffle(parts.begin(), parts.end());
 
@@ -451,8 +453,11 @@ BlockInputStreams MergeTreeDataSelectExecutor::spreadMarkRangesAmongThreadsFinal
 	bool use_uncompressed_cache,
 	ExpressionActionsPtr prewhere_actions,
 	const String & prewhere_column,
-	const Names & virt_columns)
+	const Names & virt_columns,
+	const Settings & settings)
 {
+	size_t max_marks_to_use_cache = (settings.merge_tree_max_rows_to_use_cache + data.index_granularity - 1) / data.index_granularity;
+
 	size_t sum_marks = 0;
 	for (size_t i = 0; i < parts.size(); ++i)
 		for (size_t j = 0; j < parts[i].ranges.size(); ++j)
@@ -529,8 +534,11 @@ void MergeTreeDataSelectExecutor::createPositiveSignCondition(ExpressionActionsP
 }
 
 /// Получает набор диапазонов засечек, вне которых не могут находиться ключи из заданного диапазона.
-MarkRanges MergeTreeDataSelectExecutor::markRangesFromPkRange(const MergeTreeData::DataPart::Index & index, PKCondition & key_condition)
+MarkRanges MergeTreeDataSelectExecutor::markRangesFromPkRange(
+	const MergeTreeData::DataPart::Index & index, PKCondition & key_condition, const Settings & settings)
 {
+	size_t min_marks_for_seek = (settings.merge_tree_min_rows_for_seek + data.index_granularity - 1) / data.index_granularity;
+
 	MarkRanges res;
 
 	size_t key_size = data.getSortDescription().size();
@@ -575,7 +583,7 @@ MarkRanges MergeTreeDataSelectExecutor::markRangesFromPkRange(const MergeTreeDat
 			else
 			{
 				/// Разбиваем отрезок и кладем результат в стек справа налево.
-				size_t step = (range.end - range.begin - 1) / data.settings.coarse_index_granularity + 1;
+				size_t step = (range.end - range.begin - 1) / settings.merge_tree_coarse_index_granularity + 1;
 				size_t end;
 
 				for (end = range.end; end > range.begin + step; end -= step)