inline fragment ids in proto field to repeated uint32

Author: kwannoel

proto/stream_plan.proto

@@ -31,7 +31,7 @@ message AddMutation {
   bool pause = 4;
   repeated SubscriptionUpstreamInfo subscriptions_to_add = 5;
   // nodes which should start backfill
-  common.Uint32Vector backfill_nodes_to_start = 6;
+  repeated uint32 backfill_nodes_to_start = 6;
 }
 
 message StopMutation {

src/meta/src/barrier/command.rs

@@ -26,7 +26,7 @@ use risingwave_connector::source::SplitImpl;
 use risingwave_hummock_sdk::change_log::build_table_change_log_delta;
 use risingwave_meta_model::WorkerId;
 use risingwave_pb::catalog::{CreateType, Table};
-use risingwave_pb::common::{ActorInfo, Uint32Vector};
+use risingwave_pb::common::ActorInfo;
 use risingwave_pb::source::{ConnectorSplit, ConnectorSplits};
 use risingwave_pb::stream_plan::barrier::BarrierKind as PbBarrierKind;
 use risingwave_pb::stream_plan::barrier_mutation::Mutation;
@@ -754,9 +754,7 @@ impl Command {
                     // If the cluster is already paused, the new actors should be paused too.
                     pause: is_currently_paused,
                     subscriptions_to_add,
-                    backfill_nodes_to_start: Some(Uint32Vector {
-                        data: backfill_nodes_to_start,
-                    }),
+                    backfill_nodes_to_start,
                 }));
 
                 if let CreateStreamingJobType::SinkIntoTable(ReplaceStreamJobPlan {
@@ -1009,7 +1007,7 @@ impl Command {
                     upstream_mv_table_id: upstream_mv_table_id.table_id,
                     subscriber_id: *subscription_id,
                 }],
-                backfill_nodes_to_start: Some(Uint32Vector { data: vec![] }),
+                backfill_nodes_to_start: vec![],
             })),
             Command::DropSubscription {
                 upstream_mv_table_id,

src/meta/src/barrier/rpc.rs

@@ -31,7 +31,7 @@ use risingwave_common::util::epoch::Epoch;
 use risingwave_common::util::tracing::TracingContext;
 use risingwave_connector::source::SplitImpl;
 use risingwave_meta_model::WorkerId;
-use risingwave_pb::common::{HostAddress, Uint32Vector, WorkerNode};
+use risingwave_pb::common::{HostAddress, WorkerNode};
 use risingwave_pb::hummock::HummockVersionStats;
 use risingwave_pb::stream_plan::barrier_mutation::Mutation;
 use risingwave_pb::stream_plan::{
@@ -487,12 +487,10 @@ impl ControlStreamManager {
             pause: is_paused,
             subscriptions_to_add: Default::default(),
             // TODO(kwannoel): recover using backfill order plan
-            backfill_nodes_to_start: Some(Uint32Vector {
-                data: background_jobs
-                    .values()
-                    .flat_map(|(_, fragments)| fragments.fragment_ids())
-                    .collect_vec(),
-            }),
+            backfill_nodes_to_start: background_jobs
+                .values()
+                .flat_map(|(_, fragments)| fragments.fragment_ids())
+                .collect_vec(),
         });
 
         fn resolve_jobs_committed_epoch(

src/stream/src/executor/mod.rs

@@ -165,7 +165,6 @@ pub type DispatcherMessageStreamItem = StreamExecutorResult<DispatcherMessage>;
 pub type BoxedMessageStream = BoxStream<'static, MessageStreamItem>;
 
 pub use risingwave_common::util::epoch::task_local::{curr_epoch, epoch, prev_epoch};
-use risingwave_pb::common::Uint32Vector;
 use risingwave_pb::stream_plan::stream_message_batch::{BarrierBatch, StreamMessageBatch};
 use risingwave_pb::stream_plan::throttle_mutation::RateLimit;
 
@@ -722,9 +721,7 @@ impl Mutation {
                         upstream_mv_table_id: table_id.table_id,
                     })
                     .collect(),
-                backfill_nodes_to_start: Some(Uint32Vector {
-                    data: backfill_nodes_to_start.iter().cloned().collect(),
-                }),
+                backfill_nodes_to_start: backfill_nodes_to_start.iter().copied().collect(),
             }),
             Mutation::SourceChangeSplit(changes) => PbMutation::Splits(SourceChangeSplitMutation {
                 actor_splits: changes
@@ -877,14 +874,7 @@ impl Mutation {
                         },
                     )
                     .collect(),
-                backfill_nodes_to_start: add
-                    .backfill_nodes_to_start
-                    .as_ref()
-                    .expect("not None")
-                    .data
-                    .iter()
-                    .copied()
-                    .collect(),
+                backfill_nodes_to_start: add.backfill_nodes_to_start.iter().copied().collect(),
             }),
 
             PbMutation::Splits(s) => {