[minor](log) Delete useless logics checking (apache#41589)

Author: Gabriel39

be/src/pipeline/pipeline_x/local_exchange/local_exchange_sink_operator.cpp

@@ -52,28 +52,16 @@ Status LocalExchangeSinkLocalState::open(RuntimeState* state) {
     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,
                    "{}, _use_global_shuffle: {}, _channel_id: {}, _num_partitions: {}, "
                    "_num_senders: {}, _num_sources: {}, "
-                   "_running_sink_operators: {}, _running_source_operators: {}, _release_count: {}",
+                   "_running_sink_operators: {}, _running_source_operators: {}",
                    Base::debug_string(indentation_level),
                    _parent->cast<LocalExchangeSinkOperatorX>()._use_global_shuffle, _channel_id,
                    _exchanger->_num_partitions, _exchanger->_num_senders, _exchanger->_num_sources,
-                   _exchanger->_running_sink_operators, _exchanger->_running_source_operators,
-                   _release_count);
+                   _exchanger->_running_sink_operators, _exchanger->_running_source_operators);
     return fmt::to_string(debug_string_buffer);
 }
 
@@ -121,13 +109,11 @@ Status LocalExchangeSinkOperatorX::sink(RuntimeState* state, vectorized::Block*
 
     // If all exchange sources ended due to limit reached, current task should also finish
     if (local_state._exchanger->_running_source_operators == 0) {
-        local_state._release_count = true;
         local_state._shared_state->sub_running_sink_operators();
         return Status::EndOfFile("receiver eof");
     }
     if (eos) {
         local_state._shared_state->sub_running_sink_operators();
-        local_state._release_count = true;
     }
 
     return Status::OK();

be/src/pipeline/pipeline_x/local_exchange/local_exchange_sink_operator.h

@@ -38,7 +38,6 @@ class LocalExchangeSinkLocalState final : public PipelineXSinkLocalState<LocalEx
 
     Status init(RuntimeState* state, LocalSinkStateInfo& info) override;
     Status open(RuntimeState* state) override;
-    Status close(RuntimeState* state, Status exec_status) override;
     std::string debug_string(int indentation_level) const override;
 
 private:
@@ -62,7 +61,6 @@ class LocalExchangeSinkLocalState final : public PipelineXSinkLocalState<LocalEx
 
     // 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.