[fix](exchange) Fix debug string

be/src/pipeline/exec/exchange_sink_operator.cpp

@@ -563,15 +563,16 @@ void ExchangeSinkLocalState::register_channels(pipeline::ExchangeSinkBuffer* buf
 std::string ExchangeSinkLocalState::debug_string(int indentation_level) const {
     fmt::memory_buffer debug_string_buffer;
     fmt::format_to(debug_string_buffer, "{}", Base::debug_string(indentation_level));
-    if (_sink_buffer) {
+    auto sink_buffer = _sink_buffer;
+    if (sink_buffer) {
         fmt::format_to(debug_string_buffer,
                        ", Sink Buffer: (_is_finishing = {}, blocks in queue: {}, queue capacity: "
                        "{}, queue dep: {}), _reach_limit: {}, working channels: {}, total "
                        "channels: {}, remote channels: {}, each queue size: {}",
-                       _sink_buffer->_is_failed.load(), _sink_buffer->_total_queue_size,
-                       _sink_buffer->_queue_capacity, (void*)_queue_dependency.get(),
+                       sink_buffer->_is_failed.load(), sink_buffer->_total_queue_size,
+                       sink_buffer->_queue_capacity, (void*)_queue_dependency.get(),
                        _reach_limit.load(), _working_channels_count.load(), channels.size(),
-                       _rpc_channels_num, _sink_buffer->debug_each_instance_queue_size());
+                       _rpc_channels_num, sink_buffer->debug_each_instance_queue_size());
     }
     return fmt::to_string(debug_string_buffer);
 }

be/src/pipeline/exec/exchange_source_operator.cpp

@@ -49,8 +49,9 @@ ExchangeLocalState::~ExchangeLocalState() {
 
 std::string ExchangeLocalState::debug_string(int indentation_level) const {
     fmt::memory_buffer debug_string_buffer;
+    auto recvr = stream_recvr;
     fmt::format_to(debug_string_buffer, "{}, recvr: ({})", Base::debug_string(indentation_level),
-                   stream_recvr ? stream_recvr->debug_string() : "null");
+                   recvr ? recvr->debug_string() : "null");
     return fmt::to_string(debug_string_buffer);
 }
 

be/src/pipeline/exec/group_commit_block_sink_operator.cpp

@@ -103,8 +103,9 @@ Status GroupCommitBlockSinkLocalState::close(RuntimeState* state, Status close_s
 std::string GroupCommitBlockSinkLocalState::debug_string(int indentation_level) const {
     fmt::memory_buffer debug_string_buffer;
     fmt::format_to(debug_string_buffer, "{}", Base::debug_string(indentation_level));
+    auto load_block_queue = _load_block_queue;
     fmt::format_to(debug_string_buffer, ", _load_block_queue: ({}), _base_schema_version: {}",
-                   _load_block_queue ? _load_block_queue->debug_string() : "NULL",
+                   load_block_queue ? load_block_queue->debug_string() : "NULL",
                    _parent->cast<GroupCommitBlockSinkOperatorX>()._base_schema_version);
     return fmt::to_string(debug_string_buffer);
 }

be/src/pipeline/exec/scan_operator.cpp

@@ -526,9 +526,10 @@ std::string ScanLocalState<Derived>::debug_string(int indentation_level) const {
     fmt::format_to(debug_string_buffer, "{}, _eos = {} , _opened = {}",
                    PipelineXLocalState<>::debug_string(indentation_level), _eos.load(),
                    _opened.load());
-    if (_scanner_ctx) {
+    auto ctx = _scanner_ctx;
+    if (ctx) {
         fmt::format_to(debug_string_buffer, "");
-        fmt::format_to(debug_string_buffer, ", Scanner Context: {}", _scanner_ctx->debug_string());
+        fmt::format_to(debug_string_buffer, ", Scanner Context: {}", ctx->debug_string());
     } else {
         fmt::format_to(debug_string_buffer, "");
         fmt::format_to(debug_string_buffer, ", Scanner Context: NULL");