This is an automated email from the ASF dual-hosted git repository.
zclllyybb pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new f401ccb56de [refine](pipeline) Add exchange source debug details
(#65167)
f401ccb56de is described below
commit f401ccb56dee7b79dad9917003be1d0d7b49b6b2
Author: Mryange <[email protected]>
AuthorDate: Mon Jul 6 10:19:23 2026 +0800
[refine](pipeline) Add exchange source debug details (#65167)
Intermittent query timeouts can leave an exchange receiver waiting with
non-zero remaining senders, but the existing source-side debug string
does not show enough information to tell why the receiver was
initialized to wait for senders.
This PR extends the exchange source debug string with the local task
index, effective sender count, bucket-shuffle orphan state, partition
type, and bucket destination instances. These fields make timeout dumps
more useful when only the stuck source operator is still alive,
especially for checking whether the receiver should have been treated as
a bucket-shuffle orphan or was initialized with the wrong sender count.
---
be/src/exec/operator/exchange_source_operator.cpp | 51 ++++++++++++++++++++---
1 file changed, 46 insertions(+), 5 deletions(-)
diff --git a/be/src/exec/operator/exchange_source_operator.cpp
b/be/src/exec/operator/exchange_source_operator.cpp
index acdfe02d048..c3495737f84 100644
--- a/be/src/exec/operator/exchange_source_operator.cpp
+++ b/be/src/exec/operator/exchange_source_operator.cpp
@@ -18,6 +18,7 @@
#include "exec/operator/exchange_source_operator.h"
#include <fmt/core.h>
+#include <gen_cpp/Partitions_types.h>
#include <cstdint>
#include <memory>
@@ -32,6 +33,33 @@
#include "util/defer_op.h"
namespace doris {
+namespace {
+
+std::string partition_type_to_string(TPartitionType::type partition_type) {
+ auto it = _TPartitionType_VALUES_TO_NAMES.find(partition_type);
+ if (it == _TPartitionType_VALUES_TO_NAMES.end()) {
+ return fmt::format("UNKNOWN({})", partition_type);
+ }
+ return it->second;
+}
+
+void append_limited_instance_set(fmt::memory_buffer& buffer, const
std::set<int>& instance_set) {
+ int count = 0;
+ for (int instance_idx : instance_set) {
+ if (count > 0) {
+ fmt::format_to(buffer, ",");
+ }
+ if (count >= 16) {
+ fmt::format_to(buffer, "...");
+ break;
+ }
+ fmt::format_to(buffer, "{}", instance_idx);
+ ++count;
+ }
+}
+
+} // namespace
+
ExchangeLocalState::ExchangeLocalState(RuntimeState* state, OperatorXBase*
parent)
: Base(state, parent), num_rows_skipped(0), is_ready(false) {}
@@ -48,8 +76,14 @@ ExchangeLocalState::~ExchangeLocalState() {
std::string ExchangeLocalState::debug_string(int indentation_level) const {
fmt::memory_buffer debug_string_buffer;
- fmt::format_to(debug_string_buffer, "{}, recvr: ({})",
Base::debug_string(indentation_level),
- stream_recvr->debug_string());
+ auto& p = _parent->cast<ExchangeSourceOperatorX>();
+ const bool is_bucket_shuffle_orphan =
p.is_bucket_shuffle_orphan_instance(local_task_idx);
+ const int effective_num_senders = is_bucket_shuffle_orphan ? 0 :
p.num_senders();
+ fmt::format_to(debug_string_buffer,
+ "{}, Source State: (local_task_idx = {},
effective_num_senders = {}, "
+ "is_bucket_shuffle_orphan = {}), recvr: ({})",
+ Base::debug_string(indentation_level), local_task_idx,
effective_num_senders,
+ is_bucket_shuffle_orphan, stream_recvr->debug_string());
return fmt::to_string(debug_string_buffer);
}
@@ -57,15 +91,22 @@ std::string ExchangeSourceOperatorX::debug_string(int
indentation_level) const {
fmt::memory_buffer debug_string_buffer;
fmt::format_to(debug_string_buffer, "{}",
OperatorX<ExchangeLocalState>::debug_string(indentation_level));
- fmt::format_to(debug_string_buffer, ", Info: (_num_senders = {},
_is_merging = {})",
- _num_senders, _is_merging);
+ fmt::format_to(debug_string_buffer,
+ ", Info: (_num_senders = {}, _is_merging = {},
_partition_type = {}, "
+ "_has_bucket_dest_instances = {},
_bucket_dest_instances_count = {}, "
+ "_bucket_dest_instances = [",
+ _num_senders, _is_merging,
partition_type_to_string(_partition_type),
+ _has_bucket_dest_instances, _bucket_dest_instances.size());
+ append_limited_instance_set(debug_string_buffer, _bucket_dest_instances);
+ fmt::format_to(debug_string_buffer, "])");
return fmt::to_string(debug_string_buffer);
}
void ExchangeLocalState::create_stream_recvr(RuntimeState* state) {
auto& p = _parent->cast<ExchangeSourceOperatorX>();
int num_senders = p.num_senders();
- if (p.is_bucket_shuffle_orphan_instance(local_task_idx)) {
+ const bool is_bucket_shuffle_orphan =
p.is_bucket_shuffle_orphan_instance(local_task_idx);
+ if (is_bucket_shuffle_orphan) {
// Bucket-routed senders open one channel per destination entry (one
per bucket),
// so an instance owning no bucket never gets a channel — and never
gets EOS.
// Start its receiver with zero senders so it reports EOS immediately
instead of
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
