Refactor Update in command.rs; Update replace_table in ddl_cont_v2

risingwavelabs · Feb 18, 2024 · 2936f9f · 2936f9f
1 parent 055e45c
commit 2936f9f
Show file tree

Hide file tree

Showing 4 changed files with 14 additions and 10 deletions.
diff --git a/src/meta/src/barrier/command.rs b/src/meta/src/barrier/command.rs
@@ -648,15 +648,13 @@ impl CommandContext {
             .flat_map(build_actor_connector_splits)
             .collect();
 
-        let mutation = Mutation::Update(UpdateMutation {
+        Some(Mutation::Update(UpdateMutation {
             actor_new_dispatchers,
             merge_update: merge_updates.to_owned(),
             dropped_actors,
             actor_splits,
             ..Default::default()
-        });
-
-        Some(mutation)
+        }))
     }
 
     /// Returns the paused reason after executing the current command.

diff --git a/src/meta/src/rpc/ddl_controller_v2.rs b/src/meta/src/rpc/ddl_controller_v2.rs
@@ -197,7 +197,15 @@ impl DdlController {
         let stream_job_id = streaming_job.id();
         match streaming_job.create_type() {
             CreateType::Unspecified | CreateType::Foreground => {
-                let replace_table_job_info = ctx.replace_table_job_info.clone();
+                let replace_table_job_info = ctx.replace_table_job_info.as_ref().map(
+                    |(streaming_job, ctx, table_fragments)| {
+                        (
+                            streaming_job.clone(),
+                            ctx.merge_updates.clone(),
+                            table_fragments.table_id(),
+                        )
+                    },
+                );
 
                 self.stream_manager
                     .create_streaming_job(table_fragments, ctx)
@@ -208,13 +216,13 @@ impl DdlController {
                     .finish_streaming_job(stream_job_id as _)
                     .await?;
 
-                if let Some((streaming_job, ctx, table_fragments)) = replace_table_job_info {
+                if let Some((streaming_job, merge_updates, table_id)) = replace_table_job_info {
                     version = mgr
                         .catalog_controller
                         .finish_replace_streaming_job(
-                            table_fragments.table_id().table_id as _,
+                            table_id.table_id as _,
                             streaming_job,
-                            ctx.merge_updates.clone(),
+                            merge_updates,
                             None,
                             Some(stream_job_id),
                             None,

diff --git a/src/meta/src/stream/stream_graph/schedule.rs b/src/meta/src/stream/stream_graph/schedule.rs
@@ -326,7 +326,6 @@ impl Scheduler {
 
 /// [`Locations`] represents the parallel unit and worker locations of the actors.
 #[cfg_attr(test, derive(Default))]
-#[derive(Clone)]
 pub struct Locations {
     /// actor location map.
     pub actor_locations: BTreeMap<ActorId, ParallelUnit>,

diff --git a/src/meta/src/stream/stream_manager.rs b/src/meta/src/stream/stream_manager.rs
@@ -155,7 +155,6 @@ impl CreatingStreamingJobInfo {
 
 type CreatingStreamingJobInfoRef = Arc<CreatingStreamingJobInfo>;
 
-#[derive(Clone)]
 /// [`ReplaceTableContext`] carries one-time infos for replacing the plan of an existing table.
 ///
 /// Note: for better readability, keep this struct complete and immutable once created.