[refactor](minor) Init counter in prepare phase (#39287)

Author: Gabriel39

be/src/pipeline/exec/exchange_sink_operator.cpp

@@ -99,7 +99,10 @@ Status ExchangeSinkLocalState::init(RuntimeState* state, LocalSinkStateInfo& inf
     // Make sure brpc stub is ready before execution.
     for (int i = 0; i < channels.size(); ++i) {
         RETURN_IF_ERROR(channels[i]->init_stub(state));
+        _wait_channel_timer.push_back(_profile->add_nonzero_counter(
+                fmt::format("WaitForLocalExchangeBuffer{}", i), TUnit ::TIME_NS, timer_name, 1));
     }
+    _wait_broadcast_buffer_timer = ADD_CHILD_TIMER(_profile, "WaitForBroadcastBuffer", timer_name);
     return Status::OK();
 }
 
@@ -142,18 +145,13 @@ Status ExchangeSinkLocalState::open(RuntimeState* state) {
         _sink_buffer->set_broadcast_dependency(_broadcast_dependency);
         _broadcast_pb_mem_limiter =
                 vectorized::BroadcastPBlockHolderMemLimiter::create_shared(_broadcast_dependency);
-        _wait_broadcast_buffer_timer =
-                ADD_CHILD_TIMER(_profile, "WaitForBroadcastBuffer", timer_name);
     } else if (local_size > 0) {
         size_t dep_id = 0;
         for (auto* channel : channels) {
             if (channel->is_local()) {
                 if (auto dep = channel->get_local_channel_dependency()) {
                     _local_channels_dependency.push_back(dep);
                     DCHECK(_local_channels_dependency[dep_id] != nullptr);
-                    _wait_channel_timer.push_back(_profile->add_nonzero_counter(
-                            fmt::format("WaitForLocalExchangeBuffer{}", dep_id), TUnit ::TIME_NS,
-                            timer_name, 1));
                     dep_id++;
                 } else {
                     LOG(WARNING) << "local recvr is null: query id = "

be/src/pipeline/exec/partitioned_aggregation_sink_operator.cpp

@@ -34,6 +34,7 @@ PartitionedAggSinkLocalState::PartitionedAggSinkLocalState(DataSinkOperatorXBase
             std::make_shared<Dependency>(parent->operator_id(), parent->node_id(),
                                          parent->get_name() + "_SPILL_DEPENDENCY", true);
 }
+
 Status PartitionedAggSinkLocalState::init(doris::RuntimeState* state,
                                           doris::pipeline::LocalSinkStateInfo& info) {
     RETURN_IF_ERROR(Base::init(state, info));
@@ -66,6 +67,7 @@ Status PartitionedAggSinkLocalState::open(RuntimeState* state) {
     SCOPED_TIMER(Base::_open_timer);
     return Base::open(state);
 }
+
 Status PartitionedAggSinkLocalState::close(RuntimeState* state, Status exec_status) {
     SCOPED_TIMER(Base::exec_time_counter());
     SCOPED_TIMER(Base::_close_timer);

be/src/vec/common/sort/sorter.cpp

@@ -66,6 +66,7 @@ void MergeSorterState::reset() {
     unsorted_block_ = Block::create_unique(unsorted_block_->clone_empty());
     in_mem_sorted_bocks_size_ = 0;
 }
+
 Status MergeSorterState::add_sorted_block(Block& block) {
     auto rows = block.rows();
     if (0 == rows) {
@@ -279,6 +280,7 @@ Status FullSorter::_do_sort() {
     }
     return Status::OK();
 }
+
 size_t FullSorter::data_size() const {
     return _state->data_size();
 }