This is an automated email from the ASF dual-hosted git repository.
kxiao pushed a commit to branch branch-2.0
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-2.0 by this push:
new 9589428e734 [fix](multicast) should not ignore Status of block::merge
#35886 (#35945)
9589428e734 is described below
commit 9589428e7345fce2254b8b151d3c78a941dd6a70
Author: TengJianPing <[email protected]>
AuthorDate: Wed Jun 19 12:12:46 2024 +0800
[fix](multicast) should not ignore Status of block::merge #35886 (#35945)
---
be/src/pipeline/exec/multi_cast_data_stream_source.cpp | 2 +-
be/src/pipeline/exec/multi_cast_data_streamer.cpp | 5 +++--
be/src/pipeline/exec/multi_cast_data_streamer.h | 2 +-
3 files changed, 5 insertions(+), 4 deletions(-)
diff --git a/be/src/pipeline/exec/multi_cast_data_stream_source.cpp
b/be/src/pipeline/exec/multi_cast_data_stream_source.cpp
index fcf9eea1902..048ca2abb3e 100644
--- a/be/src/pipeline/exec/multi_cast_data_stream_source.cpp
+++ b/be/src/pipeline/exec/multi_cast_data_stream_source.cpp
@@ -97,7 +97,7 @@ Status
MultiCastDataStreamerSourceOperator::get_block(RuntimeState* state, vecto
if (!_output_expr_contexts.empty()) {
output_block = &tmp_block;
}
- _multi_cast_data_streamer->pull(_consumer_id, output_block, &eos);
+ RETURN_IF_ERROR(_multi_cast_data_streamer->pull(_consumer_id,
output_block, &eos));
if (!_conjuncts.empty()) {
RETURN_IF_ERROR(vectorized::VExprContext::filter_block(_conjuncts,
output_block,
diff --git a/be/src/pipeline/exec/multi_cast_data_streamer.cpp
b/be/src/pipeline/exec/multi_cast_data_streamer.cpp
index 3929c6ced09..3948ef0b9cd 100644
--- a/be/src/pipeline/exec/multi_cast_data_streamer.cpp
+++ b/be/src/pipeline/exec/multi_cast_data_streamer.cpp
@@ -27,7 +27,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()) {
@@ -41,11 +41,12 @@ 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();
+ return Status::OK();
}
void MultiCastDataStreamer::close_sender(int sender_idx) {
diff --git a/be/src/pipeline/exec/multi_cast_data_streamer.h
b/be/src/pipeline/exec/multi_cast_data_streamer.h
index 92c0e24079e..8159ed87809 100644
--- a/be/src/pipeline/exec/multi_cast_data_streamer.h
+++ b/be/src/pipeline/exec/multi_cast_data_streamer.h
@@ -44,7 +44,7 @@ public:
~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);
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
