This is an automated email from the ASF dual-hosted git repository.
gabriellee pushed a commit to branch opt_perf
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/opt_perf by this push:
new 8fbfc38f1e [Improvement](sort) Reuse memory in sort node (#12950)
8fbfc38f1e is described below
commit 8fbfc38f1eb43743734b24ae0cb2e7e2f7fde1ab
Author: Gabriel <gabrielleeb...@gmail.com>
AuthorDate: Sun Sep 25 09:52:50 2022 +0800
[Improvement](sort) Reuse memory in sort node (#12950)
---
be/src/vec/common/sort/heap_sorter.cpp | 2 +-
be/src/vec/common/sort/heap_sorter.h | 4 ++-
be/src/vec/common/sort/sorter.cpp | 42 ++++++++++++++++----------
be/src/vec/common/sort/sorter.h | 22 +++++++++-----
be/src/vec/common/sort/topn_sorter.cpp | 32 ++++++++------------
be/src/vec/common/sort/topn_sorter.h | 6 ++--
be/src/vec/core/sort_block.cpp | 55 ++++++++++++++++++++++++++++++++++
be/src/vec/core/sort_block.h | 3 ++
be/src/vec/exec/vsort_node.cpp | 11 ++++---
9 files changed, 123 insertions(+), 54 deletions(-)
diff --git a/be/src/vec/common/sort/heap_sorter.cpp
b/be/src/vec/common/sort/heap_sorter.cpp
index 795bd66941..6520b005a4 100644
--- a/be/src/vec/common/sort/heap_sorter.cpp
+++ b/be/src/vec/common/sort/heap_sorter.cpp
@@ -29,7 +29,7 @@ HeapSorter::HeapSorter(VSortExecExprs& vsort_exec_exprs, int
limit, int64_t offs
_topn_filter_rows(0),
_init_sort_descs(false) {}
-Status HeapSorter::append_block(Block* block, bool* mem_reuse) {
+Status HeapSorter::append_block(Block* block) {
DCHECK(block->rows() > 0);
{
SCOPED_TIMER(_materialize_timer);
diff --git a/be/src/vec/common/sort/heap_sorter.h
b/be/src/vec/common/sort/heap_sorter.h
index f725d585c2..1e8bb6a496 100644
--- a/be/src/vec/common/sort/heap_sorter.h
+++ b/be/src/vec/common/sort/heap_sorter.h
@@ -63,12 +63,14 @@ public:
_materialize_timer = ADD_TIMER(runtime_profile, "MaterializeTime");
}
- Status append_block(Block* block, bool* mem_reuse) override;
+ Status append_block(Block* block) override;
Status prepare_for_read() override;
Status get_next(RuntimeState* state, Block* block, bool* eos) override;
+ bool reuse_mem() override { return false; }
+
static constexpr size_t HEAP_SORT_THRESHOLD = 1024;
private:
diff --git a/be/src/vec/common/sort/sorter.cpp
b/be/src/vec/common/sort/sorter.cpp
index 9b5641075d..61c5d4dbec 100644
--- a/be/src/vec/common/sort/sorter.cpp
+++ b/be/src/vec/common/sort/sorter.cpp
@@ -72,25 +72,25 @@ Status
MergeSorterState::merge_sort_read(doris::RuntimeState* state,
return Status::OK();
}
-Status Sorter::partial_sort(Block& block) {
- if (_vsort_exec_exprs.need_materialize_tuple()) {
+Status Sorter::partial_sort(Block& src_block, Block& dest_block) {
+ if (_materialize_sort_exprs) {
auto output_tuple_expr_ctxs =
_vsort_exec_exprs.sort_tuple_slot_expr_ctxs();
std::vector<int> valid_column_ids(output_tuple_expr_ctxs.size());
for (int i = 0; i < output_tuple_expr_ctxs.size(); ++i) {
- RETURN_IF_ERROR(output_tuple_expr_ctxs[i]->execute(&block,
&valid_column_ids[i]));
+ RETURN_IF_ERROR(output_tuple_expr_ctxs[i]->execute(&src_block,
&valid_column_ids[i]));
}
Block new_block;
for (auto column_id : valid_column_ids) {
- new_block.insert(block.get_by_position(column_id));
+ new_block.insert(src_block.get_by_position(column_id));
}
- block.swap(new_block);
+ src_block.swap(new_block);
}
_sort_description.resize(_vsort_exec_exprs.lhs_ordering_expr_ctxs().size());
for (int i = 0; i < _sort_description.size(); i++) {
const auto& ordering_expr =
_vsort_exec_exprs.lhs_ordering_expr_ctxs()[i];
- RETURN_IF_ERROR(ordering_expr->execute(&block,
&_sort_description[i].column_number));
+ RETURN_IF_ERROR(ordering_expr->execute(&src_block,
&_sort_description[i].column_number));
_sort_description[i].direction = _is_asc_order[i] ? 1 : -1;
_sort_description[i].nulls_direction =
@@ -99,7 +99,11 @@ Status Sorter::partial_sort(Block& block) {
{
SCOPED_TIMER(_partial_sort_timer);
- sort_block(block, _sort_description, _offset + _limit);
+ if (_materialize_sort_exprs) {
+ sort_block(src_block, _sort_description, _offset + _limit);
+ } else {
+ sort_block(src_block, dest_block, _sort_description, _offset +
_limit);
+ }
}
return Status::OK();
@@ -111,11 +115,12 @@ FullSorter::FullSorter(VSortExecExprs& vsort_exec_exprs,
int limit, int64_t offs
: Sorter(vsort_exec_exprs, limit, offset, pool, is_asc_order,
nulls_first),
_state(std::unique_ptr<MergeSorterState>(new
MergeSorterState(row_desc, offset))) {}
-Status FullSorter::append_block(Block* block, bool* mem_reuse) {
+Status FullSorter::append_block(Block* block) {
DCHECK(block->rows() > 0);
{
SCOPED_TIMER(_merge_block_timer);
_state->unsorted_block->merge(*block);
+ block->clear_column_data();
}
if (_reach_limit()) {
RETURN_IF_ERROR(_do_sort());
@@ -147,8 +152,11 @@ Status FullSorter::get_next(RuntimeState* state, Block*
block, bool* eos) {
}
Status FullSorter::_do_sort() {
- Block block = _state->unsorted_block->to_block(0);
- RETURN_IF_ERROR(partial_sort(block));
+ Block src_block = _state->unsorted_block->to_block(0);
+ Block desc_block = src_block.clone_without_columns();
+ RETURN_IF_ERROR(partial_sort(src_block, desc_block));
+
+ Block* result = _materialize_sort_exprs ? &src_block : &desc_block;
// dispose TOP-N logic
if (_limit != -1) {
// Here is a little opt to reduce the mem uasge, we build a max heap
@@ -156,23 +164,25 @@ Status FullSorter::_do_sort() {
// if one block totally greater the heap top of _block_priority_queue
// we can throw the block data directly.
if (_state->num_rows < _limit) {
- _state->sorted_blocks.emplace_back(std::move(block));
- _state->num_rows += block.rows();
+ _state->num_rows += result->rows();
+ _state->sorted_blocks.emplace_back(std::move(*result));
_block_priority_queue.emplace(_pool->add(
new MergeSortCursorImpl(_state->sorted_blocks.back(),
_sort_description)));
} else {
MergeSortBlockCursor block_cursor(
- _pool->add(new MergeSortCursorImpl(block,
_sort_description)));
+ _pool->add(new MergeSortCursorImpl(*result,
_sort_description)));
if (!block_cursor.totally_greater(_block_priority_queue.top())) {
- _state->sorted_blocks.emplace_back(std::move(block));
+ _state->sorted_blocks.emplace_back(std::move(*result));
_block_priority_queue.push(block_cursor);
}
}
} else {
// dispose normal sort logic
- _state->sorted_blocks.emplace_back(std::move(block));
+ _state->sorted_blocks.emplace_back(std::move(*result));
+ }
+ if (_materialize_sort_exprs) {
+ _state->reset_block();
}
- _state->reset_block();
return Status::OK();
}
diff --git a/be/src/vec/common/sort/sorter.h b/be/src/vec/common/sort/sorter.h
index 2e56f8012a..6aa215cfcb 100644
--- a/be/src/vec/common/sort/sorter.h
+++ b/be/src/vec/common/sort/sorter.h
@@ -38,15 +38,15 @@ public:
~MergeSorterState() = default;
+ void build_merge_tree(SortDescription& sort_description);
+
+ Status merge_sort_read(doris::RuntimeState* state,
doris::vectorized::Block* block, bool* eos);
+
void reset_block() {
unsorted_block.reset(
new
MutableBlock(VectorizedUtils::create_empty_columnswithtypename(_row_desc)));
}
- void build_merge_tree(SortDescription& sort_description);
-
- Status merge_sort_read(doris::RuntimeState* state,
doris::vectorized::Block* block, bool* eos);
-
std::priority_queue<MergeSortCursor> priority_queue;
std::vector<MergeSortCursorImpl> cursors;
std::unique_ptr<MutableBlock> unsorted_block;
@@ -67,7 +67,8 @@ public:
_offset(offset),
_pool(pool),
_is_asc_order(is_asc_order),
- _nulls_first(nulls_first) {}
+ _nulls_first(nulls_first),
+
_materialize_sort_exprs(vsort_exec_exprs.need_materialize_tuple()) {}
virtual ~Sorter() = default;
@@ -76,14 +77,16 @@ public:
_merge_block_timer = ADD_TIMER(runtime_profile, "MergeBlockTime");
};
- virtual Status append_block(Block* block, bool* mem_reuse) = 0;
+ virtual Status append_block(Block* block) = 0;
virtual Status prepare_for_read() = 0;
virtual Status get_next(RuntimeState* state, Block* block, bool* eos) = 0;
+ virtual bool reuse_mem() = 0;
+
protected:
- Status partial_sort(Block& block);
+ Status partial_sort(Block& src_block, Block& dest_block);
SortDescription _sort_description;
VSortExecExprs& _vsort_exec_exprs;
@@ -97,6 +100,7 @@ protected:
RuntimeProfile::Counter* _merge_block_timer = nullptr;
std::priority_queue<MergeSortBlockCursor> _block_priority_queue;
+ bool _materialize_sort_exprs;
};
class FullSorter final : public Sorter {
@@ -107,12 +111,14 @@ public:
~FullSorter() override = default;
- Status append_block(Block* block, bool* mem_reuse) override;
+ Status append_block(Block* block) override;
Status prepare_for_read() override;
Status get_next(RuntimeState* state, Block* block, bool* eos) override;
+ bool reuse_mem() override { return true; }
+
private:
bool _reach_limit() {
return _state->unsorted_block->rows() > BUFFERED_BLOCK_SIZE ||
diff --git a/be/src/vec/common/sort/topn_sorter.cpp
b/be/src/vec/common/sort/topn_sorter.cpp
index 4ed7af6d04..7a39e4ad44 100644
--- a/be/src/vec/common/sort/topn_sorter.cpp
+++ b/be/src/vec/common/sort/topn_sorter.cpp
@@ -25,9 +25,9 @@ TopNSorter::TopNSorter(VSortExecExprs& vsort_exec_exprs, int
limit, int64_t offs
: Sorter(vsort_exec_exprs, limit, offset, pool, is_asc_order,
nulls_first),
_state(std::unique_ptr<MergeSorterState>(new
MergeSorterState(row_desc, offset))) {}
-Status TopNSorter::append_block(Block* block, bool* mem_reuse) {
+Status TopNSorter::append_block(Block* block) {
DCHECK(block->rows() > 0);
- RETURN_IF_ERROR(_do_sort(block, mem_reuse));
+ RETURN_IF_ERROR(_do_sort(block));
return Status::OK();
}
@@ -51,9 +51,11 @@ Status TopNSorter::get_next(RuntimeState* state, Block*
block, bool* eos) {
return Status::OK();
}
-Status TopNSorter::_do_sort(Block* block, bool* mem_reuse) {
- *mem_reuse = false;
- RETURN_IF_ERROR(partial_sort(*block));
+Status TopNSorter::_do_sort(Block* block) {
+ Block sorted_block = _state->unsorted_block->to_block(0);
+ RETURN_IF_ERROR(partial_sort(*block, sorted_block));
+
+ Block* result = _materialize_sort_exprs ? block : &sorted_block;
// dispose TOP-N logic
if (_limit != -1) {
// Here is a little opt to reduce the mem uasge, we build a max heap
@@ -61,30 +63,20 @@ Status TopNSorter::_do_sort(Block* block, bool* mem_reuse) {
// if one block totally greater the heap top of _block_priority_queue
// we can throw the block data directly.
if (_state->num_rows < _limit) {
- Block sorted_block;
- sorted_block.swap(*block);
- _state->sorted_blocks.emplace_back(std::move(sorted_block));
- _state->num_rows += sorted_block.rows();
+ _state->num_rows += result->rows();
+ _state->sorted_blocks.emplace_back(std::move(*result));
_block_priority_queue.emplace(_pool->add(
new MergeSortCursorImpl(_state->sorted_blocks.back(),
_sort_description)));
} else {
- Block sorted_block;
- sorted_block.swap(*block);
MergeSortBlockCursor block_cursor(
- _pool->add(new MergeSortCursorImpl(sorted_block,
_sort_description)));
+ _pool->add(new MergeSortCursorImpl(*result,
_sort_description)));
if (!block_cursor.totally_greater(_block_priority_queue.top())) {
- _state->sorted_blocks.emplace_back(std::move(sorted_block));
+ _state->sorted_blocks.emplace_back(std::move(*result));
_block_priority_queue.push(block_cursor);
- } else {
- *mem_reuse = true;
- block->clear_column_data();
}
}
} else {
- Block sorted_block;
- sorted_block.swap(*block);
- // dispose normal sort logic
- _state->sorted_blocks.emplace_back(std::move(sorted_block));
+ return Status::InternalError("Should not reach TopN sorter for full
sort query");
}
return Status::OK();
}
diff --git a/be/src/vec/common/sort/topn_sorter.h
b/be/src/vec/common/sort/topn_sorter.h
index 675442f5a1..24d4671747 100644
--- a/be/src/vec/common/sort/topn_sorter.h
+++ b/be/src/vec/common/sort/topn_sorter.h
@@ -30,16 +30,18 @@ public:
~TopNSorter() override = default;
- Status append_block(Block* block, bool* mem_reuse) override;
+ Status append_block(Block* block) override;
Status prepare_for_read() override;
Status get_next(RuntimeState* state, Block* block, bool* eos) override;
+ bool reuse_mem() override { return !_materialize_sort_exprs; }
+
static constexpr size_t TOPN_SORT_THRESHOLD = 256;
private:
- Status _do_sort(Block* block, bool* mem_reuse);
+ Status _do_sort(Block* block);
std::unique_ptr<MergeSorterState> _state;
};
diff --git a/be/src/vec/core/sort_block.cpp b/be/src/vec/core/sort_block.cpp
index 657fd58d23..1dcab87d62 100644
--- a/be/src/vec/core/sort_block.cpp
+++ b/be/src/vec/core/sort_block.cpp
@@ -114,6 +114,61 @@ void sort_block(Block& block, const SortDescription&
description, UInt64 limit)
}
}
+void sort_block(Block& src_block, Block& dest_block, const SortDescription&
description,
+ UInt64 limit) {
+ if (!src_block) {
+ return;
+ }
+
+ /// If only one column to sort by
+ if (description.size() == 1) {
+ bool reverse = description[0].direction == -1;
+
+ const IColumn* column =
+ !description[0].column_name.empty()
+ ?
src_block.get_by_name(description[0].column_name).column.get()
+ :
src_block.safe_get_by_position(description[0].column_number).column.get();
+
+ IColumn::Permutation perm;
+ column->get_permutation(reverse, limit,
description[0].nulls_direction, perm);
+
+ size_t columns = src_block.columns();
+ for (size_t i = 0; i < columns; ++i) {
+ dest_block.replace_by_position(
+ i, src_block.get_by_position(i).column->permute(perm,
limit));
+ }
+ } else {
+ size_t size = src_block.rows();
+ IColumn::Permutation perm(size);
+ for (size_t i = 0; i < size; ++i) {
+ perm[i] = i;
+ }
+
+ if (limit >= size) {
+ limit = 0;
+ }
+
+ ColumnsWithSortDescriptions columns_with_sort_desc =
+ get_columns_with_sort_description(src_block, description);
+ {
+ EqualFlags flags(size, 1);
+ EqualRange range {0, size};
+
+ for (size_t i = 0; i < columns_with_sort_desc.size(); i++) {
+ ColumnSorter sorter(columns_with_sort_desc[i], limit);
+ sorter.operator()(flags, perm, range, i ==
columns_with_sort_desc.size() - 1);
+ }
+ }
+
+ size_t columns = src_block.columns();
+ for (size_t i = 0; i < columns; ++i) {
+ dest_block.replace_by_position(
+ i, src_block.get_by_position(i).column->permute(perm,
limit));
+ }
+ }
+ src_block.clear_column_data();
+}
+
void stable_get_permutation(const Block& block, const SortDescription&
description,
IColumn::Permutation& out_permutation) {
if (!block) {
diff --git a/be/src/vec/core/sort_block.h b/be/src/vec/core/sort_block.h
index cc791881c1..f20b0311a7 100644
--- a/be/src/vec/core/sort_block.h
+++ b/be/src/vec/core/sort_block.h
@@ -30,6 +30,9 @@ namespace doris::vectorized {
/// Sort one block by `description`. If limit != 0, then the partial sort of
the first `limit` rows is produced.
void sort_block(Block& block, const SortDescription& description, UInt64 limit
= 0);
+void sort_block(Block& src_block, Block& dest_block, const SortDescription&
description,
+ UInt64 limit = 0);
+
/** Used only in StorageMergeTree to sort the data with INSERT.
* Sorting is stable. This is important for keeping the order of rows in the
CollapsingMergeTree engine
* - because based on the order of rows it is determined whether to delete
or leave groups of rows when collapsing.
diff --git a/be/src/vec/exec/vsort_node.cpp b/be/src/vec/exec/vsort_node.cpp
index 54323c561c..26ce41a3b1 100644
--- a/be/src/vec/exec/vsort_node.cpp
+++ b/be/src/vec/exec/vsort_node.cpp
@@ -84,19 +84,18 @@ Status VSortNode::open(RuntimeState* state) {
// The child has been opened and the sorter created. Sort the input.
// The final merge is done on-demand as rows are requested in get_next().
bool eos = false;
- bool mem_reuse = false;
- std::unique_ptr<Block> upstream_block;
+ std::unique_ptr<Block> upstream_block(new Block());
do {
- if (!mem_reuse) {
- upstream_block.reset(new Block());
- }
RETURN_IF_ERROR_AND_CHECK_SPAN(
child(0)->get_next_after_projects(state, upstream_block.get(),
&eos),
child(0)->get_next_span(), eos);
if (upstream_block->rows() != 0) {
- RETURN_IF_ERROR(_sorter->append_block(upstream_block.get(),
&mem_reuse));
+ RETURN_IF_ERROR(_sorter->append_block(upstream_block.get()));
RETURN_IF_CANCELLED(state);
RETURN_IF_ERROR(state->check_query_state("vsort, while sorting
input."));
+ if (!_sorter->reuse_mem()) {
+ upstream_block.reset(new Block());
+ }
}
} while (!eos);
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@doris.apache.org
For additional commands, e-mail: commits-h...@doris.apache.org
