This is an automated email from the ASF dual-hosted git repository.
dataroaring 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 c4386d8639 Revert two commits on opt_perf by Gabriel (#12941)
c4386d8639 is described below
commit c4386d8639b86a5f12769df098fa0211d783fcb1
Author: Jerry Hu <[email protected]>
AuthorDate: Sat Sep 24 14:50:09 2022 +0800
Revert two commits on opt_perf by Gabriel (#12941)
This reverts commit 677b837a15d55e51424f3d12d0b3edb9cd00cc5d
This reverts commit 23301adea353f4d548bd9f79a81e11140998e84f.
---
be/src/olap/comparison_predicate.h | 5 +-
be/src/olap/rowset/segment_v2/binary_plain_page.h | 27 ++---------
be/src/olap/rowset/segment_v2/segment_iterator.cpp | 13 ++---
be/src/vec/columns/column_string.cpp | 49 -------------------
be/src/vec/columns/column_string.h | 21 ++++++++-
be/src/vec/columns/predicate_column.h | 7 ++-
be/src/vec/common/sort/heap_sorter.cpp | 2 +-
be/src/vec/common/sort/heap_sorter.h | 2 +-
be/src/vec/common/sort/sorter.cpp | 41 ++++++----------
be/src/vec/common/sort/sorter.h | 6 +--
be/src/vec/common/sort/topn_sorter.cpp | 22 ++++++---
be/src/vec/common/sort/topn_sorter.h | 4 +-
be/src/vec/core/sort_block.cpp | 55 ----------------------
be/src/vec/core/sort_block.h | 3 --
be/src/vec/exec/vsort_node.cpp | 12 +++--
15 files changed, 79 insertions(+), 190 deletions(-)
diff --git a/be/src/olap/comparison_predicate.h
b/be/src/olap/comparison_predicate.h
index 00c69efe89..6fcb45bd3c 100644
--- a/be/src/olap/comparison_predicate.h
+++ b/be/src/olap/comparison_predicate.h
@@ -518,9 +518,8 @@ private:
uint16_t _base_loop(uint16_t* sel, uint16_t size, const uint8_t*
__restrict null_map,
const TArray* __restrict data_array, const TValue&
value) const {
uint16_t new_size = 0;
- uint16_t* end = sel + size;
- for (uint16_t* cursor = sel; cursor < end; ++cursor) {
- uint16_t idx = *cursor;
+ for (uint16_t i = 0; i < size; ++i) {
+ uint16_t idx = sel[i];
if constexpr (is_nullable) {
if (_opposite ^ (!null_map[idx] && _operator(data_array[idx],
value))) {
sel[new_size++] = idx;
diff --git a/be/src/olap/rowset/segment_v2/binary_plain_page.h
b/be/src/olap/rowset/segment_v2/binary_plain_page.h
index 659df55fee..1242fd9b75 100644
--- a/be/src/olap/rowset/segment_v2/binary_plain_page.h
+++ b/be/src/olap/rowset/segment_v2/binary_plain_page.h
@@ -248,14 +248,12 @@ public:
return Status::OK();
}
const size_t max_fetch = std::min(*n, static_cast<size_t>(_num_elems -
_cur_idx));
+
uint32_t len_array[max_fetch];
uint32_t start_offset_array[max_fetch];
-
- uint32_t last_offset = guarded_offset(_cur_idx);
- for (int i = 0; i < max_fetch - 1; i++, _cur_idx++) {
- const uint32_t start_offset = last_offset;
- last_offset = guarded_offset(_cur_idx + 1);
- uint32_t len = last_offset - start_offset;
+ for (int i = 0; i < max_fetch; i++, _cur_idx++) {
+ const uint32_t start_offset = offset(_cur_idx);
+ uint32_t len = offset(_cur_idx + 1) - start_offset;
len_array[i] = len;
start_offset_array[i] = start_offset;
if constexpr (Type == OLAP_FIELD_TYPE_OBJECT) {
@@ -264,14 +262,6 @@ public:
}
}
}
- _cur_idx++;
- len_array[max_fetch - 1] = offset(_cur_idx) - last_offset;
- start_offset_array[max_fetch - 1] = last_offset;
- if constexpr (Type == OLAP_FIELD_TYPE_OBJECT) {
- if (_options.need_check_bitmap) {
- RETURN_IF_ERROR(BitmapTypeCode::validate(*(_data.data +
last_offset)));
- }
- }
dst->insert_many_binary_data(_data.mutable_data(), len_array,
start_offset_array,
max_fetch);
@@ -350,20 +340,13 @@ public:
}
private:
- static constexpr size_t SIZE_OF_INT32 = sizeof(uint32_t);
// Return the offset within '_data' where the string value with index
'idx' can be found.
uint32_t offset(size_t idx) const {
if (idx >= _num_elems) {
return _offsets_pos;
}
const uint8_t* p =
- reinterpret_cast<const uint8_t*>(&_data[_offsets_pos + idx *
SIZE_OF_INT32]);
- return decode_fixed32_le(p);
- }
-
- uint32_t guarded_offset(size_t idx) const {
- const uint8_t* p =
- reinterpret_cast<const uint8_t*>(&_data[_offsets_pos + idx *
SIZE_OF_INT32]);
+ reinterpret_cast<const uint8_t*>(&_data[_offsets_pos + idx *
sizeof(uint32_t)]);
return decode_fixed32_le(p);
}
diff --git a/be/src/olap/rowset/segment_v2/segment_iterator.cpp
b/be/src/olap/rowset/segment_v2/segment_iterator.cpp
index e6435e8be1..98f2cfae27 100644
--- a/be/src/olap/rowset/segment_v2/segment_iterator.cpp
+++ b/be/src/olap/rowset/segment_v2/segment_iterator.cpp
@@ -1150,11 +1150,8 @@ Status SegmentIterator::next_batch(vectorized::Block*
block) {
}
if (!_lazy_materialization_read) {
- Status ret = Status::OK();
- if (selected_size > 0) {
- ret = _output_column_by_sel_idx(block, _first_read_column_ids,
sel_rowid_idx,
- selected_size);
- }
+ Status ret = _output_column_by_sel_idx(block,
_first_read_column_ids, sel_rowid_idx,
+ selected_size);
if (!ret.ok()) {
return ret;
}
@@ -1179,10 +1176,8 @@ Status SegmentIterator::next_batch(vectorized::Block*
block) {
// when lazy materialization enables, _first_read_column_ids =
distinct(_short_cir_pred_column_ids + _vec_pred_column_ids)
// see _vec_init_lazy_materialization
// todo(wb) need to tell input columnids from output columnids
- if (selected_size > 0) {
- RETURN_IF_ERROR(_output_column_by_sel_idx(block,
_first_read_column_ids, sel_rowid_idx,
- selected_size));
- }
+ RETURN_IF_ERROR(_output_column_by_sel_idx(block,
_first_read_column_ids, sel_rowid_idx,
+ selected_size));
}
// shrink char_type suffix zero data
diff --git a/be/src/vec/columns/column_string.cpp
b/be/src/vec/columns/column_string.cpp
index ddc21c31d8..d988791ca0 100644
--- a/be/src/vec/columns/column_string.cpp
+++ b/be/src/vec/columns/column_string.cpp
@@ -440,55 +440,6 @@ void ColumnString::reserve(size_t n) {
chars.reserve(n);
}
-void ColumnString::insert_many_strings(const StringRef* strings, size_t num) {
- DCHECK_NE(num, 0);
- offsets.reserve(offsets.size() + num);
- std::vector<const char*> start_points(1);
- auto& head = strings[0];
- start_points[0] = head.data;
- size_t new_size = head.size;
- const char* cursor = head.data + new_size;
- std::vector<const char*> end_points;
-
- const size_t old_size = chars.size();
- size_t offset = old_size;
- offset += new_size;
- offsets.push_back(offset);
- if (num == 1) {
- end_points.push_back(cursor);
- } else {
- for (size_t i = 1; i < num; i++) {
- auto& str = strings[i];
- if (cursor != str.data) {
- end_points.push_back(cursor);
- start_points.push_back(str.data);
- cursor = str.data;
- }
- size_t sz = str.size;
- offset += sz;
- new_size += sz;
- cursor += sz;
- offsets.push_back_without_reserve(offset);
- }
- end_points.push_back(cursor);
- }
- DCHECK_EQ(end_points.size(), start_points.size());
-
- chars.resize(old_size + new_size);
-
- size_t num_range = start_points.size();
- Char* data = chars.data();
-
- offset = old_size;
- for (size_t i = 0; i < num_range; i++) {
- uint32_t len = end_points[i] - start_points[i];
- if (len) {
- memcpy(data + offset, start_points[i], len);
- offset += len;
- }
- }
-}
-
void ColumnString::resize(size_t n) {
auto origin_size = size();
if (origin_size > n) {
diff --git a/be/src/vec/columns/column_string.h
b/be/src/vec/columns/column_string.h
index a62a16be42..469bbdc6df 100644
--- a/be/src/vec/columns/column_string.h
+++ b/be/src/vec/columns/column_string.h
@@ -186,7 +186,26 @@ public:
}
};
- void insert_many_strings(const StringRef* strings, size_t num) override;
+ void insert_many_strings(const StringRef* strings, size_t num) override {
+ size_t new_size = 0;
+ for (size_t i = 0; i < num; i++) {
+ new_size += strings[i].size;
+ }
+
+ const size_t old_size = chars.size();
+ chars.resize(old_size + new_size);
+
+ Char* data = chars.data();
+ size_t offset = old_size;
+ for (size_t i = 0; i < num; i++) {
+ uint32_t len = strings[i].size;
+ if (len) {
+ memcpy(data + offset, strings[i].data, len);
+ offset += len;
+ }
+ offsets.push_back(offset);
+ }
+ }
void insert_many_dict_data(const int32_t* data_array, size_t start_index,
const StringRef* dict,
size_t num, uint32_t /*dict_num*/) override {
diff --git a/be/src/vec/columns/predicate_column.h
b/be/src/vec/columns/predicate_column.h
index 4d44217b9a..d5ad52b6ac 100644
--- a/be/src/vec/columns/predicate_column.h
+++ b/be/src/vec/columns/predicate_column.h
@@ -279,10 +279,9 @@ public:
data.resize(org_elem_num + num);
uint32_t fragment_start_offset = start_offset_array[0];
size_t fragment_len = 0;
- StringValue* cursor = data.data() + org_elem_num;
- for (size_t i = 0; i < num; i++, cursor++) {
- cursor->ptr = destination + fragment_len;
- cursor->len = len_array[i];
+ for (size_t i = 0; i < num; i++) {
+ data[org_elem_num + i].ptr = destination + fragment_len;
+ data[org_elem_num + i].len = len_array[i];
fragment_len += len_array[i];
// Compute the largest continuous memcpy block and copy them.
// If this is the last element in data array, then should copy
the current memory block.
diff --git a/be/src/vec/common/sort/heap_sorter.cpp
b/be/src/vec/common/sort/heap_sorter.cpp
index 6520b005a4..795bd66941 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) {
+Status HeapSorter::append_block(Block* block, bool* mem_reuse) {
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 6f644a9d92..f725d585c2 100644
--- a/be/src/vec/common/sort/heap_sorter.h
+++ b/be/src/vec/common/sort/heap_sorter.h
@@ -63,7 +63,7 @@ public:
_materialize_timer = ADD_TIMER(runtime_profile, "MaterializeTime");
}
- Status append_block(Block* block) override;
+ Status append_block(Block* block, bool* mem_reuse) override;
Status prepare_for_read() override;
diff --git a/be/src/vec/common/sort/sorter.cpp
b/be/src/vec/common/sort/sorter.cpp
index c766f13f74..9b5641075d 100644
--- a/be/src/vec/common/sort/sorter.cpp
+++ b/be/src/vec/common/sort/sorter.cpp
@@ -72,33 +72,25 @@ Status
MergeSorterState::merge_sort_read(doris::RuntimeState* state,
return Status::OK();
}
-Status Sorter::partial_sort(Block& src_block, Block& dest_block) {
- bool materialized = false;
+Status Sorter::partial_sort(Block& block) {
if (_vsort_exec_exprs.need_materialize_tuple()) {
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(&src_block,
&valid_column_ids[i]));
+ RETURN_IF_ERROR(output_tuple_expr_ctxs[i]->execute(&block,
&valid_column_ids[i]));
}
Block new_block;
for (auto column_id : valid_column_ids) {
- new_block.insert(src_block.get_by_position(column_id));
+ new_block.insert(block.get_by_position(column_id));
}
- dest_block.swap(new_block);
- materialized = true;
+ 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];
- if (materialized) {
- RETURN_IF_ERROR(
- ordering_expr->execute(&src_block,
&_sort_description[i].column_number));
- } else {
- RETURN_IF_ERROR(
- ordering_expr->execute(&dest_block,
&_sort_description[i].column_number));
- }
+ RETURN_IF_ERROR(ordering_expr->execute(&block,
&_sort_description[i].column_number));
_sort_description[i].direction = _is_asc_order[i] ? 1 : -1;
_sort_description[i].nulls_direction =
@@ -107,11 +99,7 @@ Status Sorter::partial_sort(Block& src_block, Block&
dest_block) {
{
SCOPED_TIMER(_partial_sort_timer);
- if (materialized) {
- sort_block(dest_block, _sort_description, _offset + _limit);
- } else {
- sort_block(src_block, dest_block, _sort_description, _offset +
_limit);
- }
+ sort_block(block, _sort_description, _offset + _limit);
}
return Status::OK();
@@ -123,7 +111,7 @@ 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) {
+Status FullSorter::append_block(Block* block, bool* mem_reuse) {
DCHECK(block->rows() > 0);
{
SCOPED_TIMER(_merge_block_timer);
@@ -159,9 +147,8 @@ Status FullSorter::get_next(RuntimeState* state, Block*
block, bool* eos) {
}
Status FullSorter::_do_sort() {
- 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 block = _state->unsorted_block->to_block(0);
+ RETURN_IF_ERROR(partial_sort(block));
// dispose TOP-N logic
if (_limit != -1) {
// Here is a little opt to reduce the mem uasge, we build a max heap
@@ -169,21 +156,21 @@ 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(desc_block));
- _state->num_rows += desc_block.rows();
+ _state->sorted_blocks.emplace_back(std::move(block));
+ _state->num_rows += block.rows();
_block_priority_queue.emplace(_pool->add(
new MergeSortCursorImpl(_state->sorted_blocks.back(),
_sort_description)));
} else {
MergeSortBlockCursor block_cursor(
- _pool->add(new MergeSortCursorImpl(desc_block,
_sort_description)));
+ _pool->add(new MergeSortCursorImpl(block,
_sort_description)));
if (!block_cursor.totally_greater(_block_priority_queue.top())) {
- _state->sorted_blocks.emplace_back(std::move(desc_block));
+ _state->sorted_blocks.emplace_back(std::move(block));
_block_priority_queue.push(block_cursor);
}
}
} else {
// dispose normal sort logic
- _state->sorted_blocks.emplace_back(std::move(desc_block));
+ _state->sorted_blocks.emplace_back(std::move(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 3d6d3d8726..2e56f8012a 100644
--- a/be/src/vec/common/sort/sorter.h
+++ b/be/src/vec/common/sort/sorter.h
@@ -76,14 +76,14 @@ public:
_merge_block_timer = ADD_TIMER(runtime_profile, "MergeBlockTime");
};
- virtual Status append_block(Block* block) = 0;
+ virtual Status append_block(Block* block, bool* mem_reuse) = 0;
virtual Status prepare_for_read() = 0;
virtual Status get_next(RuntimeState* state, Block* block, bool* eos) = 0;
protected:
- Status partial_sort(Block& src_block, Block& dest_block);
+ Status partial_sort(Block& block);
SortDescription _sort_description;
VSortExecExprs& _vsort_exec_exprs;
@@ -107,7 +107,7 @@ public:
~FullSorter() override = default;
- Status append_block(Block* block) override;
+ Status append_block(Block* block, bool* mem_reuse) override;
Status prepare_for_read() override;
diff --git a/be/src/vec/common/sort/topn_sorter.cpp
b/be/src/vec/common/sort/topn_sorter.cpp
index 6ded187cfc..4ed7af6d04 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) {
+Status TopNSorter::append_block(Block* block, bool* mem_reuse) {
DCHECK(block->rows() > 0);
- RETURN_IF_ERROR(_do_sort(block));
+ RETURN_IF_ERROR(_do_sort(block, mem_reuse));
return Status::OK();
}
@@ -51,9 +51,9 @@ Status TopNSorter::get_next(RuntimeState* state, Block*
block, bool* eos) {
return Status::OK();
}
-Status TopNSorter::_do_sort(Block* block) {
- Block sorted_block = block->clone_without_columns();
- RETURN_IF_ERROR(partial_sort(*block, sorted_block));
+Status TopNSorter::_do_sort(Block* block, bool* mem_reuse) {
+ *mem_reuse = false;
+ RETURN_IF_ERROR(partial_sort(*block));
// dispose TOP-N logic
if (_limit != -1) {
// Here is a little opt to reduce the mem uasge, we build a max heap
@@ -61,20 +61,30 @@ Status TopNSorter::_do_sort(Block* block) {
// 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();
_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)));
if (!block_cursor.totally_greater(_block_priority_queue.top())) {
_state->sorted_blocks.emplace_back(std::move(sorted_block));
_block_priority_queue.push(block_cursor);
+ } else {
+ *mem_reuse = true;
+ block->clear_column_data();
}
}
} else {
- return Status::InternalError("Should not reach TopN sorter for full
sort query");
+ Block sorted_block;
+ sorted_block.swap(*block);
+ // dispose normal sort logic
+ _state->sorted_blocks.emplace_back(std::move(sorted_block));
}
return Status::OK();
}
diff --git a/be/src/vec/common/sort/topn_sorter.h
b/be/src/vec/common/sort/topn_sorter.h
index 0cb4da31c2..675442f5a1 100644
--- a/be/src/vec/common/sort/topn_sorter.h
+++ b/be/src/vec/common/sort/topn_sorter.h
@@ -30,7 +30,7 @@ public:
~TopNSorter() override = default;
- Status append_block(Block* block) override;
+ Status append_block(Block* block, bool* mem_reuse) override;
Status prepare_for_read() override;
@@ -39,7 +39,7 @@ public:
static constexpr size_t TOPN_SORT_THRESHOLD = 256;
private:
- Status _do_sort(Block* block);
+ Status _do_sort(Block* block, bool* mem_reuse);
std::unique_ptr<MergeSorterState> _state;
};
diff --git a/be/src/vec/core/sort_block.cpp b/be/src/vec/core/sort_block.cpp
index 1dcab87d62..657fd58d23 100644
--- a/be/src/vec/core/sort_block.cpp
+++ b/be/src/vec/core/sort_block.cpp
@@ -114,61 +114,6 @@ 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 f20b0311a7..cc791881c1 100644
--- a/be/src/vec/core/sort_block.h
+++ b/be/src/vec/core/sort_block.h
@@ -30,9 +30,6 @@ 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 1764a7f124..54323c561c 100644
--- a/be/src/vec/exec/vsort_node.cpp
+++ b/be/src/vec/exec/vsort_node.cpp
@@ -84,13 +84,17 @@ 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;
- Block upstream_block;
+ bool mem_reuse = false;
+ std::unique_ptr<Block> upstream_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,
&eos),
+ 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));
+ if (upstream_block->rows() != 0) {
+ RETURN_IF_ERROR(_sorter->append_block(upstream_block.get(),
&mem_reuse));
RETURN_IF_CANCELLED(state);
RETURN_IF_ERROR(state->check_query_state("vsort, while sorting
input."));
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
