refactor(meta): rename replace_table to replace_stream_job where appropriate

src/meta/service/src/ddl_service.rs

@@ -40,7 +40,7 @@ use crate::barrier::BarrierManagerRef;
 use crate::manager::sink_coordination::SinkCoordinatorManager;
 use crate::manager::{MetaSrvEnv, StreamingJob};
 use crate::rpc::ddl_controller::{
-    DdlCommand, DdlController, DropMode, ReplaceTableInfo, StreamingJobId,
+    DdlCommand, DdlController, DropMode, ReplaceStreamJobInfo, StreamingJobId,
 };
 use crate::stream::{GlobalStreamManagerRef, SourceManagerRef};
 use crate::MetaError;
@@ -91,13 +91,13 @@ impl DdlServiceImpl {
             source,
             job_type,
         }: ReplaceTablePlan,
-    ) -> ReplaceTableInfo {
+    ) -> ReplaceStreamJobInfo {
         let table = table.unwrap();
         let col_index_mapping = table_col_index_mapping
             .as_ref()
             .map(ColIndexMapping::from_protobuf);
 
-        ReplaceTableInfo {
+        ReplaceStreamJobInfo {
             streaming_job: StreamingJob::Table(
                 source,
                 table,

src/meta/src/barrier/checkpoint/control.rs

@@ -643,7 +643,7 @@ impl DatabaseCheckpointControl {
                         node.state.resps.extend(resps);
                         finished_jobs.push(TrackingJob::New(TrackingCommand {
                             info,
-                            replace_table_info: None,
+                            replace_stream_job: None,
                         }));
                     });
                 let task = task.get_or_insert_default();

src/meta/src/barrier/command.rs

@@ -84,12 +84,14 @@ pub struct Reschedule {
     pub newly_created_actors: Vec<(StreamActor, PbActorStatus)>,
 }
 
-/// Replacing an old table with a new one. All actors in the table job will be rebuilt.
-/// Used for `ALTER TABLE` ([`Command::ReplaceTable`]) and sink into table ([`Command::CreateStreamingJob`]).
+/// Replacing an old job with a new one. All actors in the job will be rebuilt.
+/// Used for `ALTER TABLE` ([`Command::ReplaceStreamJob`]) and sink into table ([`Command::CreateStreamingJob`]).
 #[derive(Debug, Clone)]
-pub struct ReplaceTablePlan {
+pub struct ReplaceStreamJobPlan {
     pub old_fragments: StreamJobFragments,
     pub new_fragments: StreamJobFragments,
+    /// Downstream jobs of the replaced job need to update their `Merge` node to
+    /// connect to the new fragment.
     pub merge_updates: Vec<MergeUpdate>,
     pub dispatchers: HashMap<ActorId, Vec<Dispatcher>>,
     /// For a table with connector, the `SourceExecutor` actor will also be rebuilt with new actor ids.
@@ -104,7 +106,7 @@ pub struct ReplaceTablePlan {
     pub tmp_id: u32,
 }
 
-impl ReplaceTablePlan {
+impl ReplaceStreamJobPlan {
     fn fragment_changes(&self) -> HashMap<FragmentId, CommandFragmentChanges> {
         let mut fragment_changes = HashMap::new();
         for fragment in self.new_fragments.fragments.values() {
@@ -206,7 +208,7 @@ pub struct SnapshotBackfillInfo {
 #[derive(Debug, Clone)]
 pub enum CreateStreamingJobType {
     Normal,
-    SinkIntoTable(ReplaceTablePlan),
+    SinkIntoTable(ReplaceStreamJobPlan),
     SnapshotBackfill(SnapshotBackfillInfo),
 }
 
@@ -271,13 +273,13 @@ pub enum Command {
         fragment_actors: HashMap<FragmentId, HashSet<ActorId>>,
     },
 
-    /// `ReplaceTable` command generates a `Update` barrier with the given `merge_updates`. This is
+    /// `ReplaceStreamJob` command generates a `Update` barrier with the given `merge_updates`. This is
     /// essentially switching the downstream of the old table fragments to the new ones, and
     /// dropping the old table fragments. Used for table schema change.
     ///
     /// This can be treated as a special case of `RescheduleFragment`, while the upstream fragment
     /// of the Merge executors are changed additionally.
-    ReplaceTable(ReplaceTablePlan),
+    ReplaceStreamJob(ReplaceStreamJobPlan),
 
     /// `SourceSplitAssignment` generates a `Splits` barrier for pushing initialized splits or
     /// changed splits.
@@ -384,7 +386,7 @@ impl Command {
                     })
                     .collect(),
             ),
-            Command::ReplaceTable(plan) => Some(plan.fragment_changes()),
+            Command::ReplaceStreamJob(plan) => Some(plan.fragment_changes()),
             Command::MergeSnapshotBackfillStreamingJobs(_) => None,
             Command::SourceSplitAssignment(_) => None,
             Command::Throttle(_) => None,
@@ -688,7 +690,7 @@ impl Command {
                         subscriptions_to_add,
                     }));
 
-                    if let CreateStreamingJobType::SinkIntoTable(ReplaceTablePlan {
+                    if let CreateStreamingJobType::SinkIntoTable(ReplaceStreamJobPlan {
                         old_fragments,
                         new_fragments: _,
                         merge_updates,
@@ -731,7 +733,7 @@ impl Command {
                     }))
                 }
 
-                Command::ReplaceTable(ReplaceTablePlan {
+                Command::ReplaceStreamJob(ReplaceStreamJobPlan {
                     old_fragments,
                     merge_updates,
                     dispatchers,
@@ -943,7 +945,7 @@ impl Command {
                 }
                 Some(map)
             }
-            Command::ReplaceTable(replace_table) => {
+            Command::ReplaceStreamJob(replace_table) => {
                 Some(replace_table.new_fragments.actors_to_create())
             }
             _ => None,

src/meta/src/barrier/context/context_impl.rs

@@ -27,7 +27,7 @@ use crate::barrier::context::{GlobalBarrierWorkerContext, GlobalBarrierWorkerCon
 use crate::barrier::progress::TrackingJob;
 use crate::barrier::{
     BarrierManagerStatus, BarrierWorkerRuntimeInfoSnapshot, Command, CreateStreamingJobCommandInfo,
-    CreateStreamingJobType, RecoveryReason, ReplaceTablePlan, Scheduled,
+    CreateStreamingJobType, RecoveryReason, ReplaceStreamJobPlan, Scheduled,
 };
 use crate::hummock::CommitEpochInfo;
 use crate::{MetaError, MetaResult};
@@ -180,7 +180,7 @@ impl CommandContext {
                     )
                     .await?;
 
-                if let CreateStreamingJobType::SinkIntoTable(ReplaceTablePlan {
+                if let CreateStreamingJobType::SinkIntoTable(ReplaceStreamJobPlan {
                     new_fragments,
                     dispatchers,
                     init_split_assignment,
@@ -223,7 +223,7 @@ impl CommandContext {
                     .await?;
             }
 
-            Command::ReplaceTable(ReplaceTablePlan {
+            Command::ReplaceStreamJob(ReplaceStreamJobPlan {
                 old_fragments,
                 new_fragments,
                 dispatchers,

src/meta/src/barrier/mod.rs

@@ -44,8 +44,8 @@ mod utils;
 mod worker;
 
 pub use self::command::{
-    BarrierKind, Command, CreateStreamingJobCommandInfo, CreateStreamingJobType, ReplaceTablePlan,
-    Reschedule, SnapshotBackfillInfo,
+    BarrierKind, Command, CreateStreamingJobCommandInfo, CreateStreamingJobType,
+    ReplaceStreamJobPlan, Reschedule, SnapshotBackfillInfo,
 };
 pub use self::info::InflightSubscriptionInfo;
 pub use self::manager::{BarrierManagerRef, GlobalBarrierManager};

src/meta/src/barrier/progress.rs

@@ -27,7 +27,7 @@ use risingwave_pb::stream_service::PbBarrierCompleteResponse;
 
 use crate::barrier::info::BarrierInfo;
 use crate::barrier::{
-    Command, CreateStreamingJobCommandInfo, CreateStreamingJobType, ReplaceTablePlan,
+    Command, CreateStreamingJobCommandInfo, CreateStreamingJobType, ReplaceStreamJobPlan,
 };
 use crate::manager::{DdlType, MetadataManager};
 use crate::model::{ActorId, BackfillUpstreamType, StreamJobFragments};
@@ -229,7 +229,7 @@ impl TrackingJob {
                     .catalog_controller
                     .finish_streaming_job(
                         streaming_job.id() as i32,
-                        command.replace_table_info.clone(),
+                        command.replace_stream_job.clone(),
                     )
                     .await?;
                 Ok(())
@@ -274,7 +274,7 @@ pub struct RecoveredTrackingJob {
 /// The command tracking by the [`CreateMviewProgressTracker`].
 pub(super) struct TrackingCommand {
     pub info: CreateStreamingJobCommandInfo,
-    pub replace_table_info: Option<ReplaceTablePlan>,
+    pub replace_stream_job: Option<ReplaceStreamJobPlan>,
 }
 
 /// Tracking is done as follows:
@@ -379,7 +379,10 @@ impl CreateMviewProgressTracker {
 
     pub(super) fn update_tracking_jobs<'a>(
         &mut self,
-        info: Option<(&CreateStreamingJobCommandInfo, Option<&ReplaceTablePlan>)>,
+        info: Option<(
+            &CreateStreamingJobCommandInfo,
+            Option<&ReplaceStreamJobPlan>,
+        )>,
         create_mview_progress: impl IntoIterator<Item = &'a CreateMviewProgress>,
         version_stats: &HummockVersionStats,
     ) {
@@ -389,9 +392,9 @@ impl CreateMviewProgressTracker {
                 let finished_commands = {
                     let mut commands = vec![];
                     // Add the command to tracker.
-                    if let Some((create_job_info, replace_table)) = info
+                    if let Some((create_job_info, replace_stream_job)) = info
                         && let Some(command) =
-                            self.add(create_job_info, replace_table, version_stats)
+                            self.add(create_job_info, replace_stream_job, version_stats)
                     {
                         // Those with no actors to track can be finished immediately.
                         commands.push(command);
@@ -429,8 +432,8 @@ impl CreateMviewProgressTracker {
             if let Some(Command::CreateStreamingJob { info, job_type }) = command {
                 match job_type {
                     CreateStreamingJobType::Normal => Some((info, None)),
-                    CreateStreamingJobType::SinkIntoTable(replace_table) => {
-                        Some((info, Some(replace_table)))
+                    CreateStreamingJobType::SinkIntoTable(replace_stream_job) => {
+                        Some((info, Some(replace_stream_job)))
                     }
                     CreateStreamingJobType::SnapshotBackfill(_) => {
                         // The progress of SnapshotBackfill won't be tracked here
@@ -494,24 +497,24 @@ impl CreateMviewProgressTracker {
     pub fn add(
         &mut self,
         info: &CreateStreamingJobCommandInfo,
-        replace_table: Option<&ReplaceTablePlan>,
+        replace_stream_job: Option<&ReplaceStreamJobPlan>,
         version_stats: &HummockVersionStats,
     ) -> Option<TrackingJob> {
         tracing::trace!(?info, "add job to track");
         let (info, actors, replace_table_info) = {
             let CreateStreamingJobCommandInfo {
-                stream_job_fragments: table_fragments,
+                stream_job_fragments,
                 ..
             } = info;
-            let actors = table_fragments.tracking_progress_actor_ids();
+            let actors = stream_job_fragments.tracking_progress_actor_ids();
             if actors.is_empty() {
                 // The command can be finished immediately.
                 return Some(TrackingJob::New(TrackingCommand {
                     info: info.clone(),
-                    replace_table_info: replace_table.cloned(),
+                    replace_stream_job: replace_stream_job.cloned(),
                 }));
             }
-            (info.clone(), actors, replace_table.cloned())
+            (info.clone(), actors, replace_stream_job.cloned())
         };
 
         let CreateStreamingJobCommandInfo {
@@ -567,7 +570,7 @@ impl CreateMviewProgressTracker {
             // that the sink job has been created.
             Some(TrackingJob::New(TrackingCommand {
                 info,
-                replace_table_info,
+                replace_stream_job: replace_table_info,
             }))
         } else {
             let old = self.progress_map.insert(
@@ -576,7 +579,7 @@ impl CreateMviewProgressTracker {
                     progress,
                     TrackingJob::New(TrackingCommand {
                         info,
-                        replace_table_info,
+                        replace_stream_job: replace_table_info,
                     }),
                 ),
             );