Skip to content

Commit

Permalink
[fix](multicast) should not ignore Status of block::merge (#35886)
Browse files Browse the repository at this point in the history
## Proposed changes

Issue Number: close #xxx

<!--Describe your changes.-->
  • Loading branch information
jacktengg authored Jun 5, 2024
1 parent f732052 commit 5a8ea30
Show file tree
Hide file tree
Showing 3 changed files with 6 additions and 4 deletions.
3 changes: 2 additions & 1 deletion be/src/pipeline/exec/multi_cast_data_stream_source.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -87,7 +87,8 @@ Status MultiCastDataStreamerSourceOperatorX::get_block(RuntimeState* state,
if (!local_state._output_expr_contexts.empty()) {
output_block = &tmp_block;
}
local_state._shared_state->multi_cast_data_streamer->pull(_consumer_id, output_block, eos);
RETURN_IF_ERROR(local_state._shared_state->multi_cast_data_streamer->pull(_consumer_id,
output_block, eos));

if (!local_state._conjuncts.empty()) {
RETURN_IF_ERROR(vectorized::VExprContext::filter_block(local_state._conjuncts, output_block,
Expand Down
5 changes: 3 additions & 2 deletions be/src/pipeline/exec/multi_cast_data_streamer.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -29,7 +29,7 @@ MultiCastBlock::MultiCastBlock(vectorized::Block* block, int used_count, size_t
block->clear();
}

void MultiCastDataStreamer::pull(int sender_idx, doris::vectorized::Block* block, bool* eos) {
Status MultiCastDataStreamer::pull(int sender_idx, doris::vectorized::Block* block, bool* eos) {
std::lock_guard l(_mutex);
auto& pos_to_pull = _sender_pos_to_read[sender_idx];
if (pos_to_pull != _multi_cast_blocks.end()) {
Expand All @@ -43,14 +43,15 @@ void MultiCastDataStreamer::pull(int sender_idx, doris::vectorized::Block* block
} else {
pos_to_pull->_used_count--;
pos_to_pull->_block->create_same_struct_block(0)->swap(*block);
(void)vectorized::MutableBlock(block).merge(*pos_to_pull->_block);
RETURN_IF_ERROR(vectorized::MutableBlock(block).merge(*pos_to_pull->_block));
pos_to_pull++;
}
}
*eos = _eos and pos_to_pull == _multi_cast_blocks.end();
if (pos_to_pull == _multi_cast_blocks.end()) {
_block_reading(sender_idx);
}
return Status::OK();
}

void MultiCastDataStreamer::close_sender(int sender_idx) {
Expand Down
2 changes: 1 addition & 1 deletion be/src/pipeline/exec/multi_cast_data_streamer.h
Original file line number Diff line number Diff line change
Expand Up @@ -50,7 +50,7 @@ class MultiCastDataStreamer {

~MultiCastDataStreamer() = default;

void pull(int sender_idx, vectorized::Block* block, bool* eos);
Status pull(int sender_idx, vectorized::Block* block, bool* eos);

void close_sender(int sender_idx);

Expand Down

0 comments on commit 5a8ea30

Please sign in to comment.