.

Signed-off-by: xxchan <[email protected]>

e2e_test/source_inline/kafka/shared_source_alter.slt → e2e_test/source_inline/kafka/alter/add_column_shared.slt

@@ -136,5 +136,3 @@ statement ok
 drop source s cascade;
 
 # TODO: test alter source with schema registry
-
-# TODO: test alter source rename, change owner, etc.

proto/stream_plan.proto

@@ -116,19 +116,20 @@ message BarrierMutation {
     // Stop a set of actors, used for dropping materialized views. Empty dispatchers will be
     // automatically removed.
     StopMutation stop = 4;
-    // Update outputs and hash mappings for some dispatchers, used for scaling.
+    // Update outputs and hash mappings for some dispatchers, used for scaling and replace table.
     UpdateMutation update = 5;
     // Change the split of some sources.
     SourceChangeSplitMutation splits = 6;
     // Pause the dataflow of the whole streaming graph, only used for scaling.
     PauseMutation pause = 7;
     // Resume the dataflow of the whole streaming graph, only used for scaling.
     ResumeMutation resume = 8;
-    // Throttle specific source exec or chain exec.
+    // Throttle specific source exec or backfill exec.
     ThrottleMutation throttle = 10;
     // Drop subscription on mv
     DropSubscriptionsMutation drop_subscriptions = 12;
     // Combined mutation.
+    // Currently, it can only be Add & Update, which is for sink into table.
     CombinedMutation combined = 100;
   }
 }

src/frontend/src/handler/alter_source_column.rs

@@ -121,8 +121,9 @@ pub async fn handle_alter_source_column(
     catalog.version += 1;
 
     let catalog_writer = session.catalog_writer()?;
+    let replace_plan = todo!();
     catalog_writer
-        .alter_source(catalog.to_prost(schema_id, db_id), todo!())
+        .alter_source(catalog.to_prost(schema_id, db_id), replace_plan)
         .await?;
 
     Ok(PgResponse::empty_result(StatementType::ALTER_SOURCE))

src/meta/src/barrier/command.rs

@@ -205,8 +205,8 @@ pub enum CreateStreamingJobType {
 }
 
 /// [`Command`] is the input of [`crate::barrier::GlobalBarrierWorker`]. For different commands,
-/// it will build different barriers to send, and may do different stuffs after the barrier is
-/// collected.
+/// it will [build different barriers to send](Self::to_mutation),
+/// and may [do different stuffs after the barrier is collected](CommandContext::post_collect).
 #[derive(Debug, Clone, strum::Display)]
 pub enum Command {
     /// `Plain` command generates a barrier with the mutation it carries.
@@ -498,6 +498,7 @@ impl Command {
 
                 Command::Pause(_) => {
                     // Only pause when the cluster is not already paused.
+                    // XXX: what if pause(r1) - pause(r2) - resume(r1) - resume(r2)??
                     if current_paused_reason.is_none() {
                         Some(Mutation::Pause(PauseMutation {}))
                     } else {
@@ -602,7 +603,6 @@ impl Command {
                         ..
                     }) = job_type
                     {
-                        // TODO: support in v2.
                         let update = Self::generate_update_mutation_for_replace_table(
                             old_table_fragments,
                             merge_updates,

src/meta/src/rpc/ddl_controller.rs

@@ -1132,6 +1132,7 @@ impl DdlController {
         }
     }
 
+    /// `target_replace_info`: when dropping a sink into table, we need to replace the table.
     pub async fn drop_object(
         &self,
         object_type: ObjectType,