MergeTreeDataSelectExecutor.cpp 24.5 KB
Newer Older
M
Merge  
Michael Kolupaev 已提交
1
#include <DB/Storages/MergeTree/MergeTreeDataSelectExecutor.h>
A
Merge  
Alexey Milovidov 已提交
2
#include <DB/Storages/MergeTree/MergeTreeBlockInputStream.h>
A
Merge  
Andrey Mironov 已提交
3 4
#include <DB/Storages/MergeTree/MergeTreeReadPool.h>
#include <DB/Storages/MergeTree/MergeTreeThreadBlockInputStream.h>
M
Merge  
Michael Kolupaev 已提交
5 6 7 8
#include <DB/Parsers/ASTIdentifier.h>
#include <DB/DataStreams/ExpressionBlockInputStream.h>
#include <DB/DataStreams/FilterBlockInputStream.h>
#include <DB/DataStreams/CollapsingFinalBlockInputStream.h>
9
#include <DB/DataStreams/AddingConstColumnBlockInputStream.h>
10 11
#include <DB/DataStreams/CreatingSetsBlockInputStream.h>
#include <DB/DataStreams/NullBlockInputStream.h>
M
Merge  
Michael Kolupaev 已提交
12
#include <DB/Common/VirtualColumnUtils.h>
13

14

M
Merge  
Michael Kolupaev 已提交
15 16 17
namespace DB
{

18 19
MergeTreeDataSelectExecutor::MergeTreeDataSelectExecutor(MergeTreeData & data_)
	: data(data_), log(&Logger::get(data.getLogName() + " (SelectExecutor)"))
M
Merge  
Michael Kolupaev 已提交
20 21 22
{
}

M
Merge  
Michael Kolupaev 已提交
23 24 25 26 27 28 29 30 31 32 33 34 35
/// Построить блок состоящий только из возможных значений виртуальных столбцов
static Block getBlockWithVirtualColumns(const MergeTreeData::DataPartsVector & parts)
{
	Block res;
	ColumnWithNameAndType _part(new ColumnString, new DataTypeString, "_part");

	for (const auto & part : parts)
		_part.column->insert(part->name);

	res.insert(_part);
	return res;
}

M
Merge  
Michael Kolupaev 已提交
36 37 38
BlockInputStreams MergeTreeDataSelectExecutor::read(
	const Names & column_names_to_return,
	ASTPtr query,
39
	const Context & context,
M
Merge  
Michael Kolupaev 已提交
40 41
	const Settings & settings,
	QueryProcessingStage::Enum & processed_stage,
42 43
	const size_t max_block_size,
	const unsigned threads,
M
Merge  
Michael Kolupaev 已提交
44
	size_t * part_index)
M
Merge  
Michael Kolupaev 已提交
45
{
M
Merge  
Michael Kolupaev 已提交
46 47 48 49
	size_t part_index_var = 0;
	if (!part_index)
		part_index = &part_index_var;

A
Merge  
Andrey Mironov 已提交
50
	MergeTreeData::DataPartsVector parts = data.getDataPartsVector();
M
Merge  
Michael Kolupaev 已提交
51 52 53

	/// Если в запросе есть ограничения на виртуальный столбец _part, выберем только подходящие под него куски.
	Names virt_column_names, real_column_names;
M
Merge  
Michael Kolupaev 已提交
54 55 56 57
	for (const String & name : column_names_to_return)
		if (name != "_part" &&
			name != "_part_index")
			real_column_names.push_back(name);
M
Merge  
Michael Kolupaev 已提交
58
		else
M
Merge  
Michael Kolupaev 已提交
59
			virt_column_names.push_back(name);
M
Merge  
Michael Kolupaev 已提交
60

M
Merge  
Michael Kolupaev 已提交
61
	/// Если в запросе только виртуальные столбцы, надо запросить хотя бы один любой другой.
62
	if (real_column_names.empty())
M
Merge  
Michael Kolupaev 已提交
63 64
		real_column_names.push_back(ExpressionActions::getSmallestColumn(data.getColumnsList()));

M
Merge  
Michael Kolupaev 已提交
65 66 67 68
	Block virtual_columns_block = getBlockWithVirtualColumns(parts);

	/// Если запрошен хотя бы один виртуальный столбец, пробуем индексировать
	if (!virt_column_names.empty())
69
		VirtualColumnUtils::filterBlockWithQuery(query, virtual_columns_block, context);
M
Merge  
Michael Kolupaev 已提交
70

71
	std::multiset<String> values = VirtualColumnUtils::extractSingleValueFromBlock<String>(virtual_columns_block, "_part");
M
Merge  
Michael Kolupaev 已提交
72 73

	data.check(real_column_names);
M
Merge  
Michael Kolupaev 已提交
74 75
	processed_stage = QueryProcessingStage::FetchColumns;

76 77
	PKCondition key_condition(query, context, data.getColumnsList(), data.getSortDescription());
	PKCondition date_condition(query, context, data.getColumnsList(), SortDescription(1, SortColumnDescription(data.date_column_name, 1)));
M
Merge  
Michael Kolupaev 已提交
78

79
	if (settings.force_index_by_date && date_condition.alwaysUnknown())
80 81
		throw Exception("Index by date is not used and setting 'force_index_by_date' is set.", ErrorCodes::INDEX_NOT_USED);

M
Merge  
Michael Kolupaev 已提交
82
	/// Выберем куски, в которых могут быть данные, удовлетворяющие date_condition, и которые подходят под условие на _part.
M
Merge  
Michael Kolupaev 已提交
83
	{
M
Merge  
Michael Kolupaev 已提交
84 85
		auto prev_parts = parts;
		parts.clear();
M
Merge  
Michael Kolupaev 已提交
86

M
Merge  
Michael Kolupaev 已提交
87
		for (const auto & part : prev_parts)
M
Merge  
Michael Kolupaev 已提交
88
		{
M
Merge  
Michael Kolupaev 已提交
89 90 91 92 93 94 95 96
			if (values.find(part->name) == values.end())
				continue;

			Field left = static_cast<UInt64>(part->left_date);
			Field right = static_cast<UInt64>(part->right_date);

			if (!date_condition.mayBeTrueInRange(&left, &right))
				continue;
M
Merge  
Michael Kolupaev 已提交
97

M
Merge  
Michael Kolupaev 已提交
98
			parts.push_back(part);
M
Merge  
Michael Kolupaev 已提交
99 100 101 102
		}
	}

	/// Семплирование.
M
Merge  
Michael Kolupaev 已提交
103
	Names column_names_to_read = real_column_names;
M
Merge  
Michael Kolupaev 已提交
104 105 106
	typedef Poco::SharedPtr<ASTFunction> ASTFunctionPtr;
	ASTFunctionPtr filter_function;
	ExpressionActionsPtr filter_expression;
107
	double relative_sample_size = 0;
M
Merge  
Michael Kolupaev 已提交
108

A
Merge  
Andrey Mironov 已提交
109 110
	ASTSelectQuery & select = *typeid_cast<ASTSelectQuery*>(&*query);

M
Merge  
Michael Kolupaev 已提交
111 112
	if (select.sample_size)
	{
113
		relative_sample_size = apply_visitor(FieldVisitorConvertToNumber<double>(),
114
			typeid_cast<ASTLiteral&>(*select.sample_size).value);
M
Merge  
Michael Kolupaev 已提交
115

116
		if (relative_sample_size < 0)
M
Merge  
Michael Kolupaev 已提交
117 118
			throw Exception("Negative sample size", ErrorCodes::ARGUMENT_OUT_OF_BOUND);

119 120
		/// Переводим абсолютную величину сэмплирования (вида SAMPLE 1000000 - сколько строк прочитать) в относительную (какую долю данных читать).
		if (relative_sample_size > 1)
M
Merge  
Michael Kolupaev 已提交
121
		{
122
			size_t requested_count = apply_visitor(FieldVisitorConvertToNumber<UInt64>(), typeid_cast<ASTLiteral&>(*select.sample_size).value);
M
Merge  
Michael Kolupaev 已提交
123 124 125 126 127 128 129

			/// Узнаем, сколько строк мы бы прочли без семплирования.
			LOG_DEBUG(log, "Preliminary index scan with condition: " << key_condition.toString());
			size_t total_count = 0;
			for (size_t i = 0; i < parts.size(); ++i)
			{
				MergeTreeData::DataPartPtr & part = parts[i];
130
				MarkRanges ranges = markRangesFromPkRange(part->index, key_condition, settings);
M
Merge  
Michael Kolupaev 已提交
131 132 133 134 135 136

				for (size_t j = 0; j < ranges.size(); ++j)
					total_count += ranges[j].end - ranges[j].begin;
			}
			total_count *= data.index_granularity;

137
			relative_sample_size = std::min(1., static_cast<double>(requested_count) / total_count);
M
Merge  
Michael Kolupaev 已提交
138

139
			LOG_DEBUG(log, "Selected relative sample size: " << relative_sample_size);
M
Merge  
Michael Kolupaev 已提交
140 141
		}

142 143 144 145 146
		/// SAMPLE 1 - то же, что и отсутствие SAMPLE.
		if (relative_sample_size == 1)
			relative_sample_size = 0;
	}

147
	if ((settings.parallel_replicas_count > 1) && !data.sampling_expression.isNull() && (relative_sample_size == 0))
148 149
		relative_sample_size = 1;

150 151
	if (relative_sample_size != 0)
	{
M
Merge  
Michael Kolupaev 已提交
152
		UInt64 sampling_column_max = 0;
M
Merge  
Michael Kolupaev 已提交
153
		DataTypePtr type = data.getPrimaryExpression()->getSampleBlock().getByName(data.sampling_expression->getColumnName()).type;
M
Merge  
Michael Kolupaev 已提交
154 155 156 157 158 159 160 161 162 163 164 165

		if (type->getName() == "UInt64")
			sampling_column_max = std::numeric_limits<UInt64>::max();
		else if (type->getName() == "UInt32")
			sampling_column_max = std::numeric_limits<UInt32>::max();
		else if (type->getName() == "UInt16")
			sampling_column_max = std::numeric_limits<UInt16>::max();
		else if (type->getName() == "UInt8")
			sampling_column_max = std::numeric_limits<UInt8>::max();
		else
			throw Exception("Invalid sampling column type in storage parameters: " + type->getName() + ". Must be unsigned integer type.", ErrorCodes::ILLEGAL_TYPE_OF_COLUMN_FOR_FILTER);

166 167
		UInt64 sampling_column_value_lower_limit;
		UInt64 sampling_column_value_upper_limit;
168
		UInt64 upper_limit = static_cast<UInt64>(static_cast<long double>(relative_sample_size) * sampling_column_max);
169

170
		if (settings.parallel_replicas_count > 1)
171
		{
172 173 174
			sampling_column_value_lower_limit = (settings.parallel_replica_offset * upper_limit) / settings.parallel_replicas_count;
			if ((settings.parallel_replica_offset + 1) < settings.parallel_replicas_count)
				sampling_column_value_upper_limit = ((settings.parallel_replica_offset + 1) * upper_limit) / settings.parallel_replicas_count;
175 176 177 178 179 180 181 182 183
			else
				sampling_column_value_upper_limit = upper_limit + 1;
		}
		else
		{
			sampling_column_value_lower_limit = 0;
			sampling_column_value_upper_limit = upper_limit + 1;
		}

M
Merge  
Michael Kolupaev 已提交
184
		/// Добавим условие, чтобы отсечь еще что-нибудь при повторном просмотре индекса.
185 186 187 188
		if (sampling_column_value_lower_limit > 0)
			if (!key_condition.addCondition(data.sampling_expression->getColumnName(),
				Range::createLeftBounded(sampling_column_value_lower_limit, true)))
				throw Exception("Sampling column not in primary key", ErrorCodes::ILLEGAL_COLUMN);
M
Merge  
Michael Kolupaev 已提交
189

190 191 192 193 194 195 196 197 198 199 200 201
		if (!key_condition.addCondition(data.sampling_expression->getColumnName(),
			Range::createRightBounded(sampling_column_value_upper_limit, false)))
			throw Exception("Sampling column not in primary key", ErrorCodes::ILLEGAL_COLUMN);

		ASTPtr upper_filter_args = new ASTExpressionList;
		upper_filter_args->children.push_back(data.sampling_expression);
		upper_filter_args->children.push_back(new ASTLiteral(StringRange(), sampling_column_value_upper_limit));

		ASTFunctionPtr upper_filter_function = new ASTFunction;
		upper_filter_function->name = "less";
		upper_filter_function->arguments = upper_filter_args;
		upper_filter_function->children.push_back(upper_filter_function->arguments);
M
Merge  
Michael Kolupaev 已提交
202

203 204 205 206 207 208 209
		if (sampling_column_value_lower_limit > 0)
		{
			/// Выражение для фильтрации: sampling_expression in [sampling_column_value_lower_limit, sampling_column_value_upper_limit)

			ASTPtr lower_filter_args = new ASTExpressionList;
			lower_filter_args->children.push_back(data.sampling_expression);
			lower_filter_args->children.push_back(new ASTLiteral(StringRange(), sampling_column_value_lower_limit));
M
Merge  
Michael Kolupaev 已提交
210

211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229
			ASTFunctionPtr lower_filter_function = new ASTFunction;
			lower_filter_function->name = "greaterOrEquals";
			lower_filter_function->arguments = lower_filter_args;
			lower_filter_function->children.push_back(lower_filter_function->arguments);

			ASTPtr filter_function_args = new ASTExpressionList;
			filter_function_args->children.push_back(lower_filter_function);
			filter_function_args->children.push_back(upper_filter_function);

			filter_function = new ASTFunction;
			filter_function->name = "and";
			filter_function->arguments = filter_function_args;
			filter_function->children.push_back(filter_function->arguments);
		}
		else
		{
			/// Выражение для фильтрации: sampling_expression < sampling_column_value_upper_limit
			filter_function = upper_filter_function;
		}
M
Merge  
Michael Kolupaev 已提交
230

231
		filter_expression = ExpressionAnalyzer(filter_function, data.context, data.getColumnsList()).getActions(false);
M
Merge  
Michael Kolupaev 已提交
232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247

		/// Добавим столбцы, нужные для sampling_expression.
		std::vector<String> add_columns = filter_expression->getRequiredColumns();
		column_names_to_read.insert(column_names_to_read.end(), add_columns.begin(), add_columns.end());
		std::sort(column_names_to_read.begin(), column_names_to_read.end());
		column_names_to_read.erase(std::unique(column_names_to_read.begin(), column_names_to_read.end()), column_names_to_read.end());
	}

	LOG_DEBUG(log, "Key condition: " << key_condition.toString());
	LOG_DEBUG(log, "Date condition: " << date_condition.toString());

	/// PREWHERE
	ExpressionActionsPtr prewhere_actions;
	String prewhere_column;
	if (select.prewhere_expression)
	{
248
		ExpressionAnalyzer analyzer(select.prewhere_expression, data.context, data.getColumnsList());
M
Merge  
Michael Kolupaev 已提交
249 250
		prewhere_actions = analyzer.getActions(false);
		prewhere_column = select.prewhere_expression->getColumnName();
251 252 253 254 255 256 257 258
		SubqueriesForSets prewhere_subqueries = analyzer.getSubqueriesForSets();

		/** Вычислим подзапросы прямо сейчас.
		  * NOTE Недостаток - эти вычисления не вписываются в конвейер выполнения запроса.
		  * Они делаются до начала выполнения конвейера; их нельзя прервать; во время вычислений не отправляются пакеты прогресса.
		  */
		if (!prewhere_subqueries.empty())
			CreatingSetsBlockInputStream(new NullBlockInputStream, prewhere_subqueries, settings.limits).read();
M
Merge  
Michael Kolupaev 已提交
259 260 261 262 263
	}

	RangesInDataParts parts_with_ranges;

	/// Найдем, какой диапазон читать из каждого куска.
264 265
	size_t sum_marks = 0;
	size_t sum_ranges = 0;
266
	for (auto & part : parts)
M
Merge  
Michael Kolupaev 已提交
267
	{
M
Merge  
Michael Kolupaev 已提交
268
		RangesInDataPart ranges(part, (*part_index)++);
A
Merge  
Alexey Milovidov 已提交
269 270 271 272 273

		if (data.mode != MergeTreeData::Unsorted)
			ranges.ranges = markRangesFromPkRange(part->index, key_condition, settings);
		else
			ranges.ranges = MarkRanges{MarkRange{0, part->size}};
M
Merge  
Michael Kolupaev 已提交
274 275

		if (!ranges.ranges.empty())
276
		{
277
			parts_with_ranges.push_back(ranges);
278

279 280 281
			sum_ranges += ranges.ranges.size();
			for (const auto & range : ranges.ranges)
				sum_marks += range.end - range.begin;
282
		}
283 284
	}

285
	LOG_DEBUG(log, "Selected " << parts.size() << " parts by date, " << parts_with_ranges.size() << " parts by key, "
286
		<< sum_marks << " marks to read from " << sum_ranges << " ranges");
M
Merge  
Michael Kolupaev 已提交
287 288 289 290 291 292

	BlockInputStreams res;

	if (select.final)
	{
		/// Добавим столбцы, нужные для вычисления первичного ключа и знака.
M
Merge  
Michael Kolupaev 已提交
293
		std::vector<String> add_columns = data.getPrimaryExpression()->getRequiredColumns();
M
Merge  
Michael Kolupaev 已提交
294 295 296 297 298 299
		column_names_to_read.insert(column_names_to_read.end(), add_columns.begin(), add_columns.end());
		column_names_to_read.push_back(data.sign_column);
		std::sort(column_names_to_read.begin(), column_names_to_read.end());
		column_names_to_read.erase(std::unique(column_names_to_read.begin(), column_names_to_read.end()), column_names_to_read.end());

		res = spreadMarkRangesAmongThreadsFinal(
300
			parts_with_ranges,
M
Merge  
Michael Kolupaev 已提交
301 302 303 304 305
			threads,
			column_names_to_read,
			max_block_size,
			settings.use_uncompressed_cache,
			prewhere_actions,
M
Merge  
Michael Kolupaev 已提交
306
			prewhere_column,
307 308
			virt_column_names,
			settings);
M
Merge  
Michael Kolupaev 已提交
309 310 311 312
	}
	else
	{
		res = spreadMarkRangesAmongThreads(
313
			parts_with_ranges,
M
Merge  
Michael Kolupaev 已提交
314 315 316 317 318
			threads,
			column_names_to_read,
			max_block_size,
			settings.use_uncompressed_cache,
			prewhere_actions,
M
Merge  
Michael Kolupaev 已提交
319
			prewhere_column,
320 321
			virt_column_names,
			settings);
M
Merge  
Michael Kolupaev 已提交
322 323
	}

324
	if (relative_sample_size != 0)
325 326
		for (auto & stream : res)
			stream = new FilterBlockInputStream(new ExpressionBlockInputStream(stream, filter_expression), filter_function->getColumnName());
M
Merge  
Michael Kolupaev 已提交
327 328 329 330

	return res;
}

331

M
Merge  
Michael Kolupaev 已提交
332 333 334 335 336 337 338
BlockInputStreams MergeTreeDataSelectExecutor::spreadMarkRangesAmongThreads(
	RangesInDataParts parts,
	size_t threads,
	const Names & column_names,
	size_t max_block_size,
	bool use_uncompressed_cache,
	ExpressionActionsPtr prewhere_actions,
M
Merge  
Michael Kolupaev 已提交
339
	const String & prewhere_column,
340 341
	const Names & virt_columns,
	const Settings & settings)
M
Merge  
Michael Kolupaev 已提交
342
{
A
Merge  
Andrey Mironov 已提交
343
	const std::size_t min_marks_for_concurrent_read =
A
Merge  
Andrey Mironov 已提交
344
		(settings.merge_tree_min_rows_for_concurrent_read + data.index_granularity - 1) / data.index_granularity;
A
Merge  
Andrey Mironov 已提交
345
	const std::size_t max_marks_to_use_cache =
A
Merge  
Andrey Mironov 已提交
346
		(settings.merge_tree_max_rows_to_use_cache + data.index_granularity - 1) / data.index_granularity;
347

M
Merge  
Michael Kolupaev 已提交
348 349 350 351 352 353 354 355 356 357 358
	/// На всякий случай перемешаем куски.
	std::random_shuffle(parts.begin(), parts.end());

	/// Посчитаем засечки для каждого куска.
	std::vector<size_t> sum_marks_in_parts(parts.size());
	size_t sum_marks = 0;
	for (size_t i = 0; i < parts.size(); ++i)
	{
		/// Пусть отрезки будут перечислены справа налево, чтобы можно было выбрасывать самый левый отрезок с помощью pop_back().
		std::reverse(parts[i].ranges.begin(), parts[i].ranges.end());

A
Merge  
Andrey Mironov 已提交
359
		for (const auto & range : parts[i].ranges)
M
Merge  
Michael Kolupaev 已提交
360
			sum_marks_in_parts[i] += range.end - range.begin;
A
Merge  
Andrey Mironov 已提交
361

M
Merge  
Michael Kolupaev 已提交
362 363 364 365 366 367
		sum_marks += sum_marks_in_parts[i];
	}

	if (sum_marks > max_marks_to_use_cache)
		use_uncompressed_cache = false;

A
Merge  
Andrey Mironov 已提交
368
	MergeTreeReadPoolPtr pool = std::make_shared<MergeTreeReadPool>(
A
Merge  
Andrey Mironov 已提交
369
		parts, data, prewhere_actions, prewhere_column, true, column_names);
A
Merge  
Andrey Mironov 已提交
370

M
Merge  
Michael Kolupaev 已提交
371 372
	BlockInputStreams res;

373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456
	/// @todo remove old code
	if (sum_marks > 0 && settings.merge_tree_uniform_read_distribution == 0)
	{
		const size_t min_marks_per_thread = (sum_marks - 1) / threads + 1;

		for (size_t i = 0; i < threads && !parts.empty(); ++i)
		{
			size_t need_marks = min_marks_per_thread;

			/// Цикл по кускам.
			while (need_marks > 0 && !parts.empty())
			{
				RangesInDataPart & part = parts.back();
				size_t & marks_in_part = sum_marks_in_parts.back();

				/// Не будем брать из куска слишком мало строк.
				if (marks_in_part >= min_marks_for_concurrent_read &&
					need_marks < min_marks_for_concurrent_read)
					need_marks = min_marks_for_concurrent_read;

				/// Не будем оставлять в куске слишком мало строк.
				if (marks_in_part > need_marks &&
					marks_in_part - need_marks < min_marks_for_concurrent_read)
					need_marks = marks_in_part;

				MarkRanges ranges_to_get_from_part;

				/// Возьмем весь кусок, если он достаточно мал.
				if (marks_in_part <= need_marks)
				{
					/// Восстановим порядок отрезков.
					std::reverse(part.ranges.begin(), part.ranges.end());

					ranges_to_get_from_part = part.ranges;

					need_marks -= marks_in_part;
					parts.pop_back();
					sum_marks_in_parts.pop_back();
				}
				else
				{
					/// Цикл по отрезкам куска.
					while (need_marks > 0)
					{
						if (part.ranges.empty())
							throw Exception("Unexpected end of ranges while spreading marks among threads", ErrorCodes::LOGICAL_ERROR);

						MarkRange & range = part.ranges.back();

						const size_t marks_in_range = range.end - range.begin;
						const size_t marks_to_get_from_range = std::min(marks_in_range, need_marks);

						ranges_to_get_from_part.emplace_back(range.begin, range.begin + marks_to_get_from_range);
						range.begin += marks_to_get_from_range;
						marks_in_part -= marks_to_get_from_range;
						need_marks -= marks_to_get_from_range;
						if (range.begin == range.end)
							part.ranges.pop_back();
					}
				}

				BlockInputStreamPtr source_stream = new MergeTreeBlockInputStream(
					data.getFullPath() + part.data_part->name + '/', max_block_size, column_names, data,
					part.data_part, ranges_to_get_from_part, use_uncompressed_cache,
					prewhere_actions, prewhere_column, true, settings.min_bytes_to_use_direct_io, settings.max_read_buffer_size);

				res.push_back(source_stream);

				for (const String & virt_column : virt_columns)
				{
					if (virt_column == "_part")
						res.back() = new AddingConstColumnBlockInputStream<String>(
							res.back(), new DataTypeString, part.data_part->name, "_part");
					else if (virt_column == "_part_index")
						res.back() = new AddingConstColumnBlockInputStream<UInt64>(
							res.back(), new DataTypeUInt64, part.part_index_in_query, "_part_index");
				}
			}
		}

		if (!parts.empty())
			throw Exception("Couldn't spread marks among threads", ErrorCodes::LOGICAL_ERROR);
	}
	else if (sum_marks > 0)
M
Merge  
Michael Kolupaev 已提交
457
	{
A
Merge  
Andrey Mironov 已提交
458 459 460 461 462
		for (std::size_t i = 0; i < threads; ++i)
			res.emplace_back(new MergeTreeThreadBlockInputStream{
				pool, min_marks_for_concurrent_read, max_block_size, data, use_uncompressed_cache, prewhere_actions,
				prewhere_column, settings.min_bytes_to_use_direct_io, settings.max_read_buffer_size, virt_columns
			});
M
Merge  
Michael Kolupaev 已提交
463

A
Merge  
Andrey Mironov 已提交
464 465
		/// Оценим общее количество строк - для прогресс-бара.
		const std::size_t total_rows = data.index_granularity * sum_marks;
M
Merge  
Michael Kolupaev 已提交
466

A
Merge  
Andrey Mironov 已提交
467 468
		/// Выставим приблизительное количество строк только для первого источника
		static_cast<IProfilingBlockInputStream &>(*res.front()).setTotalRowsApprox(total_rows);
M
Merge  
Michael Kolupaev 已提交
469

A
Merge  
Andrey Mironov 已提交
470
		LOG_TRACE(log, "Reading approx. " << total_rows);
M
Merge  
Michael Kolupaev 已提交
471 472 473 474 475 476 477 478 479 480 481 482
	}

	return res;
}

BlockInputStreams MergeTreeDataSelectExecutor::spreadMarkRangesAmongThreadsFinal(
	RangesInDataParts parts,
	size_t threads,
	const Names & column_names,
	size_t max_block_size,
	bool use_uncompressed_cache,
	ExpressionActionsPtr prewhere_actions,
M
Merge  
Michael Kolupaev 已提交
483
	const String & prewhere_column,
484 485
	const Names & virt_columns,
	const Settings & settings)
M
Merge  
Michael Kolupaev 已提交
486
{
487 488
	size_t max_marks_to_use_cache = (settings.merge_tree_max_rows_to_use_cache + data.index_granularity - 1) / data.index_granularity;

M
Merge  
Michael Kolupaev 已提交
489 490 491 492 493 494 495 496 497 498
	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)
			sum_marks += parts[i].ranges[j].end - parts[i].ranges[j].begin;

	if (sum_marks > max_marks_to_use_cache)
		use_uncompressed_cache = false;

	ExpressionActionsPtr sign_filter_expression;
	String sign_filter_column;
499
	createPositiveSignCondition(sign_filter_expression, sign_filter_column);
M
Merge  
Michael Kolupaev 已提交
500 501 502 503 504 505 506 507

	BlockInputStreams to_collapse;

	for (size_t part_index = 0; part_index < parts.size(); ++part_index)
	{
		RangesInDataPart & part = parts[part_index];

		BlockInputStreamPtr source_stream = new MergeTreeBlockInputStream(
508 509
			data.getFullPath() + part.data_part->name + '/', max_block_size, column_names, data,
			part.data_part, part.ranges, use_uncompressed_cache,
A
Merge  
Alexey Milovidov 已提交
510
			prewhere_actions, prewhere_column, true, settings.min_bytes_to_use_direct_io, settings.max_read_buffer_size);
A
Merge  
Alexey Arno 已提交
511

M
Merge  
Michael Kolupaev 已提交
512 513 514 515 516 517 518 519 520
		for (const String & virt_column : virt_columns)
		{
			if (virt_column == "_part")
				source_stream = new AddingConstColumnBlockInputStream<String>(
					source_stream, new DataTypeString, part.data_part->name, "_part");
			else if (virt_column == "_part_index")
				source_stream = new AddingConstColumnBlockInputStream<UInt64>(
					source_stream, new DataTypeUInt64, part.part_index_in_query, "_part_index");
		}
M
Merge  
Michael Kolupaev 已提交
521

M
Merge  
Michael Kolupaev 已提交
522
		to_collapse.push_back(new ExpressionBlockInputStream(source_stream, data.getPrimaryExpression()));
M
Merge  
Michael Kolupaev 已提交
523 524 525 526 527 528
	}

	BlockInputStreams res;
	if (to_collapse.size() == 1)
		res.push_back(new FilterBlockInputStream(new ExpressionBlockInputStream(to_collapse[0], sign_filter_expression), sign_filter_column));
	else if (to_collapse.size() > 1)
M
Merge  
Michael Kolupaev 已提交
529
		res.push_back(new CollapsingFinalBlockInputStream(to_collapse, data.getSortDescription(), data.sign_column));
M
Merge  
Michael Kolupaev 已提交
530 531 532 533

	return res;
}

534
void MergeTreeDataSelectExecutor::createPositiveSignCondition(ExpressionActionsPtr & out_expression, String & out_column)
M
Merge  
Michael Kolupaev 已提交
535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558 559 560
{
	ASTFunction * function = new ASTFunction;
	ASTPtr function_ptr = function;

	ASTExpressionList * arguments = new ASTExpressionList;
	ASTPtr arguments_ptr = arguments;

	ASTIdentifier * sign = new ASTIdentifier;
	ASTPtr sign_ptr = sign;

	ASTLiteral * one = new ASTLiteral;
	ASTPtr one_ptr = one;

	function->name = "equals";
	function->arguments = arguments_ptr;
	function->children.push_back(arguments_ptr);

	arguments->children.push_back(sign_ptr);
	arguments->children.push_back(one_ptr);

	sign->name = data.sign_column;
	sign->kind = ASTIdentifier::Column;

	one->type = new DataTypeInt8;
	one->value = Field(static_cast<Int64>(1));

561
	out_expression = ExpressionAnalyzer(function_ptr, data.context, data.getColumnsList()).getActions(false);
M
Merge  
Michael Kolupaev 已提交
562 563 564 565
	out_column = function->getColumnName();
}

/// Получает набор диапазонов засечек, вне которых не могут находиться ключи из заданного диапазона.
566 567
MarkRanges MergeTreeDataSelectExecutor::markRangesFromPkRange(
	const MergeTreeData::DataPart::Index & index, PKCondition & key_condition, const Settings & settings)
M
Merge  
Michael Kolupaev 已提交
568
{
569 570
	size_t min_marks_for_seek = (settings.merge_tree_min_rows_for_seek + data.index_granularity - 1) / data.index_granularity;

M
Merge  
Michael Kolupaev 已提交
571 572
	MarkRanges res;

M
Merge  
Michael Kolupaev 已提交
573
	size_t key_size = data.getSortDescription().size();
M
Merge  
Michael Kolupaev 已提交
574 575 576
	size_t marks_count = index.size() / key_size;

	/// Если индекс не используется.
577
	if (key_condition.alwaysUnknown())
M
Merge  
Michael Kolupaev 已提交
578 579 580 581 582 583 584 585 586 587 588 589 590 591 592 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607 608 609 610 611 612 613 614
	{
		res.push_back(MarkRange(0, marks_count));
	}
	else
	{
		/** В стеке всегда будут находиться непересекающиеся подозрительные отрезки, самый левый наверху (back).
			* На каждом шаге берем левый отрезок и проверяем, подходит ли он.
			* Если подходит, разбиваем его на более мелкие и кладем их в стек. Если нет - выбрасываем его.
			* Если отрезок уже длиной в одну засечку, добавляем его в ответ и выбрасываем.
			*/
		std::vector<MarkRange> ranges_stack;
		ranges_stack.push_back(MarkRange(0, marks_count));
		while (!ranges_stack.empty())
		{
			MarkRange range = ranges_stack.back();
			ranges_stack.pop_back();

			bool may_be_true;
			if (range.end == marks_count)
				may_be_true = key_condition.mayBeTrueAfter(&index[range.begin * key_size]);
			else
				may_be_true = key_condition.mayBeTrueInRange(&index[range.begin * key_size], &index[range.end * key_size]);

			if (!may_be_true)
				continue;

			if (range.end == range.begin + 1)
			{
				/// Увидели полезный промежуток между соседними засечками. Либо добавим его к последнему диапазону, либо начнем новый диапазон.
				if (res.empty() || range.begin - res.back().end > min_marks_for_seek)
					res.push_back(range);
				else
					res.back().end = range.end;
			}
			else
			{
				/// Разбиваем отрезок и кладем результат в стек справа налево.
615
				size_t step = (range.end - range.begin - 1) / settings.merge_tree_coarse_index_granularity + 1;
M
Merge  
Michael Kolupaev 已提交
616 617 618 619 620 621 622 623 624 625 626 627 628 629
				size_t end;

				for (end = range.end; end > range.begin + step; end -= step)
					ranges_stack.push_back(MarkRange(end - step, end));

				ranges_stack.push_back(MarkRange(range.begin, end));
			}
		}
	}

	return res;
}

}