refactor(stream): send barriers in batch when possible

proto/stream_plan.proto

@@ -175,6 +175,17 @@ 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

@@ -138,7 +138,7 @@
 }
 
 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/common/src/config.rs

@@ -1164,6 +1164,10 @@ pub struct StreamingDeveloperConfig {
     /// will be switched from jdbc postgresql sinks to rust native (connector='postgres') sinks.
     pub switch_jdbc_pg_to_native: bool,
 
+    /// The maximum number of consecutive barriers allowed in a message when sent between actors.
+    #[serde(default = "default::developer::stream_max_barrier_batch_size")]
+    pub max_barrier_batch_size: u32,
+
     /// Configure the system-wide cache row cardinality of hash join.
     /// For example, if this is set to 1000, it means we can have at most 1000 rows in cache.
     #[serde(default = "default::developer::streaming_hash_join_entry_state_max_rows")]
@@ -2031,6 +2035,10 @@ pub mod default {
             32768
         }
 
+        pub fn stream_max_barrier_batch_size() -> u32 {
+            1024
+        }
+
         pub fn stream_hash_agg_max_dirty_groups_heap_size() -> usize {
             64 << 20 // 64MB
         }

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

@@ -25,7 +25,7 @@ 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::executor::DispatcherMessageBatch;
 use risingwave_stream::task::LocalStreamManager;
 use thiserror_ext::AsReport;
 use tokio_stream::wrappers::ReceiverStream;
@@ -178,7 +178,7 @@ impl ExchangeServiceImpl {
                         message: Some(proto),
                         permits: Some(PbPermits { value: permits }),
                     };
-                    let bytes = DispatcherMessage::get_encoded_len(&response);
+                    let bytes = DispatcherMessageBatch::get_encoded_len(&response);
 
                     yield response;
 

src/config/example.toml

@@ -160,6 +160,7 @@ stream_exchange_connection_pool_size = 1
 stream_enable_auto_schema_change = true
 stream_enable_shared_source = true
 stream_switch_jdbc_pg_to_native = false
+stream_max_barrier_batch_size = 1024
 stream_hash_join_entry_state_max_rows = 30000
 
 [storage]