feat(snapshot-backfill): only receive mutation from barrier worker fo…

…r snapshot backfill (#18210)

src/meta/src/barrier/command.rs

@@ -497,16 +497,16 @@ impl CommandContext {
     }
 }
 
-impl CommandContext {
+impl Command {
     /// Generate a mutation for the given command.
-    pub fn to_mutation(&self) -> Option<Mutation> {
+    pub fn to_mutation(&self, current_paused_reason: Option<&PausedReason>) -> Option<Mutation> {
         let mutation =
-            match &self.command {
+            match self {
                 Command::Plain(mutation) => mutation.clone(),
 
                 Command::Pause(_) => {
                     // Only pause when the cluster is not already paused.
-                    if self.current_paused_reason.is_none() {
+                    if current_paused_reason.is_none() {
                         Some(Mutation::Pause(PauseMutation {}))
                     } else {
                         None
@@ -515,7 +515,7 @@ impl CommandContext {
 
                 Command::Resume(reason) => {
                     // Only resume when the cluster is paused with the same reason.
-                    if self.current_paused_reason == Some(*reason) {
+                    if current_paused_reason == Some(reason) {
                         Some(Mutation::Resume(ResumeMutation {}))
                     } else {
                         None
@@ -607,7 +607,7 @@ impl CommandContext {
                         added_actors,
                         actor_splits,
                         // If the cluster is already paused, the new actors should be paused too.
-                        pause: self.current_paused_reason.is_some(),
+                        pause: current_paused_reason.is_some(),
                         subscriptions_to_add,
                     }));
 
@@ -846,7 +846,7 @@ impl CommandContext {
     }
 
     pub fn actors_to_create(&self) -> Option<HashMap<WorkerId, Vec<StreamActor>>> {
-        match &self.command {
+        match self {
             Command::CreateStreamingJob { info, job_type } => {
                 let mut map = match job_type {
                     CreateStreamingJobType::Normal => HashMap::new(),
@@ -914,6 +914,13 @@ impl CommandContext {
             ..Default::default()
         }))
     }
+}
+
+impl CommandContext {
+    pub fn to_mutation(&self) -> Option<Mutation> {
+        self.command
+            .to_mutation(self.current_paused_reason.as_ref())
+    }
 
     /// Returns the paused reason after executing the current command.
     pub fn next_paused_reason(&self) -> Option<PausedReason> {

src/meta/src/barrier/creating_job/mod.rs

@@ -28,6 +28,7 @@ use risingwave_common::util::epoch::Epoch;
 use risingwave_pb::common::WorkerNode;
 use risingwave_pb::ddl_service::DdlProgress;
 use risingwave_pb::hummock::HummockVersionStats;
+use risingwave_pb::stream_plan::barrier_mutation::Mutation;
 use risingwave_pb::stream_service::{BarrierCompleteResponse, BuildActorInfo};
 use tracing::{debug, info};
 
@@ -67,6 +68,7 @@ impl CreatingStreamingJobControl {
         backfill_epoch: u64,
         version_stat: &HummockVersionStats,
         metrics: &MetaMetrics,
+        initial_mutation: Mutation,
     ) -> Self {
         info!(
             table_id = info.table_fragments.table_id().table_id,
@@ -108,7 +110,7 @@ impl CreatingStreamingJobControl {
                 backfill_epoch,
                 pending_non_checkpoint_barriers: vec![],
                 snapshot_backfill_actors,
-                actors_to_create: Some(
+                initial_barrier_info: Some((
                     actors_to_create
                         .into_iter()
                         .map(|(worker_id, actors)| {
@@ -124,7 +126,8 @@ impl CreatingStreamingJobControl {
                             )
                         })
                         .collect(),
-                ),
+                    initial_mutation,
+                )),
             },
             upstream_lag: metrics
                 .snapshot_backfill_lag
@@ -283,11 +286,12 @@ impl CreatingStreamingJobControl {
                 prev_epoch,
                 kind,
                 new_actors,
+                mutation,
             } in barriers_to_inject
             {
                 let node_to_collect = control_stream_manager.inject_barrier(
                     Some(table_id),
-                    None,
+                    mutation,
                     (&curr_epoch, &prev_epoch),
                     &kind,
                     graph_info,

src/meta/src/barrier/creating_job/status.rs

@@ -18,6 +18,7 @@ use std::sync::Arc;
 
 use risingwave_common::util::epoch::Epoch;
 use risingwave_pb::hummock::HummockVersionStats;
+use risingwave_pb::stream_plan::barrier_mutation::Mutation;
 use risingwave_pb::stream_service::barrier_complete_response::CreateMviewProgress;
 use risingwave_pb::stream_service::BuildActorInfo;
 
@@ -40,7 +41,9 @@ pub(super) enum CreatingStreamingJobStatus {
         /// The `prev_epoch` of pending non checkpoint barriers
         pending_non_checkpoint_barriers: Vec<u64>,
         snapshot_backfill_actors: HashMap<WorkerId, HashSet<ActorId>>,
-        actors_to_create: Option<HashMap<WorkerId, Vec<BuildActorInfo>>>,
+        /// Info of the first barrier: (`actors_to_create`, `mutation`)
+        /// Take the mutation out when injecting the first barrier
+        initial_barrier_info: Option<(HashMap<WorkerId, Vec<BuildActorInfo>>, Mutation)>,
     },
     ConsumingLogStore {
         graph_info: InflightGraphInfo,
@@ -60,6 +63,7 @@ pub(super) struct CreatingJobInjectBarrierInfo {
     pub prev_epoch: TracedEpoch,
     pub kind: BarrierKind,
     pub new_actors: Option<HashMap<WorkerId, Vec<BuildActorInfo>>>,
+    pub mutation: Option<Mutation>,
 }
 
 impl CreatingStreamingJobStatus {
@@ -104,12 +108,12 @@ impl CreatingStreamingJobStatus {
             graph_info,
             pending_non_checkpoint_barriers,
             ref backfill_epoch,
-            actors_to_create,
+            initial_barrier_info,
             ..
         } = self
         {
             if create_mview_tracker.has_pending_finished_jobs() {
-                assert!(actors_to_create.is_none());
+                assert!(initial_barrier_info.is_none());
                 pending_non_checkpoint_barriers.push(*backfill_epoch);
 
                 let prev_epoch = Epoch::from_physical_time(*prev_epoch_fake_physical_time);
@@ -119,6 +123,7 @@ impl CreatingStreamingJobStatus {
                         prev_epoch: TracedEpoch::new(prev_epoch),
                         kind: BarrierKind::Checkpoint(take(pending_non_checkpoint_barriers)),
                         new_actors: None,
+                        mutation: None,
                     }]
                     .into_iter()
                     .chain(pending_commands.drain(..).map(|command_ctx| {
@@ -127,6 +132,7 @@ impl CreatingStreamingJobStatus {
                             prev_epoch: command_ctx.prev_epoch.clone(),
                             kind: command_ctx.kind.clone(),
                             new_actors: None,
+                            mutation: None,
                         }
                     }))
                     .collect();
@@ -145,12 +151,19 @@ impl CreatingStreamingJobStatus {
                 } else {
                     BarrierKind::Barrier
                 };
+                let (new_actors, mutation) =
+                    if let Some((new_actors, mutation)) = initial_barrier_info.take() {
+                        (Some(new_actors), Some(mutation))
+                    } else {
+                        Default::default()
+                    };
                 Some((
                     vec![CreatingJobInjectBarrierInfo {
                         curr_epoch,
                         prev_epoch,
                         kind,
-                        new_actors: actors_to_create.take(),
+                        new_actors,
+                        mutation,
                     }],
                     None,
                 ))

src/meta/src/barrier/mod.rs

@@ -965,6 +965,19 @@ impl GlobalBarrierManager {
             info,
         } = &command
         {
+            if self.state.paused_reason().is_some() {
+                warn!("cannot create streaming job with snapshot backfill when paused");
+                for notifier in notifiers {
+                    notifier.notify_start_failed(
+                        anyhow!("cannot create streaming job with snapshot backfill when paused",)
+                            .into(),
+                    );
+                }
+                return Ok(());
+            }
+            let mutation = command
+                .to_mutation(None)
+                .expect("should have some mutation in `CreateStreamingJob` command");
             self.checkpoint_control
                 .creating_streaming_job_controls
                 .insert(
@@ -975,6 +988,7 @@ impl GlobalBarrierManager {
                         prev_epoch.value().0,
                         &self.checkpoint_control.hummock_version_stats,
                         &self.context.metrics,
+                        mutation,
                     ),
                 );
         }

src/meta/src/barrier/rpc.rs

@@ -263,39 +263,42 @@ impl ControlStreamManager {
             pre_applied_graph_info,
             applied_graph_info,
             actor_ids_to_pre_sync_mutation,
-            command_ctx.actors_to_create().map(|actors_to_create| {
-                actors_to_create
-                    .into_iter()
-                    .map(|(worker_id, actors)| {
-                        (
-                            worker_id,
-                            actors
-                                .into_iter()
-                                .map(|actor| BuildActorInfo {
-                                    actor: Some(actor),
-                                    // TODO: consider subscriber of backfilling mv
-                                    related_subscriptions: command_ctx
-                                        .subscription_info
-                                        .mv_depended_subscriptions
-                                        .iter()
-                                        .map(|(table_id, subscriptions)| {
-                                            (
-                                                table_id.table_id,
-                                                SubscriptionIds {
-                                                    subscription_ids: subscriptions
-                                                        .keys()
-                                                        .cloned()
-                                                        .collect(),
-                                                },
-                                            )
-                                        })
-                                        .collect(),
-                                })
-                                .collect_vec(),
-                        )
-                    })
-                    .collect()
-            }),
+            command_ctx
+                .command
+                .actors_to_create()
+                .map(|actors_to_create| {
+                    actors_to_create
+                        .into_iter()
+                        .map(|(worker_id, actors)| {
+                            (
+                                worker_id,
+                                actors
+                                    .into_iter()
+                                    .map(|actor| BuildActorInfo {
+                                        actor: Some(actor),
+                                        // TODO: consider subscriber of backfilling mv
+                                        related_subscriptions: command_ctx
+                                            .subscription_info
+                                            .mv_depended_subscriptions
+                                            .iter()
+                                            .map(|(table_id, subscriptions)| {
+                                                (
+                                                    table_id.table_id,
+                                                    SubscriptionIds {
+                                                        subscription_ids: subscriptions
+                                                            .keys()
+                                                            .cloned()
+                                                            .collect(),
+                                                    },
+                                                )
+                                            })
+                                            .collect(),
+                                    })
+                                    .collect_vec(),
+                            )
+                        })
+                        .collect()
+                }),
         )
     }