[fix](sink) crash caused by wild pointer of counter in VDataStreamSen…

…der (#26947) (#27148)

If preparation fails, the counter _peak_memory_usage_counter will be a wild pointer.

*** SIGSEGV address not mapped to object (@0x454d49545f) received by PID 16992 (TID 18856 OR 0x7f4d05444700) from PID 1296651359; stack trace: ***
 0# doris::signal::(anonymous namespace)::FailureSignalHandler(int, siginfo_t*, void*) at /root/doris/be/src/common/signal_handler.h:417
 1# os::Linux::chained_handler(int, siginfo*, void*) in /app/doris/Nexchip-doris-1.2.4.2-bin-x86_64/java8/jre/lib/amd64/server/libjvm.so
 2# JVM_handle_linux_signal in /app/doris/Nexchip-doris-1.2.4.2-bin-x86_64/java8/jre/lib/amd64/server/libjvm.so
 3# signalHandler(int, siginfo*, void*) in /app/doris/Nexchip-doris-1.2.4.2-bin-x86_64/java8/jre/lib/amd64/server/libjvm.so
 4# 0x00007F55C85B9400 in /lib64/libc.so.6
 5# doris::vectorized::VDataStreamSender::close(doris::RuntimeState*, doris::Status) at /root/doris/be/src/vec/sink/vdata_stream_sender.cpp:734
 6# doris::PlanFragmentExecutor::close() at /root/doris/be/src/runtime/plan_fragment_executor.cpp:543
 7# doris::PlanFragmentExecutor::~PlanFragmentExecutor() at /root/doris/be/src/runtime/plan_fragment_executor.cpp:95
 8# doris::FragmentExecState::~FragmentExecState() at /root/doris/be/src/runtime/fragment_mgr.cpp:112
 9# std::_Sp_counted_ptr<doris::FragmentExecState*, (__gnu_cxx::_Lock_policy)2>::_M_dispose() at /root/ldb/ldb_toolchain/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/bits/shared_ptr_base.h:348
10# doris::FragmentMgr::exec_plan_fragment(doris::TExecPlanFragmentParams const&, std::function<void (doris::RuntimeState*, doris::Status*)> const&) at /root/doris/be/src/runtime/fragment_mgr.cpp:855
11# doris::FragmentMgr::exec_plan_fragment(doris::TExecPlanFragmentParams const&) at /root/doris/be/src/runtime/fragment_mgr.cpp:592
12# doris::PInternalServiceImpl::_exec_plan_fragment_impl(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, doris::PFragmentRequestVersion, bool) at /root/doris/be/src/service/internal_service.cpp:463
13# doris::PInternalServiceImpl::_exec_plan_fragment_in_pthread(google::protobuf::RpcController*, doris::PExecPlanFragmentRequest const*, doris::PExecPlanFragmentResult*, google::protobuf::Closure*) at /root/doris/be/src/service/internal_service.cpp:305
14# doris::WorkThreadPool<false>::work_thread(int) at /root/doris/be/src/util/work_thread_pool.hpp:160
15# execute_native_thread_routine at ../../../../../libstdc++-v3/src/c++11/thread.cc:84
16# start_thread in /lib64/libpthread.so.0
17# clone in /lib64/libc.so.6

be/src/vec/sink/vdata_stream_sender.cpp

@@ -315,14 +315,6 @@ VDataStreamSender::VDataStreamSender(RuntimeState* state, ObjectPool* pool, int
  _current_channel_idx(0),
  _part_type(sink.output_partition.type),
  _ignore_not_found(sink.__isset.ignore_not_found ? sink.ignore_not_found : true),
- _profile(nullptr),
- _serialize_batch_timer(nullptr),
- _bytes_sent_counter(nullptr),
- _local_send_timer(nullptr),
- _split_block_hash_compute_timer(nullptr),
- _split_block_distribute_by_channel_timer(nullptr),
- _blocks_sent_counter(nullptr),
- _local_bytes_send_counter(nullptr),
  _dest_node_id(sink.dest_node_id),
  _transfer_large_data_by_brpc(config::transfer_large_data_by_brpc) {
  DCHECK_GT(destinations.size(), 0);
@@ -379,18 +371,6 @@ VDataStreamSender::VDataStreamSender(ObjectPool* pool, int sender_id, const RowD
  _row_desc(row_desc),
  _current_channel_idx(0),
  _part_type(TPartitionType::UNPARTITIONED),
- _ignore_not_found(true),
- _profile(nullptr),
- _serialize_batch_timer(nullptr),
- _compress_timer(nullptr),
- _brpc_send_timer(nullptr),
- _brpc_wait_timer(nullptr),
- _bytes_sent_counter(nullptr),
- _local_send_timer(nullptr),
- _split_block_hash_compute_timer(nullptr),
- _split_block_distribute_by_channel_timer(nullptr),
- _blocks_sent_counter(nullptr),
- _local_bytes_send_counter(nullptr),
  _dest_node_id(dest_node_id) {
  _cur_pb_block = &_pb_block1;
  _name = "VDataStreamSender";

be/src/vec/sink/vdata_stream_sender.h

@@ -169,22 +169,22 @@ class VDataStreamSender : public DataSink {
  std::vector<Channel*> _channels;
  std::vector<std::shared_ptr<Channel>> _channel_shared_ptrs;
 
- RuntimeProfile* _profile; // Allocated from _pool
- RuntimeProfile::Counter* _serialize_batch_timer;
- RuntimeProfile::Counter* _compress_timer;
- RuntimeProfile::Counter* _brpc_send_timer;
- RuntimeProfile::Counter* _brpc_wait_timer;
- RuntimeProfile::Counter* _bytes_sent_counter;
- RuntimeProfile::Counter* _uncompressed_bytes_counter;
- RuntimeProfile::Counter* _ignore_rows;
- RuntimeProfile::Counter* _local_sent_rows;
- RuntimeProfile::Counter* _local_send_timer;
- RuntimeProfile::Counter* _split_block_hash_compute_timer;
- RuntimeProfile::Counter* _split_block_distribute_by_channel_timer;
- RuntimeProfile::Counter* _blocks_sent_counter;
- RuntimeProfile::Counter* _merge_block_timer;
- RuntimeProfile::Counter* _memory_usage_counter;
- RuntimeProfile::Counter* _peak_memory_usage_counter;
+ RuntimeProfile* _profile {}; // Allocated from _pool
+ RuntimeProfile::Counter* _serialize_batch_timer {};
+ RuntimeProfile::Counter* _compress_timer {};
+ RuntimeProfile::Counter* _brpc_send_timer {};
+ RuntimeProfile::Counter* _brpc_wait_timer {};
+ RuntimeProfile::Counter* _bytes_sent_counter {};
+ RuntimeProfile::Counter* _uncompressed_bytes_counter {};
+ RuntimeProfile::Counter* _ignore_rows {};
+ RuntimeProfile::Counter* _local_sent_rows {};
+ RuntimeProfile::Counter* _local_send_timer {};
+ RuntimeProfile::Counter* _split_block_hash_compute_timer {};
+ RuntimeProfile::Counter* _split_block_distribute_by_channel_timer {};
+ RuntimeProfile::Counter* _blocks_sent_counter {};
+ RuntimeProfile::Counter* _merge_block_timer {};
+ RuntimeProfile::Counter* _memory_usage_counter {};
+ RuntimeProfile::Counter* _peak_memory_usage_counter {};
 
  std::unique_ptr<MemTracker> _mem_tracker;