You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by yi...@apache.org on 2024/03/01 17:06:25 UTC
(doris) branch master updated: [fix](pipelineX)error output_block settings result in excessive memory usage. #31680
This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 97b8c85b4c2 [fix](pipelineX)error output_block settings result in excessive memory usage. #31680
97b8c85b4c2 is described below
commit 97b8c85b4c235272fb525fd46f2bb2e166a5a08c
Author: Mryange <59...@users.noreply.github.com>
AuthorDate: Sat Mar 2 01:06:18 2024 +0800
[fix](pipelineX)error output_block settings result in excessive memory usage. #31680
---
be/src/pipeline/exec/union_source_operator.cpp | 15 +++++++++------
be/src/pipeline/exec/union_source_operator.h | 4 ++--
2 files changed, 11 insertions(+), 8 deletions(-)
diff --git a/be/src/pipeline/exec/union_source_operator.cpp b/be/src/pipeline/exec/union_source_operator.cpp
index 213b9bc0d3f..9e15c2bd9f6 100644
--- a/be/src/pipeline/exec/union_source_operator.cpp
+++ b/be/src/pipeline/exec/union_source_operator.cpp
@@ -26,6 +26,7 @@
#include "pipeline/exec/union_sink_operator.h"
#include "pipeline/pipeline_x/dependency.h"
#include "runtime/descriptors.h"
+#include "util/defer_op.h"
#include "vec/core/block.h"
namespace doris {
@@ -161,6 +162,13 @@ std::string UnionSourceLocalState::debug_string(int indentation_level) const {
Status UnionSourceOperatorX::get_block(RuntimeState* state, vectorized::Block* block, bool* eos) {
auto& local_state = get_local_state(state);
+ Defer set_eos {[&]() {
+ //have executing const expr, queue have no data anymore, and child could be closed
+ *eos = (_child_size == 0 && !local_state._need_read_for_const_expr) ||
+ (_child_size > 0 && local_state._shared_state->data_queue.is_all_finish() &&
+ !_has_data(state));
+ }};
+
SCOPED_TIMER(local_state.exec_time_counter());
if (local_state._need_read_for_const_expr) {
if (has_more_const(state)) {
@@ -168,7 +176,7 @@ Status UnionSourceOperatorX::get_block(RuntimeState* state, vectorized::Block* b
}
local_state._need_read_for_const_expr = has_more_const(state);
} else if (_child_size != 0) {
- std::unique_ptr<vectorized::Block> output_block = vectorized::Block::create_unique();
+ std::unique_ptr<vectorized::Block> output_block;
int child_idx = 0;
RETURN_IF_ERROR(local_state._shared_state->data_queue.get_block_from_queue(&output_block,
&child_idx));
@@ -180,11 +188,6 @@ Status UnionSourceOperatorX::get_block(RuntimeState* state, vectorized::Block* b
local_state._shared_state->data_queue.push_free_block(std::move(output_block), child_idx);
}
local_state.reached_limit(block, eos);
- //have executing const expr, queue have no data anymore, and child could be closed
- *eos = (_child_size == 0 && !local_state._need_read_for_const_expr) ||
- (_child_size > 0 && local_state._shared_state->data_queue.is_all_finish() &&
- !_has_data(state));
-
return Status::OK();
}
diff --git a/be/src/pipeline/exec/union_source_operator.h b/be/src/pipeline/exec/union_source_operator.h
index 40d02324cbd..a70d55ab606 100644
--- a/be/src/pipeline/exec/union_source_operator.h
+++ b/be/src/pipeline/exec/union_source_operator.h
@@ -137,14 +137,14 @@ public:
private:
bool _has_data(RuntimeState* state) const {
- auto& local_state = state->get_local_state(operator_id())->cast<UnionSourceLocalState>();
+ auto& local_state = get_local_state(state);
if (_child_size == 0) {
return local_state._need_read_for_const_expr;
}
return local_state._shared_state->data_queue.remaining_has_data();
}
bool has_more_const(RuntimeState* state) const {
- auto& local_state = state->get_local_state(operator_id())->cast<UnionSourceLocalState>();
+ auto& local_state = get_local_state(state);
return state->per_fragment_instance_idx() == 0 &&
local_state._const_expr_list_idx < local_state._const_expr_lists.size();
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org