This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-2.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-2.1 by this push:
new a173513e27c [fix](pipelinex) exchange sink not set ready when source
limit #34241
a173513e27c is described below
commit a173513e27c0b1404c61526ae4904008264adfae
Author: Mryange <[email protected]>
AuthorDate: Mon Apr 29 20:58:09 2024 +0800
[fix](pipelinex) exchange sink not set ready when source limit #34241
---
be/src/pipeline/exec/exchange_source_operator.cpp | 6 ++++++
be/src/vec/runtime/vdata_stream_recvr.cpp | 9 +++++++++
be/src/vec/runtime/vdata_stream_recvr.h | 7 +++++++
3 files changed, 22 insertions(+)
diff --git a/be/src/pipeline/exec/exchange_source_operator.cpp
b/be/src/pipeline/exec/exchange_source_operator.cpp
index a23dae6dd62..3f11b30b752 100644
--- a/be/src/pipeline/exec/exchange_source_operator.cpp
+++ b/be/src/pipeline/exec/exchange_source_operator.cpp
@@ -22,6 +22,7 @@
#include "pipeline/exec/operator.h"
#include "runtime/exec_env.h"
#include "runtime/runtime_state.h"
+#include "util/defer_op.h"
#include "vec/common/sort/vsort_exec_exprs.h"
#include "vec/exec/vexchange_node.h"
#include "vec/exprs/vexpr_context.h"
@@ -149,6 +150,11 @@ Status ExchangeSourceOperatorX::open(RuntimeState* state) {
Status ExchangeSourceOperatorX::get_block(RuntimeState* state,
vectorized::Block* block,
bool* eos) {
auto& local_state = get_local_state(state);
+ Defer is_eos([&]() {
+ if (*eos) {
+ local_state.stream_recvr->set_sink_dep_always_ready();
+ }
+ });
SCOPED_TIMER(local_state.exec_time_counter());
if (_is_merging && !local_state.is_ready) {
RETURN_IF_ERROR(local_state.stream_recvr->create_merger(
diff --git a/be/src/vec/runtime/vdata_stream_recvr.cpp
b/be/src/vec/runtime/vdata_stream_recvr.cpp
index 1cfc359de30..fe46743053a 100644
--- a/be/src/vec/runtime/vdata_stream_recvr.cpp
+++ b/be/src/vec/runtime/vdata_stream_recvr.cpp
@@ -558,4 +558,13 @@ void VDataStreamRecvr::PipSenderQueue::add_block(Block*
block, bool use_move) {
}
}
+void VDataStreamRecvr::set_sink_dep_always_ready() const {
+ for (auto* sender_queues : sender_queues()) {
+ auto dep = sender_queues->local_channel_dependency();
+ if (dep) {
+ dep->set_always_ready();
+ }
+ }
+}
+
} // namespace doris::vectorized
diff --git a/be/src/vec/runtime/vdata_stream_recvr.h
b/be/src/vec/runtime/vdata_stream_recvr.h
index 06b05d2a577..3832a10c4f2 100644
--- a/be/src/vec/runtime/vdata_stream_recvr.h
+++ b/be/src/vec/runtime/vdata_stream_recvr.h
@@ -104,6 +104,9 @@ public:
void close();
+ // When the source reaches eos = true
+ void set_sink_dep_always_ready() const;
+
// Careful: stream sender will call this function for a local receiver,
// accessing members of receiver that are allocated by Object pool
// in this function is not safe.
@@ -193,6 +196,10 @@ public:
_local_channel_dependency = local_channel_dependency;
}
+ std::shared_ptr<pipeline::Dependency> local_channel_dependency() {
+ return _local_channel_dependency;
+ }
+
bool should_wait();
virtual Status get_batch(Block* next_block, bool* eos);
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
