[Chore](exchange) add some check about eos with exhcnage source (#57357)

This pull request introduces improvements to the handling of data limits
and error detection in the data exchange and streaming components. The
main changes ensure correct behavior when limits are unset and add
better error reporting for unexpected sender states.

**Improvements to limit handling:**

* Updated the condition in `ExchangeSourceOperatorX::get_block` (in
`exchange_source_operator.cpp`) to correctly handle cases when `_limit`
is unset (`-1`), ensuring that the row limit logic only applies when a
limit is set.

**Robustness and error detection:**

* Enhanced `VDataStreamRecvr::SenderQueue::get_batch` (in
`vdata_stream_recvr.cpp`) to return an internal error status if the data
queue is empty but there are still remaining senders, improving error
detection and debugging.

Author: BiteTheDDDDt

be/src/pipeline/exec/exchange_source_operator.cpp

@@ -186,7 +186,7 @@ Status ExchangeSourceOperatorX::get_block(RuntimeState* state, vectorized::Block
             }
         }
         // Merge actually also handles the limit, but handling the limit one more time will not cause correctness issues
-        if (local_state.num_rows_returned() + block->rows() < _limit) {
+        if (_limit == -1 || local_state.num_rows_returned() + block->rows() < _limit) {
             local_state.add_num_rows_returned(block->rows());
         } else {
             *eos = true;

be/src/vec/runtime/vdata_stream_recvr.cpp

@@ -88,7 +88,12 @@ Status VDataStreamRecvr::SenderQueue::get_batch(Block* block, bool* eos) {
         }
 
         if (_block_queue.empty()) {
-            DCHECK_EQ(_num_remaining_senders, 0);
+            if (_num_remaining_senders != 0) {
+                return Status::InternalError(
+                        "Data queue is empty but there are still remaining senders. "
+                        "_num_remaining_senders: {}",
+                        _num_remaining_senders);
+            }
             *eos = true;
             return Status::OK();
         }
@@ -171,10 +176,9 @@ Status VDataStreamRecvr::SenderQueue::add_block(std::unique_ptr<PBlock> pblock,
         auto iter = _packet_seq_map.find(be_number);
         if (iter != _packet_seq_map.end()) {
             if (iter->second >= packet_seq) {
-                LOG(WARNING) << fmt::format(
+                return Status::InternalError(
                         "packet already exist [cur_packet_id= {} receive_packet_id={}]",
                         iter->second, packet_seq);
-                return Status::OK();
             }
             iter->second = packet_seq;
         } else {