refactor(stream): send barriers in batch when possible

proto/stream_plan.proto

@@ -175,6 +175,18 @@ message StreamMessage {
   }
 }
 
+message StreamMessageBatch {
+  message BarrierBatch {
+    repeated Barrier barriers = 1;
+  }
+  oneof stream_message_batch {
+    data.StreamChunk stream_chunk = 1;
+    BarrierBatch barrier_batch = 2;
+    Watermark watermark = 3;
+  }
+}
+
+
 // Hash mapping for compute node. Stores mapping from virtual node to actor id.
 message ActorMapping {
   repeated uint32 original_indices = 1;

proto/task_service.proto

@@ -107,7 +107,7 @@ message GetStreamRequest {
 }
 
 message GetStreamResponse {
-  stream_plan.StreamMessage message = 1;
+  stream_plan.StreamMessageBatch message = 1;
   // The number of permits acquired for this message, which should be sent back to the upstream with `add_permits`.
   // In theory, this can also be derived from the message itself by the receiver. Here we send it explicitly to
   // avoid any sense of inconsistency for the derivation logic, so the receiver can simply send it back verbatim.

src/compute/src/rpc/service/exchange_service.rs

@@ -25,7 +25,6 @@ use risingwave_pb::task_service::{
     permits, GetDataRequest, GetDataResponse, GetStreamRequest, GetStreamResponse, PbPermits,
 };
 use risingwave_stream::executor::exchange::permit::{MessageWithPermits, Receiver};
-use risingwave_stream::executor::DispatcherMessage;
 use risingwave_stream::task::LocalStreamManager;
 use thiserror_ext::AsReport;
 use tokio_stream::wrappers::ReceiverStream;
@@ -172,13 +171,14 @@ impl ExchangeServiceImpl {
                     permits.add_permits(permits_to_add);
                 }
                 Either::Right(MessageWithPermits { message, permits }) => {
+                    use risingwave_stream::executor::DispatcherMessageBatch;
                     let proto = message.to_protobuf();
                     // forward the acquired permit to the downstream
                     let response = GetStreamResponse {
                         message: Some(proto),
                         permits: Some(PbPermits { value: permits }),
                     };
-                    let bytes = DispatcherMessage::get_encoded_len(&response);
+                    let bytes = DispatcherMessageBatch::get_encoded_len(&response);
 
                     yield response;