ClickHouse/dbms/include/DB/Storages/MergeTree/MergeTreeReadPool.h

423 lines
16 KiB
C
Raw Normal View History

2015-06-24 11:03:53 +00:00
#pragma once
#include <DB/Core/NamesAndTypes.h>
#include <DB/Storages/MergeTree/RangesInDataPart.h>
2015-10-05 00:33:43 +00:00
#include <ext/range.hpp>
2015-06-24 11:03:53 +00:00
#include <mutex>
namespace DB
{
/// A batch of work for MergeTreeThreadBlockInputStream
2015-06-24 11:03:53 +00:00
struct MergeTreeReadTask
{
/// data part which should be read while performing this task
2015-06-24 11:03:53 +00:00
MergeTreeData::DataPartPtr data_part;
2015-09-16 15:26:20 +00:00
/** Ranges to read from `data_part`.
* Specified in reverse order for MergeTreeThreadBlockInputStream's convenience of calling .pop_back(). */
2015-06-24 11:03:53 +00:00
MarkRanges mark_ranges;
/// for virtual `part_index` virtual column
2015-06-24 11:03:53 +00:00
std::size_t part_index_in_query;
/// ordered list of column names used in this query, allows returning blocks with consistent ordering
2015-06-24 11:03:53 +00:00
const Names & ordered_names;
/// used to determine whether column should be filtered during PREWHERE or WHERE
2015-06-24 11:03:53 +00:00
const NameSet & column_name_set;
/// column names to read during WHERE
2015-06-24 11:03:53 +00:00
const NamesAndTypesList & columns;
/// column names to read during PREWHERE
2015-06-24 11:03:53 +00:00
const NamesAndTypesList & pre_columns;
/// should PREWHERE column be returned to requesting side?
2015-06-24 11:03:53 +00:00
const bool remove_prewhere_column;
/// resulting block may require reordering in accordance with `ordered_names`
2015-07-03 15:08:21 +00:00
const bool should_reorder;
2015-06-24 11:03:53 +00:00
2015-07-03 15:08:21 +00:00
MergeTreeReadTask(
const MergeTreeData::DataPartPtr & data_part, const MarkRanges & ranges, const std::size_t part_index_in_query,
const Names & ordered_names, const NameSet & column_name_set, const NamesAndTypesList & columns,
const NamesAndTypesList & pre_columns, const bool remove_prewhere_column, const bool should_reorder)
2015-06-24 11:03:53 +00:00
: data_part{data_part}, mark_ranges{ranges}, part_index_in_query{part_index_in_query},
ordered_names{ordered_names}, column_name_set{column_name_set}, columns{columns}, pre_columns{pre_columns},
2015-07-03 15:08:21 +00:00
remove_prewhere_column{remove_prewhere_column}, should_reorder{should_reorder}
2015-06-24 11:03:53 +00:00
{}
};
using MergeTreeReadTaskPtr = std::unique_ptr<MergeTreeReadTask>;
/** Provides read tasks for MergeTreeThreadBlockInputStream`s in fine-grained batches, allowing for more
* uniform distribution of work amongst multiple threads. All parts and their ranges are divided into `threads`
* workloads with at most `sum_marks / threads` marks. Then, threads are performing reads from these workloads
* in "sequential" manner, requesting work in small batches. As soon as some thread some thread has exhausted
* it's workload, it either is signaled that no more work is available (`do_not_steal_tasks == false`) or
* continues taking small batches from other threads' workloads (`do_not_steal_tasks == true`).
*/
2015-06-24 11:03:53 +00:00
class MergeTreeReadPool
{
public:
2015-07-03 15:08:21 +00:00
MergeTreeReadPool(
2015-07-23 17:54:07 +00:00
const std::size_t threads, const std::size_t sum_marks, const std::size_t min_marks_for_concurrent_read,
RangesInDataParts parts, MergeTreeData & data, const ExpressionActionsPtr & prewhere_actions,
const String & prewhere_column_name, const bool check_columns, const Names & column_names,
const bool do_not_steal_tasks = false)
: data{data}, column_names{column_names}, do_not_steal_tasks{do_not_steal_tasks}
2015-06-24 11:03:53 +00:00
{
2015-07-23 17:54:07 +00:00
const auto per_part_sum_marks = fillPerPartInfo(parts, prewhere_actions, prewhere_column_name, check_columns);
fillPerThreadInfo(threads, sum_marks, per_part_sum_marks, parts, min_marks_for_concurrent_read);
2015-06-24 11:03:53 +00:00
}
MergeTreeReadPool(const MergeTreeReadPool &) = delete;
MergeTreeReadPool & operator=(const MergeTreeReadPool &) = delete;
2015-07-23 13:11:27 +00:00
MergeTreeReadTaskPtr getTask(const std::size_t min_marks_to_read, const std::size_t thread)
2015-06-24 11:03:53 +00:00
{
const std::lock_guard<std::mutex> lock{mutex};
2015-07-23 17:54:07 +00:00
if (remaining_thread_tasks.empty())
2015-06-24 11:03:53 +00:00
return nullptr;
const auto tasks_remaining_for_this_thread = !threads_tasks[thread].sum_marks_in_parts.empty();
if (!tasks_remaining_for_this_thread && do_not_steal_tasks)
return nullptr;
const auto thread_idx = tasks_remaining_for_this_thread ? thread : *std::begin(remaining_thread_tasks);
2015-07-23 17:54:07 +00:00
auto & thread_tasks = threads_tasks[thread_idx];
2015-06-24 11:03:53 +00:00
2015-07-23 17:54:07 +00:00
auto & thread_task = thread_tasks.parts_and_ranges.back();
const auto part_idx = thread_task.part_idx;
auto & part = parts[part_idx];
auto & marks_in_part = thread_tasks.sum_marks_in_parts.back();
2015-06-24 11:03:53 +00:00
/// Берём весь кусок, если он достаточно мал
auto need_marks = std::min(marks_in_part, min_marks_to_read);
/// Не будем оставлять в куске слишком мало строк.
if (marks_in_part > need_marks &&
marks_in_part - need_marks < min_marks_to_read)
need_marks = marks_in_part;
MarkRanges ranges_to_get_from_part;
/// Возьмем весь кусок, если он достаточно мал.
if (marks_in_part <= need_marks)
{
const auto marks_to_get_from_range = marks_in_part;
2015-09-16 15:26:20 +00:00
/** Отрезки уже перечислены справа налево, reverse изначально сделан в MergeTreeDataSelectExecutor и
* поддержан в fillPerThreadInfo. */
2015-07-23 17:54:07 +00:00
ranges_to_get_from_part = thread_task.ranges;
2015-06-24 11:03:53 +00:00
marks_in_part -= marks_to_get_from_range;
2015-07-03 15:08:21 +00:00
2015-07-23 17:54:07 +00:00
thread_tasks.parts_and_ranges.pop_back();
thread_tasks.sum_marks_in_parts.pop_back();
if (thread_tasks.sum_marks_in_parts.empty())
remaining_thread_tasks.erase(thread_idx);
2015-06-24 11:03:53 +00:00
}
else
{
/// Цикл по отрезкам куска.
2015-07-23 17:54:07 +00:00
while (need_marks > 0 && !thread_task.ranges.empty())
2015-06-24 11:03:53 +00:00
{
2015-07-23 17:54:07 +00:00
auto & range = thread_task.ranges.back();
2015-06-24 11:03:53 +00:00
const std::size_t marks_in_range = range.end - range.begin;
const std::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;
if (range.begin == range.end)
2015-07-23 13:11:27 +00:00
{
2015-07-23 17:54:07 +00:00
std::swap(range, thread_task.ranges.back());
thread_task.ranges.pop_back();
2015-07-23 13:11:27 +00:00
}
2015-06-24 11:03:53 +00:00
marks_in_part -= marks_to_get_from_range;
need_marks -= marks_to_get_from_range;
}
2015-09-16 15:26:20 +00:00
/** Перечислим справа налево, чтобы MergeTreeThreadBlockInputStream забирал
* отрезки с помощью .pop_back() (их порядок был сменен на "слева направо"
* из-за .pop_back() в этой ветке). */
std::reverse(std::begin(ranges_to_get_from_part), std::end(ranges_to_get_from_part));
2015-06-24 11:03:53 +00:00
}
return std::make_unique<MergeTreeReadTask>(
2015-08-25 09:31:55 +00:00
part.data_part, ranges_to_get_from_part, part.part_index_in_query, column_names,
2015-07-23 17:54:07 +00:00
per_part_column_name_set[part_idx], per_part_columns[part_idx], per_part_pre_columns[part_idx],
per_part_remove_prewhere_column[part_idx], per_part_should_reorder[part_idx]);
2015-06-24 11:03:53 +00:00
}
private:
2015-07-23 17:54:07 +00:00
std::vector<std::size_t> fillPerPartInfo(
RangesInDataParts & parts, const ExpressionActionsPtr & prewhere_actions, const String & prewhere_column_name,
const bool check_columns)
2015-06-24 11:03:53 +00:00
{
2015-07-23 17:54:07 +00:00
std::vector<std::size_t> per_part_sum_marks;
2015-07-03 15:08:21 +00:00
for (const auto i : ext::range(0, parts.size()))
2015-06-24 11:03:53 +00:00
{
2015-07-03 15:08:21 +00:00
auto & part = parts[i];
/// Посчитаем засечки для каждого куска.
size_t sum_marks = 0;
2015-09-03 09:17:25 +00:00
/// Отрезки уже перечислены справа налево, reverse в MergeTreeDataSelectExecutor.
2015-07-03 15:08:21 +00:00
for (const auto & range : part.ranges)
sum_marks += range.end - range.begin;
per_part_sum_marks.push_back(sum_marks);
2015-06-24 11:03:53 +00:00
per_part_columns_lock.push_back(std::make_unique<Poco::ScopedReadRWLock>(
part.data_part->columns_lock));
/// inject column names required for DEFAULT evaluation in current part
auto required_column_names = column_names;
const auto injected_columns = injectRequiredColumns(part.data_part, required_column_names);
2015-07-03 15:08:21 +00:00
auto should_reoder = !injected_columns.empty();
2015-06-24 11:03:53 +00:00
Names required_pre_column_names;
if (prewhere_actions)
{
/// collect columns required for PREWHERE evaluation
required_pre_column_names = prewhere_actions->getRequiredColumns();
/// there must be at least one column required for PREWHERE
if (required_pre_column_names.empty())
required_pre_column_names.push_back(required_column_names[0]);
/// PREWHERE columns may require some additional columns for DEFAULT evaluation
2015-07-03 15:08:21 +00:00
const auto injected_pre_columns = injectRequiredColumns(part.data_part, required_pre_column_names);
if (!injected_pre_columns.empty())
should_reoder = true;
2015-06-24 11:03:53 +00:00
/// will be used to distinguish between PREWHERE and WHERE columns when applying filter
const NameSet pre_name_set{
std::begin(required_pre_column_names), std::end(required_pre_column_names)
};
/** Если выражение в PREWHERE - не столбец таблицы, не нужно отдавать наружу столбец с ним
* (от storage ожидают получить только столбцы таблицы). */
per_part_remove_prewhere_column.push_back(0 == pre_name_set.count(prewhere_column_name));
Names post_column_names;
for (const auto & name : required_column_names)
if (!pre_name_set.count(name))
post_column_names.push_back(name);
required_column_names = post_column_names;
}
else
per_part_remove_prewhere_column.push_back(false);
per_part_column_name_set.emplace_back(std::begin(required_column_names), std::end(required_column_names));
if (check_columns)
{
/** Под part->columns_lock проверим, что все запрошенные столбцы в куске того же типа, что в таблице.
* Это может быть не так во время ALTER MODIFY. */
if (!required_pre_column_names.empty())
data.check(part.data_part->columns, required_pre_column_names);
if (!required_column_names.empty())
data.check(part.data_part->columns, required_column_names);
per_part_pre_columns.push_back(data.getColumnsList().addTypes(required_pre_column_names));
per_part_columns.push_back(data.getColumnsList().addTypes(required_column_names));
}
else
{
per_part_pre_columns.push_back(part.data_part->columns.addTypes(required_pre_column_names));
per_part_columns.push_back(part.data_part->columns.addTypes(required_column_names));
}
2015-07-03 15:08:21 +00:00
per_part_should_reorder.push_back(should_reoder);
2015-07-23 17:54:07 +00:00
this->parts.push_back({ part.data_part, part.part_index_in_query });
}
return per_part_sum_marks;
}
void fillPerThreadInfo(
const std::size_t threads, const std::size_t sum_marks, std::vector<std::size_t> per_part_sum_marks,
RangesInDataParts & parts, const std::size_t min_marks_for_concurrent_read)
{
threads_tasks.resize(threads);
const size_t min_marks_per_thread = (sum_marks - 1) / threads + 1;
for (std::size_t i = 0; i < threads && !parts.empty(); ++i)
{
auto need_marks = min_marks_per_thread;
while (need_marks > 0 && !parts.empty())
{
const auto part_idx = parts.size() - 1;
RangesInDataPart & part = parts.back();
size_t & marks_in_part = per_part_sum_marks.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;
size_t marks_in_ranges = need_marks;
/// Возьмем весь кусок, если он достаточно мал.
if (marks_in_part <= need_marks)
{
2015-09-16 15:26:20 +00:00
/// Оставим отрезки перечисленными справа налево для удобства использования .pop_back() в .getTask()
2015-07-23 17:54:07 +00:00
ranges_to_get_from_part = part.ranges;
marks_in_ranges = marks_in_part;
need_marks -= marks_in_part;
parts.pop_back();
per_part_sum_marks.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();
}
2015-09-03 09:17:25 +00:00
2015-09-16 15:26:20 +00:00
/** Вновь перечислим отрезки справа налево, чтобы .getTask() мог забирать их
* с помощью .pop_back() (их порядок был сменен на "слева направо"
* из-за .pop_back() в этой ветке). */
2015-09-03 09:17:25 +00:00
std::reverse(std::begin(ranges_to_get_from_part), std::end(ranges_to_get_from_part));
2015-07-23 17:54:07 +00:00
}
threads_tasks[i].parts_and_ranges.push_back({ part_idx, ranges_to_get_from_part });
threads_tasks[i].sum_marks_in_parts.push_back(marks_in_ranges);
if (marks_in_ranges != 0)
remaining_thread_tasks.insert(i);
}
2015-06-24 11:03:53 +00:00
}
}
/** Если некоторых запрошенных столбцов нет в куске,
* то выясняем, какие столбцы может быть необходимо дополнительно прочитать,
* чтобы можно было вычислить DEFAULT выражение для этих столбцов.
* Добавляет их в columns. */
NameSet injectRequiredColumns(const MergeTreeData::DataPartPtr & part, Names & columns) const
{
NameSet required_columns{std::begin(columns), std::end(columns)};
NameSet injected_columns;
2015-07-03 15:08:21 +00:00
auto all_column_files_missing = true;
2015-06-24 11:03:53 +00:00
for (size_t i = 0; i < columns.size(); ++i)
{
const auto & column_name = columns[i];
/// column has files and hence does not require evaluation
if (part->hasColumnFiles(column_name))
2015-07-03 15:08:21 +00:00
{
all_column_files_missing = false;
2015-06-24 11:03:53 +00:00
continue;
2015-07-03 15:08:21 +00:00
}
2015-06-24 11:03:53 +00:00
const auto default_it = data.column_defaults.find(column_name);
/// columns has no explicit default expression
if (default_it == std::end(data.column_defaults))
continue;
/// collect identifiers required for evaluation
IdentifierNameSet identifiers;
default_it->second.expression->collectIdentifierNames(identifiers);
for (const auto & identifier : identifiers)
{
if (data.hasColumn(identifier))
{
/// ensure each column is added only once
if (required_columns.count(identifier) == 0)
{
columns.emplace_back(identifier);
required_columns.emplace(identifier);
injected_columns.emplace(identifier);
}
}
}
}
2015-09-16 14:33:58 +00:00
/** Добавить столбец минимального размера.
* Используется в случае, когда ни один столбец не нужен или файлы отсутствуют, но нужно хотя бы знать количество строк.
* Добавляет в columns.
*/
2015-07-03 15:08:21 +00:00
if (all_column_files_missing)
{
2015-09-16 14:33:58 +00:00
const auto minimum_size_column_name = part->getMinimumSizeColumnName();
columns.push_back(minimum_size_column_name);
2015-07-03 15:08:21 +00:00
/// correctly report added column
injected_columns.insert(columns.back());
}
2015-06-24 11:03:53 +00:00
return injected_columns;
}
std::vector<std::unique_ptr<Poco::ScopedReadRWLock>> per_part_columns_lock;
MergeTreeData & data;
2015-07-03 15:08:21 +00:00
Names column_names;
const bool do_not_steal_tasks;
2015-06-24 11:03:53 +00:00
std::vector<NameSet> per_part_column_name_set;
std::vector<NamesAndTypesList> per_part_columns;
std::vector<NamesAndTypesList> per_part_pre_columns;
/// @todo actually all of these values are either true or false for the whole query, thus no vector required
std::vector<char> per_part_remove_prewhere_column;
std::vector<char> per_part_should_reorder;
2015-06-24 11:03:53 +00:00
struct Part
2015-07-23 17:54:07 +00:00
{
MergeTreeData::DataPartPtr data_part;
std::size_t part_index_in_query;
};
std::vector<Part> parts;
2015-07-23 17:54:07 +00:00
struct ThreadTask
2015-07-23 17:54:07 +00:00
{
struct PartIndexAndRange
2015-07-23 17:54:07 +00:00
{
std::size_t part_idx;
MarkRanges ranges;
};
std::vector<PartIndexAndRange> parts_and_ranges;
2015-07-23 17:54:07 +00:00
std::vector<std::size_t> sum_marks_in_parts;
};
std::vector<ThreadTask> threads_tasks;
2015-07-23 17:54:07 +00:00
2015-09-25 13:39:06 +00:00
std::set<std::size_t> remaining_thread_tasks;
2015-07-23 17:54:07 +00:00
2015-06-24 11:03:53 +00:00
mutable std::mutex mutex;
};
using MergeTreeReadPoolPtr = std::shared_ptr<MergeTreeReadPool>;
}