From b9fc9b4569323b9fa4353cd45a4b0e28a443c6b3 Mon Sep 17 00:00:00 2001 From: CurtizJ Date: Sat, 8 Feb 2020 00:07:18 +0300 Subject: [PATCH] fix order of ranges in MergeTreeDataSelectExecutor --- .../MergeTree/MergeTreeBlockReadUtils.cpp | 5 ++++- .../MergeTree/MergeTreeDataSelectExecutor.cpp | 5 ++--- .../Storages/MergeTree/MergeTreeReadPool.cpp | 21 +++++++------------ .../MergeTree/MergeTreeSelectProcessor.cpp | 7 +------ ...01078_merge_tree_read_one_thread.reference | 1 + .../01078_merge_tree_read_one_thread.sql | 16 ++++++++++++++ 6 files changed, 31 insertions(+), 24 deletions(-) create mode 100644 dbms/tests/queries/0_stateless/01078_merge_tree_read_one_thread.reference create mode 100644 dbms/tests/queries/0_stateless/01078_merge_tree_read_one_thread.sql diff --git a/dbms/src/Storages/MergeTree/MergeTreeBlockReadUtils.cpp b/dbms/src/Storages/MergeTree/MergeTreeBlockReadUtils.cpp index 920697f3c32..7f71bfc28c1 100644 --- a/dbms/src/Storages/MergeTree/MergeTreeBlockReadUtils.cpp +++ b/dbms/src/Storages/MergeTree/MergeTreeBlockReadUtils.cpp @@ -73,7 +73,10 @@ MergeTreeReadTask::MergeTreeReadTask( : data_part{data_part_}, mark_ranges{mark_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_}, remove_prewhere_column{remove_prewhere_column_}, should_reorder{should_reorder_}, size_predictor{std::move(size_predictor_)} -{} +{ + /// We need to save marks in reverse order. + std::reverse(mark_ranges.begin(), mark_ranges.end()); +} MergeTreeReadTask::~MergeTreeReadTask() = default; diff --git a/dbms/src/Storages/MergeTree/MergeTreeDataSelectExecutor.cpp b/dbms/src/Storages/MergeTree/MergeTreeDataSelectExecutor.cpp index 33b5ac4999d..6fc592bdc2c 100644 --- a/dbms/src/Storages/MergeTree/MergeTreeDataSelectExecutor.cpp +++ b/dbms/src/Storages/MergeTree/MergeTreeDataSelectExecutor.cpp @@ -718,9 +718,6 @@ Pipes MergeTreeDataSelectExecutor::spreadMarkRangesAmongStreams( for (size_t i = 0; i < parts.size(); ++i) { total_rows += parts[i].getRowsCount(); - /// Let the ranges be listed from right to left so that the leftmost range can be dropped using `pop_back()`. - std::reverse(parts[i].ranges.begin(), parts[i].ranges.end()); - sum_marks_in_parts[i] = parts[i].getMarksCount(); sum_marks += sum_marks_in_parts[i]; @@ -897,6 +894,8 @@ Pipes MergeTreeDataSelectExecutor::spreadMarkRangesAmongStreamsWithOrder( } new_ranges.emplace_back(range.begin, range.end); } + + /// Restore left-to-right order. std::reverse(new_ranges.begin(), new_ranges.end()); } diff --git a/dbms/src/Storages/MergeTree/MergeTreeReadPool.cpp b/dbms/src/Storages/MergeTree/MergeTreeReadPool.cpp index a70dfc2d78c..2003b2f0af1 100644 --- a/dbms/src/Storages/MergeTree/MergeTreeReadPool.cpp +++ b/dbms/src/Storages/MergeTree/MergeTreeReadPool.cpp @@ -28,11 +28,6 @@ MergeTreeReadPool::MergeTreeReadPool( column_names{column_names_}, do_not_steal_tasks{do_not_steal_tasks_}, predict_block_size_bytes{preferred_block_size_bytes_ > 0}, prewhere_info{prewhere_info_}, parts_ranges{parts_} { - /// reverse from right-to-left to left-to-right - /// because 'reverse' was done in MergeTreeDataSelectExecutor - for (auto & part_ranges : parts_ranges) - std::reverse(std::begin(part_ranges.ranges), std::end(part_ranges.ranges)); - /// parts don't contain duplicate MergeTreeDataPart's. const auto per_part_sum_marks = fillPerPartInfo(parts_, check_columns_); fillPerThreadInfo(threads_, sum_marks_, per_part_sum_marks, parts_, min_marks_for_concurrent_read_); @@ -79,10 +74,9 @@ MergeTreeReadTaskPtr MergeTreeReadPool::getTask(const size_t min_marks_to_read, { const auto marks_to_get_from_range = marks_in_part; - /** Ranges are in right-to-left order, because 'reverse' was done in MergeTreeDataSelectExecutor - * and that order is supported in 'fillPerThreadInfo'. - */ + /// Ranges are in right-to-left order, because 'reverse' was done in 'fillPerThreadInfo'. ranges_to_get_from_part = thread_task.ranges; + std::reverse(ranges_to_get_from_part.begin(), ranges_to_get_from_part.end()); marks_in_part -= marks_to_get_from_range; @@ -113,11 +107,7 @@ MergeTreeReadTaskPtr MergeTreeReadPool::getTask(const size_t min_marks_to_read, marks_in_part -= marks_to_get_from_range; need_marks -= marks_to_get_from_range; } - - /** Change order to right-to-left, for MergeTreeThreadSelectBlockInputStream to get ranges with .pop_back() - * (order was changed to left-to-right due to .pop_back() above). - */ - std::reverse(std::begin(ranges_to_get_from_part), std::end(ranges_to_get_from_part)); + /// Order of ranges was changed to left-to-right due to .pop_back() above. } auto curr_task_size_predictor = !per_part_size_predictor[part_idx] ? nullptr @@ -211,7 +201,6 @@ std::vector MergeTreeReadPool::fillPerPartInfo( /// Read marks for every data part. size_t sum_marks = 0; - /// Ranges are in right-to-left order, due to 'reverse' in MergeTreeDataSelectExecutor. for (const auto & range : part.ranges) sum_marks += range.end - range.begin; @@ -251,6 +240,10 @@ void MergeTreeReadPool::fillPerThreadInfo( { threads_tasks.resize(threads); + /// Let the ranges be listed from right to left so that the leftmost range can be dropped using `pop_back()`. + for (auto & part : parts) + std::reverse(part.ranges.begin(), part.ranges.end()); + const size_t min_marks_per_thread = (sum_marks - 1) / threads + 1; for (size_t i = 0; i < threads && !parts.empty(); ++i) diff --git a/dbms/src/Storages/MergeTree/MergeTreeSelectProcessor.cpp b/dbms/src/Storages/MergeTree/MergeTreeSelectProcessor.cpp index dac42859eef..0f0822cd88c 100644 --- a/dbms/src/Storages/MergeTree/MergeTreeSelectProcessor.cpp +++ b/dbms/src/Storages/MergeTree/MergeTreeSelectProcessor.cpp @@ -95,11 +95,6 @@ try task_columns = getReadTaskColumns(storage, data_part, required_columns, prewhere_info, check_columns); - /** @note you could simply swap `reverse` in if and else branches of MergeTreeDataSelectExecutor, - * and remove this reverse. */ - MarkRanges remaining_mark_ranges = all_mark_ranges; - std::reverse(remaining_mark_ranges.begin(), remaining_mark_ranges.end()); - auto size_predictor = (preferred_block_size_bytes == 0) ? nullptr : std::make_unique(data_part, ordered_names, data_part->storage.getSampleBlock()); @@ -109,7 +104,7 @@ try column_name_set = NameSet{column_names.begin(), column_names.end()}; task = std::make_unique( - data_part, remaining_mark_ranges, part_index_in_query, ordered_names, column_name_set, task_columns.columns, + data_part, all_mark_ranges, part_index_in_query, ordered_names, column_name_set, task_columns.columns, task_columns.pre_columns, prewhere_info && prewhere_info->remove_prewhere_column, task_columns.should_reorder, std::move(size_predictor)); diff --git a/dbms/tests/queries/0_stateless/01078_merge_tree_read_one_thread.reference b/dbms/tests/queries/0_stateless/01078_merge_tree_read_one_thread.reference new file mode 100644 index 00000000000..697cb3a26d7 --- /dev/null +++ b/dbms/tests/queries/0_stateless/01078_merge_tree_read_one_thread.reference @@ -0,0 +1 @@ +300 diff --git a/dbms/tests/queries/0_stateless/01078_merge_tree_read_one_thread.sql b/dbms/tests/queries/0_stateless/01078_merge_tree_read_one_thread.sql new file mode 100644 index 00000000000..41fcbe81709 --- /dev/null +++ b/dbms/tests/queries/0_stateless/01078_merge_tree_read_one_thread.sql @@ -0,0 +1,16 @@ +drop table if exists t; + +create table t (a Int, b Int) engine = MergeTree order by (a, b) settings index_granularity = 400; + +insert into t select 0, 0 from numbers(50); +insert into t select 0, 1 from numbers(350); +insert into t select 1, 2 from numbers(400); +insert into t select 2, 2 from numbers(400); +insert into t select 3, 0 from numbers(100); + +select sleep(1) format Null; -- sleep a bit to wait possible merges after insert + +set max_threads = 1; +optimize table t final; + +select sum(a) from t where a in (0, 3) and b = 0;