database/doris
2
0
Fork 0
Code Issues Pull Requests Actions 3 Packages Projects Releases Wiki Activity

[fix](pipelineX)error output_block settings result in excessive memory usage. #31680

Browse Source
This commit is contained in:
Mryange
2024-03-02 01:06:18 +08:00
committed by yiguolei
parent 8a44c180bf
commit ff5e9341bf
2 changed files with 11 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
be/src/pipeline/exec/union_source_operator.cpp
View File

@ -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();
}

4
be/src/pipeline/exec/union_source_operator.h
View File

@ -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();
}