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 db3179edaf7 [pipelineX](local exchange) Fix potential timeout problem
(#33022)
db3179edaf7 is described below
commit db3179edaf7513c02c9bff45fbff9462b01e58d2
Author: Gabriel <[email protected]>
AuthorDate: Fri Mar 29 22:22:25 2024 +0800
[pipelineX](local exchange) Fix potential timeout problem (#33022)
---
.../local_exchange/local_exchange_sink_operator.cpp | 17 +++++++++++++++--
.../local_exchange/local_exchange_sink_operator.h | 2 ++
2 files changed, 17 insertions(+), 2 deletions(-)
diff --git
a/be/src/pipeline/pipeline_x/local_exchange/local_exchange_sink_operator.cpp
b/be/src/pipeline/pipeline_x/local_exchange/local_exchange_sink_operator.cpp
index 0b5d7a2cc2a..068b8d1701f 100644
--- a/be/src/pipeline/pipeline_x/local_exchange/local_exchange_sink_operator.cpp
+++ b/be/src/pipeline/pipeline_x/local_exchange/local_exchange_sink_operator.cpp
@@ -40,13 +40,25 @@ Status LocalExchangeSinkLocalState::init(RuntimeState*
state, LocalSinkStateInfo
return Status::OK();
}
+Status LocalExchangeSinkLocalState::close(RuntimeState* state, Status
exec_status) {
+ if (_closed) {
+ return Status::OK();
+ }
+ RETURN_IF_ERROR(Base::close(state, exec_status));
+ if (exec_status.ok()) {
+ DCHECK(_release_count) << "Do not finish correctly! " <<
debug_string(0);
+ }
+ return Status::OK();
+}
+
std::string LocalExchangeSinkLocalState::debug_string(int indentation_level)
const {
fmt::memory_buffer debug_string_buffer;
fmt::format_to(debug_string_buffer,
- "{}, _channel_id: {}, _num_partitions: {}, _num_senders:
{}, _num_sources: {}",
+ "{}, _channel_id: {}, _num_partitions: {}, _num_senders:
{}, _num_sources: {}, "
+ "_running_sink_operators: {}, _release_count: {}",
Base::debug_string(indentation_level), _channel_id,
_exchanger->_num_partitions,
_exchanger->_num_senders, _exchanger->_num_sources,
- _exchanger->_running_sink_operators);
+ _exchanger->_running_sink_operators, _release_count);
return fmt::to_string(debug_string_buffer);
}
@@ -59,6 +71,7 @@ Status LocalExchangeSinkOperatorX::sink(RuntimeState* state,
vectorized::Block*
if (eos) {
local_state._shared_state->sub_running_sink_operators();
+ local_state._release_count = true;
}
return Status::OK();
diff --git
a/be/src/pipeline/pipeline_x/local_exchange/local_exchange_sink_operator.h
b/be/src/pipeline/pipeline_x/local_exchange/local_exchange_sink_operator.h
index 6e787b21a6c..b7e0d754655 100644
--- a/be/src/pipeline/pipeline_x/local_exchange/local_exchange_sink_operator.h
+++ b/be/src/pipeline/pipeline_x/local_exchange/local_exchange_sink_operator.h
@@ -37,6 +37,7 @@ public:
~LocalExchangeSinkLocalState() override = default;
Status init(RuntimeState* state, LocalSinkStateInfo& info) override;
+ Status close(RuntimeState* state, Status exec_status) override;
std::string debug_string(int indentation_level) const override;
private:
@@ -58,6 +59,7 @@ private:
// Used by random passthrough exchanger
int _channel_id = 0;
+ bool _release_count = false;
};
// A single 32-bit division on a recent x64 processor has a throughput of one
instruction every six cycles with a latency of 26 cycles.
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
