Repository: risingwavelabs/risingwave (language: python)

Issue:
Support `ALTER SOURCE ADD COLUMN` for shared source

feat(frontend): support alter add column for shared source
I hereby agree to the terms of the [RisingWave Labs, Inc. Contributor License Agreement](https://gist.github.com/TennyZhuang/f00be7f16996ea48effb049aa7be4d66#file-rw_cla).

## What's changed and what's your intention?
fe part of #19063
<!--

**Please do not leave this empty!**

Please explain **IN DETAIL** what the changes are in this PR and why they are needed:

- Summarize your change (**mandatory**)
- How does this PR work? Need a brief introduction for the changed logic (optional)
- Describe clearly one logical change and avoid lazy messages (optional)
- Describe any limitations of the current code (optional)
- Refer to a related PR or issue link (optional)

-->

## Checklist

- [ ] I have written necessary rustdoc comments
- [ ] I have added necessary unit tests and integration tests
- [ ] I have added test labels as necessary. See [details](https://github.com/risingwavelabs/risingwave/blob/main/docs/developer-guide.md#ci-labels-guide).
- [ ] I have added fuzzing tests or opened an issue to track them. (Optional, recommended for new SQL features #7934).
- [ ] My PR contains breaking changes. (If it deprecates some features, please create a tracking issue to remove them in the future).
- [ ] All checks passed in `./risedev check` (or alias, `./risedev c`)
- [ ] My PR changes performance-critical code. (Please run macro/micro-benchmarks and show the results.)
<!-- To manually trigger a benchmark, please check out [Notion](https://www.notion.so/risingwave-labs/Manually-trigger-nexmark-performance-dashboard-test-b784f1eae1cf48889b2645d020b6b7d3). -->
- [ ] My PR contains critical fixes that are necessary to be merged into the latest release. (Please check out the [details](https://github.com/risingwavelabs/risingwave/blob/main/CONTRIBUTING.md))

## Documentation

- [ ] My PR needs documentation updates. (Please use the **Release note** section below to summarize the impact on users)

## Release note

If this PR includes changes that directly affect users or other significant modifications relevant to the community, kindly draft a release note to provide a concise summary of these changes. Please prioritize highlighting the impact these changes will have on users.


<!--
Please create a release note for your changes.

Discuss technical details in the "What's changed" section, and
focus on the impact on users in the release note.

You should also mention the environment or conditions where the impact may occur.
-->

</details>

Sqlsmith: Sql feature generation
As we add more features, these should be tested via sqlsmith as well.

SQL Features:
- [ ] Generate other table functions: `generate_series`, `unnest`. See: https://github.com/singularity-data/risingwave/issues/4540
- [ ] Generate array_agg, jsonb_agg, jsonb_object_agg
- [ ] Generate select from System Tables
- [ ] Generate `NATURAL JOIN`
- [ ] Generate `JOIN` with `USING`
- [ ] Generate query w/o `GROUP BY`
- [ ] Randomly toggle session variables
- [ ] Generate `pow`/`exp`.
- [ ] Generate TopN by group: https://www.risingwave.dev/docs/current/sql-pattern-topn/
- [ ] Generate jsonb exprs, see: https://github.com/risingwavelabs/risingwave/issues/7714
- [ ] #7132
- [x] Generate order-by-clauses with `NULLS { FIRST | LAST }` (will be done in https://github.com/risingwavelabs/risingwave/pull/10864/commits/000cbc11cbc04f20abfd127b1bf4b3fd24e75c44)
- [ ] Generate `max`,`min`,`count` for `timestamptz`
- [ ] Generate `bit_and`, `bit_or` and `big_xor` in batch queries, and `bit_xor` in streaming queries.
- [ ] Generate array range access.
- [ ] https://github.com/risingwavelabs/risingwave/issues/10241
- [ ] Generate window functions other than `row_number`, `rank`.
- [ ] Generate `ALTER` DDL statements.\
- [ ] https://github.com/risingwavelabs/risingwave/issues/10516
- [ ] Generate temporal filter in sqlsmith


Discussion / hints:

Fix patch:
diff --git a/proto/ddl_service.proto b/proto/ddl_service.proto
index 3b94b4d9f2bd9..0b623e840bfa0 100644
--- a/proto/ddl_service.proto
+++ b/proto/ddl_service.proto
@@ -85,7 +85,7 @@ message CreateSinkRequest {
   catalog.Sink sink = 1;
   stream_plan.StreamFragmentGraph fragment_graph = 2;
   // It is used to provide a replace plan for the downstream table in `create sink into table` requests.
-  optional ReplaceTablePlan affected_table_change = 3;
+  optional ReplaceJobPlan affected_table_change = 3;
   // The list of object IDs that this sink depends on.
   repeated uint32 dependencies = 4;
 }
@@ -98,7 +98,7 @@ message CreateSinkResponse {
 message DropSinkRequest {
   uint32 sink_id = 1;
   bool cascade = 2;
-  optional ReplaceTablePlan affected_table_change = 3;
+  optional ReplaceJobPlan affected_table_change = 3;
 }
 
 message DropSinkResponse {
@@ -353,26 +353,38 @@ message DropIndexResponse {
   WaitVersion version = 2;
 }
 
-message ReplaceTablePlan {
-  // The new table catalog, with the correct (old) table ID and a new version.
-  // If the new version does not match the subsequent version in the meta service's
-  // catalog, this request will be rejected.
-  catalog.Table table = 1;
+message ReplaceJobPlan {
   // The new materialization plan, where all schema are updated.
-  stream_plan.StreamFragmentGraph fragment_graph = 2;
+  stream_plan.StreamFragmentGraph fragment_graph = 1;
   // The mapping from the old columns to the new columns of the table.
   // If no column modifications occur (such as for sinking into table), this will be None.
-  catalog.ColIndexMapping table_col_index_mapping = 3;
-  // Source catalog of table's associated source
-  catalog.Source source = 4;
-  TableJobType job_type = 5;
+  catalog.ColIndexMapping table_col_index_mapping = 2;
+
+  message ReplaceTable {
+    // The new table catalog, with the correct (old) table ID and a new version.
+    // If the new version does not match the subsequent version in the meta service's
+    // catalog, this request will be rejected.
+    catalog.Table table = 1;
+    // Source catalog of table's associated source
+    catalog.Source source = 2;
+    TableJobType job_type = 3;
+  }
+
+  message ReplaceSource {
+    catalog.Source source = 1;
+  }
+
+  oneof replace_job {
+    ReplaceTable replace_table = 3;
+    ReplaceSource replace_source = 4;
+  }
 }
 
-message ReplaceTablePlanRequest {
-  ReplaceTablePlan plan = 1;
+message ReplaceJobPlanRequest {
+  ReplaceJobPlan plan = 1;
 }
 
-message ReplaceTablePlanResponse {
+message ReplaceJobPlanResponse {
   common.Status status = 1;
   // The new global catalog version.
   WaitVersion version = 2;
@@ -543,7 +555,7 @@ service DdlService {
   rpc DropIndex(DropIndexRequest) returns (DropIndexResponse);
   rpc CreateFunction(CreateFunctionRequest) returns (CreateFunctionResponse);
   rpc DropFunction(DropFunctionRequest) returns (DropFunctionResponse);
-  rpc ReplaceTablePlan(ReplaceTablePlanRequest) returns (ReplaceTablePlanResponse);
+  rpc ReplaceJobPlan(ReplaceJobPlanRequest) returns (ReplaceJobPlanResponse);
   rpc GetTable(GetTableRequest) returns (GetTableResponse);
   rpc GetDdlProgress(GetDdlProgressRequest) returns (GetDdlProgressResponse);
   rpc CreateConnection(CreateConnectionRequest) returns (CreateConnectionResponse);
diff --git a/proto/frontend_service.proto b/proto/frontend_service.proto
index 7cdac815fde66..f97e91eb6a32b 100644
--- a/proto/frontend_service.proto
+++ b/proto/frontend_service.proto
@@ -15,7 +15,7 @@ message GetTableReplacePlanRequest {
 }
 
 message GetTableReplacePlanResponse {
-  ddl_service.ReplaceTablePlan replace_plan = 1;
+  ddl_service.ReplaceJobPlan replace_plan = 1;
 }
 
 service FrontendService {
diff --git a/proto/stream_plan.proto b/proto/stream_plan.proto
index fb8a5d436a01e..de5dec126fda4 100644
--- a/proto/stream_plan.proto
+++ b/proto/stream_plan.proto
@@ -60,6 +60,7 @@ message UpdateMutation {
     // Added upstream actors.
     repeated uint32 added_upstream_actor_id = 3;
     // Removed upstream actors.
+    // Note: this is empty for replace job.
     repeated uint32 removed_upstream_actor_id = 4;
   }
   // Dispatcher updates.
diff --git a/src/frontend/src/catalog/catalog_service.rs b/src/frontend/src/catalog/catalog_service.rs
index f7dcb919a6ad7..21002f0121b0b 100644
--- a/src/frontend/src/catalog/catalog_service.rs
+++ b/src/frontend/src/catalog/catalog_service.rs
@@ -27,7 +27,7 @@ use risingwave_pb::catalog::{
 };
 use risingwave_pb::ddl_service::{
     alter_name_request, alter_owner_request, alter_set_schema_request, alter_swap_rename_request,
-    create_connection_request, PbReplaceTablePlan, PbTableJobType, ReplaceTablePlan, TableJobType,
+    create_connection_request, PbReplaceJobPlan, PbTableJobType, ReplaceJobPlan, TableJobType,
     WaitVersion,
 };
 use risingwave_pb::meta::PbTableParallelism;
@@ -116,7 +116,7 @@ pub trait CatalogWriter: Send + Sync {
         &self,
         sink: PbSink,
         graph: StreamFragmentGraph,
-        affected_table_change: Option<PbReplaceTablePlan>,
+        affected_table_change: Option<PbReplaceJobPlan>,
         dependencies: HashSet<ObjectId>,
     ) -> Result<()>;
 
@@ -161,7 +161,7 @@ pub trait CatalogWriter: Send + Sync {
         &self,
         sink_id: u32,
         cascade: bool,
-        affected_table_change: Option<PbReplaceTablePlan>,
+        affected_table_change: Option<PbReplaceJobPlan>,
     ) -> Result<()>;
 
     async fn drop_subscription(&self, subscription_id: u32, cascade: bool) -> Result<()>;
@@ -329,7 +329,7 @@ impl CatalogWriter for CatalogWriterImpl {
         &self,
         sink: PbSink,
         graph: StreamFragmentGraph,
-        affected_table_change: Option<ReplaceTablePlan>,
+        affected_table_change: Option<ReplaceJobPlan>,
         dependencies: HashSet<ObjectId>,
     ) -> Result<()> {
         let version = self
@@ -425,7 +425,7 @@ impl CatalogWriter for CatalogWriterImpl {
         &self,
         sink_id: u32,
         cascade: bool,
-        affected_table_change: Option<ReplaceTablePlan>,
+        affected_table_change: Option<ReplaceJobPlan>,
     ) -> Result<()> {
         let version = self
             .meta_client
diff --git a/src/frontend/src/handler/create_sink.rs b/src/frontend/src/handler/create_sink.rs
index 713060e849543..15e4488ddcf8e 100644
--- a/src/frontend/src/handler/create_sink.rs
+++ b/src/frontend/src/handler/create_sink.rs
@@ -37,7 +37,7 @@ use risingwave_connector::sink::{
 use risingwave_connector::WithPropertiesExt;
 use risingwave_pb::catalog::connection_params::PbConnectionType;
 use risingwave_pb::catalog::{PbSink, PbSource, Table};
-use risingwave_pb::ddl_service::{ReplaceTablePlan, TableJobType};
+use risingwave_pb::ddl_service::{replace_job_plan, ReplaceJobPlan, TableJobType};
 use risingwave_pb::stream_plan::stream_node::{NodeBody, PbNodeBody};
 use risingwave_pb::stream_plan::{MergeNode, StreamFragmentGraph, StreamNode};
 use risingwave_pb::telemetry::TelemetryDatabaseObject;
@@ -521,12 +521,16 @@ pub async fn handle_create_sink(
         // for new creating sink, we don't have a unique identity because the sink id is not generated yet.
         hijack_merger_for_target_table(&mut graph, &columns_without_rw_timestamp, &sink, None)?;
 
-        target_table_replace_plan = Some(ReplaceTablePlan {
-            source,
-            table: Some(table),
+        target_table_replace_plan = Some(ReplaceJobPlan {
+            replace_job: Some(replace_job_plan::ReplaceJob::ReplaceTable(
+                replace_job_plan::ReplaceTable {
+                    table: Some(table),
+                    source,
+                    job_type: TableJobType::General as _,
+                },
+            )),
             fragment_graph: Some(graph),
             table_col_index_mapping: None,
-            job_type: TableJobType::General as _,
         });
     }
 
diff --git a/src/frontend/src/handler/drop_sink.rs b/src/frontend/src/handler/drop_sink.rs
index 8e0d015d0f632..ab98894330b69 100644
--- a/src/frontend/src/handler/drop_sink.rs
+++ b/src/frontend/src/handler/drop_sink.rs
@@ -15,7 +15,7 @@
 use std::collections::HashSet;
 
 use pgwire::pg_response::{PgResponse, StatementType};
-use risingwave_pb::ddl_service::{ReplaceTablePlan, TableJobType};
+use risingwave_pb::ddl_service::{replace_job_plan, ReplaceJobPlan, TableJobType};
 use risingwave_sqlparser::ast::ObjectName;
 
 use super::RwPgResponse;
@@ -94,12 +94,16 @@ pub async fn handle_drop_sink(
             )?;
         }
 
-        affected_table_change = Some(ReplaceTablePlan {
-            source,
-            table: Some(table),
+        affected_table_change = Some(ReplaceJobPlan {
+            replace_job: Some(replace_job_plan::ReplaceJob::ReplaceTable(
+                replace_job_plan::ReplaceTable {
+                    table: Some(table),
+                    source,
+                    job_type: TableJobType::General as _,
+                },
+            )),
             fragment_graph: Some(graph),
             table_col_index_mapping: None,
-            job_type: TableJobType::General as _,
         });
     }
 
diff --git a/src/frontend/src/rpc/mod.rs b/src/frontend/src/rpc/mod.rs
index 2f5207ca6e928..69db7e5ae00f5 100644
--- a/src/frontend/src/rpc/mod.rs
+++ b/src/frontend/src/rpc/mod.rs
@@ -14,7 +14,7 @@
 
 use itertools::Itertools;
 use pgwire::pg_server::{BoxedError, SessionManager};
-use risingwave_pb::ddl_service::{ReplaceTablePlan, TableSchemaChange};
+use risingwave_pb::ddl_service::{replace_job_plan, ReplaceJobPlan, TableSchemaChange};
 use risingwave_pb::frontend_service::frontend_service_server::FrontendService;
 use risingwave_pb::frontend_service::{GetTableReplacePlanRequest, GetTableReplacePlanResponse};
 use risingwave_rpc_client::error::ToTonicStatus;
@@ -81,7 +81,7 @@ async fn get_new_table_plan(
     table_name: String,
     database_id: u32,
     owner: u32,
-) -> Result<ReplaceTablePlan, AutoSchemaChangeError> {
+) -> Result<ReplaceJobPlan, AutoSchemaChangeError> {
     tracing::info!("get_new_table_plan for table {}", table_name);
 
     let session_mgr = SESSION_MANAGER
@@ -109,11 +109,15 @@ async fn get_new_table_plan(
     )
     .await?;
 
-    Ok(ReplaceTablePlan {
-        table: Some(table),
+    Ok(ReplaceJobPlan {
+        replace_job: Some(replace_job_plan::ReplaceJob::ReplaceTable(
+            replace_job_plan::ReplaceTable {
+                table: Some(table),
+                source: None, // none for cdc table
+                job_type: job_type as _,
+            },
+        )),
         fragment_graph: Some(graph),
         table_col_index_mapping: Some(col_index_mapping.to_protobuf()),
-        source: None, // none for cdc table
-        job_type: job_type as _,
     })
 }
diff --git a/src/meta/service/src/ddl_service.rs b/src/meta/service/src/ddl_service.rs
index 166bb142bd7f6..11ade22502255 100644
--- a/src/meta/service/src/ddl_service.rs
+++ b/src/meta/service/src/ddl_service.rs
@@ -18,6 +18,7 @@ use std::sync::Arc;
 use anyhow::anyhow;
 use rand::seq::SliceRandom;
 use rand::thread_rng;
+use replace_job_plan::{ReplaceSource, ReplaceTable};
 use risingwave_common::catalog::ColumnCatalog;
 use risingwave_common::types::DataType;
 use risingwave_common::util::column_index_mapping::ColIndexMapping;
@@ -85,25 +86,31 @@ impl DdlServiceImpl {
     }
 
     fn extract_replace_table_info(
-        ReplaceTablePlan {
-            table,
+        ReplaceJobPlan {
             fragment_graph,
             table_col_index_mapping,
-            source,
-            job_type,
-        }: ReplaceTablePlan,
+            replace_job,
+        }: ReplaceJobPlan,
     ) -> ReplaceStreamJobInfo {
-        let table = table.unwrap();
         let col_index_mapping = table_col_index_mapping
             .as_ref()
             .map(ColIndexMapping::from_protobuf);
 
         ReplaceStreamJobInfo {
-            streaming_job: StreamingJob::Table(
-                source,
-                table,
-                TableJobType::try_from(job_type).unwrap(),
-            ),
+            streaming_job: match replace_job.unwrap() {
+                replace_job_plan::ReplaceJob::ReplaceTable(ReplaceTable {
+                    table,
+                    source,
+                    job_type,
+                }) => StreamingJob::Table(
+                    source,
+                    table.unwrap(),
+                    TableJobType::try_from(job_type).unwrap(),
+                ),
+                replace_job_plan::ReplaceJob::ReplaceSource(ReplaceSource { source }) => {
+                    StreamingJob::Source(source.unwrap())
+                }
+            },
             fragment_graph: fragment_graph.unwrap(),
             col_index_mapping,
         }
@@ -303,7 +310,11 @@ impl DdlService for DdlServiceImpl {
 
         let sink = req.get_sink()?.clone();
         let fragment_graph = req.get_fragment_graph()?.clone();
-        let affected_table_change = req.get_affected_table_change().cloned().ok();
+        let affected_table_change = req
+            .get_affected_table_change()
+            .cloned()
+            .ok()
+            .map(Self::extract_replace_table_info);
         let dependencies = req
             .get_dependencies()
             .iter()
@@ -313,8 +324,11 @@ impl DdlService for DdlServiceImpl {
         let stream_job = match &affected_table_change {
             None => StreamingJob::Sink(sink, None),
             Some(change) => {
-                let table = change.table.clone().unwrap();
-                let source = change.source.clone();
+                let (source, table, _) = change
+                    .streaming_job
+                    .clone()
+                    .try_as_table()
+                    .expect("must be replace table");
                 StreamingJob::Sink(sink, Some((table, source)))
             }
         };
@@ -323,7 +337,7 @@ impl DdlService for DdlServiceImpl {
             stream_job,
             fragment_graph,
             CreateType::Foreground,
-            affected_table_change.map(Self::extract_replace_table_info),
+            affected_table_change,
             dependencies,
         );
 
@@ -646,20 +660,20 @@ impl DdlService for DdlServiceImpl {
         Ok(Response::new(RisectlListStateTablesResponse { tables }))
     }
 
-    async fn replace_table_plan(
+    async fn replace_job_plan(
         &self,
-        request: Request<ReplaceTablePlanRequest>,
-    ) -> Result<Response<ReplaceTablePlanResponse>, Status> {
+        request: Request<ReplaceJobPlanRequest>,
+    ) -> Result<Response<ReplaceJobPlanResponse>, Status> {
         let req = request.into_inner().get_plan().cloned()?;
 
         let version = self
             .ddl_controller
-            .run_command(DdlCommand::ReplaceTable(Self::extract_replace_table_info(
-                req,
-            )))
+            .run_command(DdlCommand::ReplaceStreamJob(
+                Self::extract_replace_table_info(req),
+            ))
             .await?;
 
-        Ok(Response::new(ReplaceTablePlanResponse {
+        Ok(Response::new(ReplaceJobPlanResponse {
             status: None,
             version,
         }))
@@ -973,7 +987,7 @@ impl DdlService for DdlServiceImpl {
                     .auto_schema_change_latency
                     .with_guarded_label_values(&[&table.id.to_string(), &table.name])
                     .start_timer();
-                // send a request to the frontend to get the ReplaceTablePlan
+                // send a request to the frontend to get the ReplaceJobPlan
                 // will retry with exponential backoff if the request fails
                 let resp = client
                     .get_table_replace_plan(GetTableReplacePlanRequest {
@@ -988,7 +1002,8 @@ impl DdlService for DdlServiceImpl {
                     Ok(resp) => {
                         let resp = resp.into_inner();
                         if let Some(plan) = resp.replace_plan {
-                            plan.table.as_ref().inspect(|t| {
+                            let plan = Self::extract_replace_table_info(plan);
+                            plan.streaming_job.table().inspect(|t| {
                                 tracing::info!(
                                     target: "auto_schema_change",
                                     table_id = t.id,
@@ -999,9 +1014,7 @@ impl DdlService for DdlServiceImpl {
                             // start the schema change procedure
                             let replace_res = self
                                 .ddl_controller
-                                .run_command(DdlCommand::ReplaceTable(
-                                    Self::extract_replace_table_info(plan),
-                                ))
+                                .run_command(DdlCommand::ReplaceStreamJob(plan))
                                 .await;
 
                             match replace_res {
diff --git a/src/meta/src/barrier/command.rs b/src/meta/src/barrier/command.rs
index d315587546c32..3ded5c438a670 100644
--- a/src/meta/src/barrier/command.rs
+++ b/src/meta/src/barrier/command.rs
@@ -145,6 +145,31 @@ impl ReplaceStreamJobPlan {
         }
         fragment_changes
     }
+
+    /// `old_fragment_id` -> `new_fragment_id`
+    pub fn fragment_replacements(&self) -> HashMap<FragmentId, FragmentId> {
+        let mut fragment_replacements = HashMap::new();
+        for merge_update in &self.merge_updates {
+            if let Some(new_upstream_fragment_id) = merge_update.new_upstream_fragment_id {
+                let r = fragment_replacements
+                    .insert(merge_update.upstream_fragment_id, new_upstream_fragment_id);
+                if let Some(r) = r {
+                    assert_eq!(
+                        new_upstream_fragment_id, r,
+                        "one fragment is replaced by multiple fragments"
+                    );
+                }
+            }
+        }
+        fragment_replacements
+    }
+
+    pub fn dropped_actors(&self) -> HashSet<ActorId> {
+        self.merge_updates
+            .iter()
+            .flat_map(|merge_update| merge_update.removed_upstream_actor_id.clone())
+            .collect()
+    }
 }
 
 #[derive(educe::Educe, Clone)]
diff --git a/src/meta/src/barrier/context/context_impl.rs b/src/meta/src/barrier/context/context_impl.rs
index dba212d04b5c5..8fa96ba128ece 100644
--- a/src/meta/src/barrier/context/context_impl.rs
+++ b/src/meta/src/barrier/context/context_impl.rs
@@ -161,7 +161,7 @@ impl CommandContext {
                     .await?;
                 barrier_manager_context
                     .source_manager
-                    .apply_source_change(None, None, Some(split_assignment.clone()), None)
+                    .apply_source_change(None, None, Some(split_assignment.clone()), None, None)
                     .await;
             }
 
@@ -193,12 +193,16 @@ impl CommandContext {
                     )
                     .await?;
 
-                if let CreateStreamingJobType::SinkIntoTable(ReplaceStreamJobPlan {
-                    new_fragments,
-                    dispatchers,
-                    init_split_assignment,
-                    ..
-                }) = job_type
+                let mut fragment_replacements = None;
+                let mut dropped_actors = None;
+                if let CreateStreamingJobType::SinkIntoTable(
+                    replace_plan @ ReplaceStreamJobPlan {
+                        new_fragments,
+                        dispatchers,
+                        init_split_assignment,
+                        ..
+                    },
+                ) = job_type
                 {
                     barrier_manager_context
                         .metadata_manager
@@ -210,6 +214,8 @@ impl CommandContext {
                             init_split_assignment,
                         )
                         .await?;
+                    fragment_replacements = Some(replace_plan.fragment_replacements());
+                    dropped_actors = Some(replace_plan.dropped_actors());
                 }
 
                 // Extract the fragments that include source operators.
@@ -221,7 +227,8 @@ impl CommandContext {
                         Some(source_fragments),
                         Some(backfill_fragments),
                         Some(init_split_assignment.clone()),
-                        None,
+                        dropped_actors,
+                        fragment_replacements,
                     )
                     .await;
             }
@@ -236,13 +243,15 @@ impl CommandContext {
                     .await?;
             }
 
-            Command::ReplaceStreamJob(ReplaceStreamJobPlan {
-                old_fragments,
-                new_fragments,
-                dispatchers,
-                init_split_assignment,
-                ..
-            }) => {
+            Command::ReplaceStreamJob(
+                replace_plan @ ReplaceStreamJobPlan {
+                    old_fragments,
+                    new_fragments,
+                    dispatchers,
+                    init_split_assignment,
+                    ..
+                },
+            ) => {
                 // Update actors and actor_dispatchers for new table fragments.
                 barrier_manager_context
                     .metadata_manager
@@ -269,7 +278,8 @@ impl CommandContext {
                         Some(source_fragments),
                         Some(backfill_fragments),
                         Some(init_split_assignment.clone()),
-                        None,
+                        Some(replace_plan.dropped_actors()),
+                        Some(replace_plan.fragment_replacements()),
                     )
                     .await;
             }
diff --git a/src/meta/src/controller/fragment.rs b/src/meta/src/controller/fragment.rs
index 73ff00d4c506e..14be4ad5a2883 100644
--- a/src/meta/src/controller/fragment.rs
+++ b/src/meta/src/controller/fragment.rs
@@ -1352,25 +1352,28 @@ impl CatalogController {
     /// Root fragment connects to downstream jobs.
     ///
     /// ## What can be the root fragment
+    /// - For sink, it should have one `Sink` fragment.
     /// - For MV, it should have one `MView` fragment.
     /// - For table, it should have one `MView` fragment and one or two `Source` fragments. `MView` should be the root.
     /// - For source, it should have one `Source` fragment.
     ///
-    /// In other words, it's the `MView` fragment if it exists, otherwise it's the `Source` fragment.
+    /// In other words, it's the `MView` or `Sink` fragment if it exists, otherwise it's the `Source` fragment.
     pub async fn get_root_fragments(
         &self,
         job_ids: Vec<ObjectId>,
     ) -> MetaResult<(HashMap<ObjectId, PbFragment>, Vec<(ActorId, WorkerId)>)> {
         let inner = self.inner.read().await;
 
-        let all_upstream_fragments = Fragment::find()
+        let all_fragments = Fragment::find()
             .filter(fragment::Column::JobId.is_in(job_ids))
             .all(&inner.db)
             .await?;
         // job_id -> fragment
         let mut root_fragments = HashMap::<ObjectId, fragment::Model>::new();
-        for fragment in all_upstream_fragments {
-            if fragment.fragment_type_mask & PbFragmentTypeFlag::Mview as i32 != 0 {
+        for fragment in all_fragments {
+            if (fragment.fragment_type_mask & PbFragmentTypeFlag::Mview as i32) != 0
+                || (fragment.fragment_type_mask & PbFragmentTypeFlag::Sink as i32) != 0
+            {
                 _ = root_fragments.insert(fragment.job_id, fragment);
             } else if fragment.fragment_type_mask & PbFragmentTypeFlag::Source as i32 != 0 {
                 // look for Source fragment only if there's no MView fragment
diff --git a/src/meta/src/controller/streaming_job.rs b/src/meta/src/controller/streaming_job.rs
index 48bd7786b8aaf..88a39a42b9107 100644
--- a/src/meta/src/controller/streaming_job.rs
+++ b/src/meta/src/controller/streaming_job.rs
@@ -974,7 +974,7 @@ impl CatalogController {
         tmp_id: ObjectId,
         streaming_job: StreamingJob,
         merge_updates: Vec<PbMergeUpdate>,
-        table_col_index_mapping: Option<ColIndexMapping>,
+        col_index_mapping: Option<ColIndexMapping>,
         sink_into_table_context: SinkIntoTableContext,
     ) -> MetaResult<NotificationVersion> {
         let inner = self.inner.write().await;
@@ -983,7 +983,7 @@ impl CatalogController {
         let (relations, fragment_mapping) = Self::finish_replace_streaming_job_inner(
             tmp_id,
             merge_updates,
-            table_col_index_mapping,
+            col_index_mapping,
             sink_into_table_context,
             &txn,
             streaming_job,
@@ -1009,12 +1009,10 @@ impl CatalogController {
         Ok(version)
     }
 
-    /// TODO: make it general for other streaming jobs.
-    /// Currently only for replacing table.
     pub async fn finish_replace_streaming_job_inner(
         tmp_id: ObjectId,
         merge_updates: Vec<PbMergeUpdate>,
-        table_col_index_mapping: Option<ColIndexMapping>,
+        col_index_mapping: Option<ColIndexMapping>,
         SinkIntoTableContext {
             creating_sink_id,
             dropping_sink_id,
@@ -1065,7 +1063,11 @@ impl CatalogController {
                 table.incoming_sinks = Set(incoming_sinks.into());
                 table.update(txn).await?;
             }
-            // TODO: support other streaming jobs
+            StreamingJob::Source(source) => {
+                // Update the source catalog with the new one.
+                let source = source::ActiveModel::from(source);
+                source.update(txn).await?;
+            }
             _ => unreachable!(
                 "invalid streaming job type: {:?}",
                 streaming_job.job_type_str()
@@ -1223,9 +1225,21 @@ impl CatalogController {
                     )),
                 })
             }
-            _ => unreachable!("invalid streaming job type: {:?}", job_type),
+            StreamingJobType::Source => {
+                let (source, source_obj) = Source::find_by_id(original_job_id)
+                    .find_also_related(Object)
+                    .one(txn)
+                    .await?
+                    .ok_or_else(|| MetaError::catalog_id_not_found("object", original_job_id))?;
+                relations.push(PbRelation {
+                    relation_info: Some(PbRelationInfo::Source(
+                        ObjectModel(source, source_obj.unwrap()).into(),
+                    )),
+                })
+            }
+            _ => unreachable!("invalid streaming job type for replace: {:?}", job_type),
         }
-        if let Some(table_col_index_mapping) = table_col_index_mapping {
+        if let Some(table_col_index_mapping) = col_index_mapping {
             let expr_rewriter = ReplaceTableExprRewriter {
                 table_col_index_mapping,
             };
diff --git a/src/meta/src/manager/metadata.rs b/src/meta/src/manager/metadata.rs
index 1431e204899ba..c31a47a17d4df 100644
--- a/src/meta/src/manager/metadata.rs
+++ b/src/meta/src/manager/metadata.rs
@@ -409,17 +409,7 @@ impl MetadataManager {
     /// Get and filter the "**root**" fragments of the specified relations.
     /// The root fragment is the bottom-most fragment of its fragment graph, and can be a `MView` or a `Source`.
     ///
-    /// ## What can be the root fragment
-    /// - For MV, it should have one `MView` fragment.
-    /// - For table, it should have one `MView` fragment and one or two `Source` fragments. `MView` should be the root.
-    /// - For source, it should have one `Source` fragment.
-    ///
-    /// In other words, it's the `MView` fragment if it exists, otherwise it's the `Source` fragment.
-    ///
-    /// ## What do we expect to get for different creating streaming job
-    /// - MV/Sink/Index should have MV upstream fragments for upstream MV/Tables, and Source upstream fragments for upstream shared sources.
-    /// - CDC Table has a Source upstream fragment.
-    /// - Sources and other Tables shouldn't have an upstream fragment.
+    /// See also [`crate::controller::catalog::CatalogController::get_root_fragments`].
     pub async fn get_upstream_root_fragments(
         &self,
         upstream_table_ids: &HashSet<TableId>,
diff --git a/src/meta/src/manager/streaming_job.rs b/src/meta/src/manager/streaming_job.rs
index b91c3f5e9bd01..727a526896e8e 100644
--- a/src/meta/src/manager/streaming_job.rs
+++ b/src/meta/src/manager/streaming_job.rs
@@ -24,7 +24,7 @@ use risingwave_pb::catalog::{CreateType, Index, PbSource, Sink, Table};
 use risingwave_pb::ddl_service::TableJobType;
 use sea_orm::entity::prelude::*;
 use sea_orm::{DatabaseTransaction, QuerySelect};
-use strum::EnumIs;
+use strum::{EnumIs, EnumTryAs};
 
 use super::{
     get_referred_connection_ids_from_sink, get_referred_connection_ids_from_source,
@@ -35,7 +35,7 @@ use crate::{MetaError, MetaResult};
 
 // This enum is used in order to re-use code in `DdlServiceImpl` for creating MaterializedView and
 // Sink.
-#[derive(Debug, Clone, EnumIs)]
+#[derive(Debug, Clone, EnumIs, EnumTryAs)]
 pub enum StreamingJob {
     MaterializedView(Table),
     Sink(Sink, Option<(Table, Option<PbSource>)>),
diff --git a/src/meta/src/model/stream.rs b/src/meta/src/model/stream.rs
index dd5aa72a16c32..3bfbfc7a03d9f 100644
--- a/src/meta/src/model/stream.rs
+++ b/src/meta/src/model/stream.rs
@@ -369,6 +369,12 @@ impl StreamJobFragments {
         actor_ids
     }
 
+    pub fn root_fragment(&self) -> Option<Fragment> {
+        self.mview_fragment()
+            .or_else(|| self.sink_fragment())
+            .or_else(|| self.source_fragment())
+    }
+
     /// Returns the fragment with the `Mview` type flag.
     pub fn mview_fragment(&self) -> Option<Fragment> {
         self.fragments
diff --git a/src/meta/src/rpc/ddl_controller.rs b/src/meta/src/rpc/ddl_controller.rs
index 6830c97f1e82e..b3c8b47b29af2 100644
--- a/src/meta/src/rpc/ddl_controller.rs
+++ b/src/meta/src/rpc/ddl_controller.rs
@@ -145,7 +145,7 @@ pub enum DdlCommand {
     DropStreamingJob(StreamingJobId, DropMode, Option<ReplaceStreamJobInfo>),
     AlterName(alter_name_request::Object, String),
     AlterSwapRename(alter_swap_rename_request::Object),
-    ReplaceTable(ReplaceStreamJobInfo),
+    ReplaceStreamJob(ReplaceStreamJobInfo),
     AlterNonSharedSource(Source),
     AlterObjectOwner(Object, UserId),
     AlterSetSchema(alter_set_schema_request::Object, SchemaId),
@@ -185,7 +185,7 @@ impl DdlCommand {
             | DdlCommand::AlterSwapRename(_) => true,
             DdlCommand::CreateStreamingJob(_, _, _, _, _)
             | DdlCommand::CreateNonSharedSource(_)
-            | DdlCommand::ReplaceTable(_)
+            | DdlCommand::ReplaceStreamJob(_)
             | DdlCommand::AlterNonSharedSource(_)
             | DdlCommand::CreateSubscription(_) => false,
         }
@@ -330,7 +330,7 @@ impl DdlController {
                     ctrl.drop_streaming_job(job_id, drop_mode, target_replace_info)
                         .await
                 }
-                DdlCommand::ReplaceTable(ReplaceStreamJobInfo {
+                DdlCommand::ReplaceStreamJob(ReplaceStreamJobInfo {
                     streaming_job,
                     fragment_graph,
                     col_index_mapping,
@@ -354,7 +354,9 @@ impl DdlController {
                 DdlCommand::CreateSecret(secret) => ctrl.create_secret(secret).await,
                 DdlCommand::DropSecret(secret_id) => ctrl.drop_secret(secret_id).await,
                 DdlCommand::AlterSecret(secret) => ctrl.alter_secret(secret).await,
-                DdlCommand::AlterNonSharedSource(source) => ctrl.alter_source(source).await,
+                DdlCommand::AlterNonSharedSource(source) => {
+                    ctrl.alter_non_shared_source(source).await
+                }
                 DdlCommand::CommentOn(comment) => ctrl.comment_on(comment).await,
                 DdlCommand::CreateSubscription(subscription) => {
                     ctrl.create_subscription(subscription).await
@@ -464,7 +466,7 @@ impl DdlController {
 
     /// This replaces the source in the catalog.
     /// Note: `StreamSourceInfo` in downstream MVs' `SourceExecutor`s are not updated.
-    async fn alter_source(&self, source: Source) -> MetaResult<NotificationVersion> {
+    async fn alter_non_shared_source(&self, source: Source) -> MetaResult<NotificationVersion> {
         self.metadata_manager
             .catalog_controller
             .alter_non_shared_source(source)
@@ -1299,7 +1301,7 @@ impl DdlController {
         &self,
         mut streaming_job: StreamingJob,
         fragment_graph: StreamFragmentGraphProto,
-        table_col_index_mapping: Option<ColIndexMapping>,
+        col_index_mapping: Option<ColIndexMapping>,
     ) -> MetaResult<NotificationVersion> {
         match &mut streaming_job {
             StreamingJob::Table(..) | StreamingJob::Source(..) => {}
@@ -1352,7 +1354,7 @@ impl DdlController {
                     ctx,
                     &streaming_job,
                     fragment_graph,
-                    table_col_index_mapping.clone(),
+                    col_index_mapping.as_ref(),
                     tmp_id as _,
                 )
                 .await?;
@@ -1412,7 +1414,7 @@ impl DdlController {
                         tmp_id,
                         streaming_job,
                         merge_updates,
-                        table_col_index_mapping,
+                        col_index_mapping,
                         SinkIntoTableContext {
                             creating_sink_id: None,
                             dropping_sink_id: None,
@@ -1690,14 +1692,12 @@ impl DdlController {
         stream_ctx: StreamContext,
         stream_job: &StreamingJob,
         mut fragment_graph: StreamFragmentGraph,
-        // TODO(alter-source): check what does this mean
-        table_col_index_mapping: Option<ColIndexMapping>,
+        col_index_mapping: Option<&ColIndexMapping>,
         tmp_job_id: TableId,
     ) -> MetaResult<(ReplaceStreamJobContext, StreamJobFragments)> {
         match &stream_job {
-            StreamingJob::Table(..) => {}
-            StreamingJob::Source(..)
-            | StreamingJob::MaterializedView(..)
+            StreamingJob::Table(..) | StreamingJob::Source(..) => {}
+            StreamingJob::MaterializedView(..)
             | StreamingJob::Sink(..)
             | StreamingJob::Index(..) => {
                 bail_not_implemented!("schema change for {}", stream_job.job_type_str())
@@ -1721,16 +1721,16 @@ impl DdlController {
 
         // 1. Resolve the edges to the downstream fragments, extend the fragment graph to a complete
         // graph that contains all information needed for building the actor graph.
-        let original_mview_fragment = old_fragments
-            .mview_fragment()
-            .expect("mview fragment not found");
+        let original_root_fragment = old_fragments
+            .root_fragment()
+            .expect("root fragment not found");
 
         let job_type = StreamingJobType::from(stream_job);
 
         // Map the column indices in the dispatchers with the given mapping.
         let (mut downstream_fragments, downstream_actor_location) =
             self.metadata_manager.get_downstream_fragments(id).await?;
-        if let Some(mapping) = &table_col_index_mapping {
+        if let Some(mapping) = &col_index_mapping {
             for (d, _f) in &mut downstream_fragments {
                 *d = mapping.rewrite_dispatch_strategy(d).ok_or_else(|| {
                     // The `rewrite` only fails if some column is dropped.
@@ -1742,11 +1742,11 @@ impl DdlController {
         }
 
         // build complete graph based on the table job type
-        let complete_graph = match job_type {
-            StreamingJobType::Table(TableJobType::General) => {
+        let complete_graph = match &job_type {
+            StreamingJobType::Table(TableJobType::General) | StreamingJobType::Source => {
                 CompleteStreamFragmentGraph::with_downstreams(
                     fragment_graph,
-                    original_mview_fragment.fragment_id,
+                    original_root_fragment.fragment_id,
                     downstream_fragments,
                     downstream_actor_location,
                     job_type,
@@ -1763,7 +1763,7 @@ impl DdlController {
                     fragment_graph,
                     upstream_root_fragments,
                     upstream_actor_location,
-                    original_mview_fragment.fragment_id,
+                    original_root_fragment.fragment_id,
                     downstream_fragments,
                     downstream_actor_location,
                     job_type,
@@ -1777,7 +1777,7 @@ impl DdlController {
 
         // XXX: what is this parallelism?
         // Is it "assigned parallelism"?
-        let parallelism = NonZeroUsize::new(original_mview_fragment.get_actors().len())
+        let parallelism = NonZeroUsize::new(original_root_fragment.get_actors().len())
             .expect("The number of actors in the original table fragment should be greater than 0");
 
         let actor_graph_builder =
@@ -1791,8 +1791,11 @@ impl DdlController {
             merge_updates,
         } = actor_graph_builder.generate_graph(&self.env, stream_job, expr_context)?;
 
-        // general table job type does not have upstream job, so the dispatchers should be empty
-        if matches!(job_type, StreamingJobType::Table(TableJobType::General)) {
+        // general table & source does not have upstream job, so the dispatchers should be empty
+        if matches!(
+            job_type,
+            StreamingJobType::Source | StreamingJobType::Table(TableJobType::General)
+        ) {
             assert!(dispatchers.is_empty());
         }
 
diff --git a/src/meta/src/stream/scale.rs b/src/meta/src/stream/scale.rs
index 00ea06341b42f..d40d6c9d66cca 100644
--- a/src/meta/src/stream/scale.rs
+++ b/src/meta/src/stream/scale.rs
@@ -1774,6 +1774,7 @@ impl ScaleController {
                     None,
                     Some(stream_source_actor_splits),
                     Some(stream_source_dropped_actors),
+                    None,
                 )
                 .await;
         }
diff --git a/src/meta/src/stream/source_manager.rs b/src/meta/src/stream/source_manager.rs
index de357069da30c..07199e07c5f80 100644
--- a/src/meta/src/stream/source_manager.rs
+++ b/src/meta/src/stream/source_manager.rs
@@ -32,6 +32,7 @@ use risingwave_connector::{dispatch_source_prop, WithOptionsSecResolved};
 use risingwave_meta_model::SourceId;
 use risingwave_pb::catalog::Source;
 use risingwave_pb::source::{ConnectorSplit, ConnectorSplits};
+use risingwave_pb::stream_plan::update_mutation::MergeUpdate;
 use risingwave_pb::stream_plan::Dispatcher;
 use thiserror_ext::AsReport;
 use tokio::sync::mpsc::{UnboundedReceiver, UnboundedSender};
@@ -441,7 +442,7 @@ impl SourceManagerCore {
                     };
                     split_assignment.insert(
                         *fragment_id,
-                        align_backfill_splits(
+                        align_splits(
                             actors,
                             upstream_assignment,
                             *fragment_id,
@@ -463,6 +464,7 @@ impl SourceManagerCore {
         added_backfill_fragments: Option<HashMap<SourceId, BTreeSet<(FragmentId, FragmentId)>>>,
         split_assignment: Option<SplitAssignment>,
         dropped_actors: Option<HashSet<ActorId>>,
+        fragment_replacements: Option<HashMap<FragmentId, FragmentId>>,
     ) {
         if let Some(source_fragments) = added_source_fragments {
             for (source_id, mut fragment_ids) in source_fragments {
@@ -495,6 +497,32 @@ impl SourceManagerCore {
                 self.actor_splits.remove(actor_id);
             }
         }
+
+        if let Some(fragment_replacements) = fragment_replacements {
+            for (old_fragment_id, new_fragment_id) in fragment_replacements {
+                for fragment_ids in self.source_fragments.values_mut() {
+                    if fragment_ids.remove(&old_fragment_id) {
+                        fragment_ids.insert(new_fragment_id);
+                    }
+                }
+
+                for fragment_ids in self.backfill_fragments.values_mut() {
+                    let mut new_backfill_fragment_ids = fragment_ids.clone();
+                    for (fragment_id, upstream_fragment_id) in fragment_ids.iter() {
+                        assert_ne!(
+                            fragment_id, upstream_fragment_id,
+                            "backfill fragment should not be replaced"
+                        );
+                        if *upstream_fragment_id == old_fragment_id {
+                            new_backfill_fragment_ids
+                                .remove(&(*fragment_id, *upstream_fragment_id));
+                            new_backfill_fragment_ids.insert((*fragment_id, new_fragment_id));
+                        }
+                    }
+                    *fragment_ids = new_backfill_fragment_ids;
+                }
+            }
+        }
     }
 
     fn drop_source_fragments(
@@ -574,7 +602,7 @@ impl Default for SplitDiffOptions {
 ///
 /// If an actor has an upstream actor, it should be a backfill executor,
 /// and its splits should be aligned with the upstream actor. **`reassign_splits` should not be used in this case.
-/// Use `align_backfill_splits` instead.**
+/// Use [`align_splits`] instead.**
 ///
 /// - `fragment_id`: just for logging
 ///
@@ -688,18 +716,27 @@ where
     )
 }
 
-fn align_backfill_splits(
-    backfill_actors: impl IntoIterator<Item = (ActorId, ActorId)>,
-    upstream_assignment: &HashMap<ActorId, Vec<SplitImpl>>,
+/// Assign splits to a new set of actors, according to existing assignment.
+///
+/// illustration:
+/// ```text
+/// upstream                               new
+/// actor x1 [split 1, split2]      ->     actor y1 [split 1, split2]
+/// actor x2 [split 3]              ->     actor y2 [split 3]
+/// ...
+/// ```
+fn align_splits(
+    // (actor_id, upstream_actor_id)
+    aligned_actors: impl IntoIterator<Item = (ActorId, ActorId)>,
+    existing_assignment: &HashMap<ActorId, Vec<SplitImpl>>,
     fragment_id: FragmentId,
     upstream_source_fragment_id: FragmentId,
 ) -> anyhow::Result<HashMap<ActorId, Vec<SplitImpl>>> {
-    backfill_actors
+    aligned_actors
         .into_iter()
         .map(|(actor_id, upstream_actor_id)| {
-            let err = || anyhow::anyhow!("source backfill actor should have one upstream source actor, fragment_id: {fragment_id}, upstream_fragment_id: {upstream_source_fragment_id}, actor_id: {actor_id}, upstream_assignment: {upstream_assignment:?}, upstream_actor_id: {upstream_actor_id:?}");
-            let Some(splits) = upstream_assignment.get(&upstream_actor_id) else {
-                return Err(err());
+            let Some(splits) = existing_assignment.get(&upstream_actor_id) else {
+                return Err(anyhow::anyhow!("upstream assignment not found, fragment_id: {fragment_id}, upstream_fragment_id: {upstream_source_fragment_id}, actor_id: {actor_id}, upstream_assignment: {existing_assignment:?}, upstream_actor_id: {upstream_actor_id:?}"));
             };
             Ok((
                 actor_id,
@@ -827,6 +864,7 @@ impl SourceManager {
         added_backfill_fragments: Option<HashMap<SourceId, BTreeSet<(FragmentId, FragmentId)>>>,
         split_assignment: Option<SplitAssignment>,
         dropped_actors: Option<HashSet<ActorId>>,
+        fragment_replacements: Option<HashMap<FragmentId, FragmentId>>,
     ) {
         let mut core = self.core.lock().await;
         core.apply_source_change(
@@ -834,6 +872,7 @@ impl SourceManager {
             added_backfill_fragments,
             split_assignment,
             dropped_actors,
+            fragment_replacements,
         );
     }
 
@@ -903,7 +942,7 @@ impl SourceManager {
             ?upstream_assignment,
             "migrate_splits_for_backfill_actors"
         );
-        Ok(align_backfill_splits(
+        Ok(align_splits(
             actors,
             upstream_assignment,
             fragment_id,
@@ -972,13 +1011,89 @@ impl SourceManager {
         Ok(assigned)
     }
 
+    /// Allocates splits to actors for replace source job.
+    pub async fn allocate_splits_for_replace_source(
+        &self,
+        job_id: &TableId,
+        merge_updates: &Vec<MergeUpdate>,
+    ) -> MetaResult<SplitAssignment> {
+        tracing::debug!(?merge_updates, "allocate_splits_for_replace_source");
+        if merge_updates.is_empty() {
+            // no existing downstream. We can just re-allocate splits arbitrarily.
+            return self.allocate_splits(job_id).await;
+        }
+
+        let core = self.core.lock().await;
+        let table_fragments = core
+            .metadata_manager
+            .get_job_fragments_by_id(job_id)
+            .await?;
+
+        let source_fragments = table_fragments.stream_source_fragments();
+        assert_eq!(
+            source_fragments.len(),
+            1,
+            "replace source job should only have one source"
+        );
+        let (_source_id, fragments) = source_fragments.into_iter().next().unwrap();
+        assert_eq!(
+            fragments.len(),
+            1,
+            "replace source job should only have one fragment"
+        );
+        let fragment_id = fragments.into_iter().next().unwrap();
+
+        debug_assert!(
+            !merge_updates.is_empty()
+                && merge_updates.iter().all(|merge_update| {
+                    merge_update.upstream_fragment_id == merge_updates[0].upstream_fragment_id
+                        && merge_update.new_upstream_fragment_id == Some(fragment_id)
+                }),
+            "merge update should only replace one fragment: {:?}",
+            merge_updates
+        );
+        let prev_fragment_id = merge_updates[0].upstream_fragment_id;
+        // Here we align the new source executor to backfill executors
+        //
+        // old_source => new_source            backfill_1
+        // actor_x1   => actor_y1 -----┬------>actor_a1
+        // actor_x2   => actor_y2 -----┼-┬---->actor_a2
+        //                             │ │
+        //                             │ │     backfill_2
+        //                             └─┼---->actor_b1
+        //                               └---->actor_b2
+        //
+        // Note: we can choose any backfill actor to align here.
+        // We use `HashMap` to dedup.
+        let aligned_actors: HashMap<ActorId, ActorId> = merge_updates
+            .iter()
+            .map(|merge_update| {
+                assert_eq!(merge_update.added_upstream_actor_id.len(), 1);
+                // Note: removed_upstream_actor_id is not set for replace job, so we can't use it.
+                assert_eq!(merge_update.removed_upstream_actor_id.len(), 0);
+                (
+                    merge_update.added_upstream_actor_id[0],
+                    merge_update.actor_id,
+                )
+            })
+            .collect();
+        let assignment = align_splits(
+            aligned_actors.into_iter(),
+            &core.actor_splits,
+            fragment_id,
+            prev_fragment_id,
+        )?;
+        Ok(HashMap::from([(fragment_id, assignment)]))
+    }
+
     /// Allocates splits to actors for a newly created `SourceBackfill` executor.
     ///
     /// Unlike [`Self::allocate_splits`], which creates a new assignment,
-    /// this method aligns the splits for backfill fragments with its upstream source fragment ([`align_backfill_splits`]).
+    /// this method aligns the splits for backfill fragments with its upstream source fragment ([`align_splits`]).
     pub async fn allocate_splits_for_backfill(
         &self,
         table_id: &TableId,
+        // dispatchers from SourceExecutor to SourceBackfillExecutor
         dispatchers: &HashMap<ActorId, Vec<Dispatcher>>,
     ) -> MetaResult<SplitAssignment> {
         let core = self.core.lock().await;
@@ -987,7 +1102,6 @@ impl SourceManager {
             .get_job_fragments_by_id(table_id)
             .await?;
 
-        let upstream_assignment = &core.actor_splits;
         let source_backfill_fragments = table_fragments.source_backfill_fragments()?;
 
         let mut assigned = HashMap::new();
@@ -1020,9 +1134,9 @@ impl SourceManager {
                 }
                 assigned.insert(
                     fragment_id,
-                    align_backfill_splits(
+                    align_splits(
                         backfill_actors,
-                        upstream_assignment,
+                        &core.actor_splits,
                         fragment_id,
                         upstream_source_fragment_id,
                     )?,
diff --git a/src/meta/src/stream/stream_graph/fragment.rs b/src/meta/src/stream/stream_graph/fragment.rs
index ab848c2b365b9..21103fe1460c1 100644
--- a/src/meta/src/stream/stream_graph/fragment.rs
+++ b/src/meta/src/stream/stream_graph/fragment.rs
@@ -689,7 +689,7 @@ pub struct FragmentGraphUpstreamContext {
 }
 
 pub struct FragmentGraphDownstreamContext {
-    original_table_fragment_id: FragmentId,
+    original_root_fragment_id: FragmentId,
     downstream_fragments: Vec<(DispatchStrategy, Fragment)>,
     downstream_actor_location: HashMap<ActorId, WorkerId>,
 }
@@ -732,7 +732,7 @@ impl CompleteStreamFragmentGraph {
     /// with the downstream existing `StreamScan`/`StreamSourceScan` fragments.
     pub fn with_downstreams(
         graph: StreamFragmentGraph,
-        original_table_fragment_id: FragmentId,
+        original_root_fragment_id: FragmentId,
         downstream_fragments: Vec<(DispatchStrategy, Fragment)>,
         existing_actor_location: HashMap<ActorId, WorkerId>,
         job_type: StreamingJobType,
@@ -741,7 +741,7 @@ impl CompleteStreamFragmentGraph {
             graph,
             None,
             Some(FragmentGraphDownstreamContext {
-                original_table_fragment_id,
+                original_root_fragment_id,
                 downstream_fragments,
                 downstream_actor_location: existing_actor_location,
             }),
@@ -754,7 +754,7 @@ impl CompleteStreamFragmentGraph {
         graph: StreamFragmentGraph,
         upstream_root_fragments: HashMap<TableId, Fragment>,
         upstream_actor_location: HashMap<ActorId, WorkerId>,
-        original_table_fragment_id: FragmentId,
+        original_root_fragment_id: FragmentId,
         downstream_fragments: Vec<(DispatchStrategy, Fragment)>,
         downstream_actor_location: HashMap<ActorId, WorkerId>,
         job_type: StreamingJobType,
@@ -766,7 +766,7 @@ impl CompleteStreamFragmentGraph {
                 upstream_actor_location,
             }),
             Some(FragmentGraphDownstreamContext {
-                original_table_fragment_id,
+                original_root_fragment_id,
                 downstream_fragments,
                 downstream_actor_location,
             }),
@@ -959,12 +959,12 @@ impl CompleteStreamFragmentGraph {
         }
 
         if let Some(FragmentGraphDownstreamContext {
-            original_table_fragment_id,
+            original_root_fragment_id,
             downstream_fragments,
             downstream_actor_location,
         }) = downstream_ctx
         {
-            let original_table_fragment_id = GlobalFragmentId::new(original_table_fragment_id);
+            let original_table_fragment_id = GlobalFragmentId::new(original_root_fragment_id);
             let table_fragment_id = GlobalFragmentId::new(graph.table_fragment_id());
 
             // Build the extra edges between the `Materialize` and the downstream `StreamScan` of the
diff --git a/src/meta/src/stream/stream_manager.rs b/src/meta/src/stream/stream_manager.rs
index 5f52e1a79a708..dbe47567fcaef 100644
--- a/src/meta/src/stream/stream_manager.rs
+++ b/src/meta/src/stream/stream_manager.rs
@@ -476,7 +476,13 @@ impl GlobalStreamManager {
         }: ReplaceStreamJobContext,
     ) -> MetaResult<()> {
         let tmp_table_id = new_fragments.stream_job_id();
-        let init_split_assignment = self.source_manager.allocate_splits(&tmp_table_id).await?;
+        let init_split_assignment = if streaming_job.is_source() {
+            self.source_manager
+                .allocate_splits_for_replace_source(&tmp_table_id, &merge_updates)
+                .await?
+        } else {
+            self.source_manager.allocate_splits(&tmp_table_id).await?
+        };
 
         self.barrier_scheduler
             .run_config_change_command_with_pause(
diff --git a/src/rpc_client/src/meta_client.rs b/src/rpc_client/src/meta_client.rs
index edf83355651aa..f8e45f69cd982 100644
--- a/src/rpc_client/src/meta_client.rs
+++ b/src/rpc_client/src/meta_client.rs
@@ -27,6 +27,7 @@ use either::Either;
 use futures::stream::BoxStream;
 use list_rate_limits_response::RateLimitInfo;
 use lru::LruCache;
+use replace_job_plan::{ReplaceJob, ReplaceTable};
 use risingwave_common::catalog::{FunctionId, IndexId, ObjectId, SecretId, TableId};
 use risingwave_common::config::{MetaConfig, MAX_CONNECTION_WINDOW_SIZE};
 use risingwave_common::hash::WorkerSlotMapping;
@@ -443,7 +444,7 @@ impl MetaClient {
         &self,
         sink: PbSink,
         graph: StreamFragmentGraph,
-        affected_table_change: Option<ReplaceTablePlan>,
+        affected_table_change: Option<ReplaceJobPlan>,
         dependencies: HashSet<ObjectId>,
     ) -> Result<WaitVersion> {
         let request = CreateSinkRequest {
@@ -621,16 +622,18 @@ impl MetaClient {
         table_col_index_mapping: ColIndexMapping,
         job_type: PbTableJobType,
     ) -> Result<WaitVersion> {
-        let request = ReplaceTablePlanRequest {
-            plan: Some(ReplaceTablePlan {
-                source,
-                table: Some(table),
+        let request = ReplaceJobPlanRequest {
+            plan: Some(ReplaceJobPlan {
                 fragment_graph: Some(graph),
                 table_col_index_mapping: Some(table_col_index_mapping.to_protobuf()),
-                job_type: job_type as _,
+                replace_job: Some(ReplaceJob::ReplaceTable(ReplaceTable {
+                    source,
+                    table: Some(table),
+                    job_type: job_type as _,
+                })),
             }),
         };
-        let resp = self.inner.replace_table_plan(request).await?;
+        let resp = self.inner.replace_job_plan(request).await?;
         // TODO: handle error in `resp.status` here
         Ok(resp
             .version
@@ -710,7 +713,7 @@ impl MetaClient {
         &self,
         sink_id: u32,
         cascade: bool,
-        affected_table_change: Option<ReplaceTablePlan>,
+        affected_table_change: Option<ReplaceJobPlan>,
     ) -> Result<WaitVersion> {
         let request = DropSinkRequest {
             sink_id,
@@ -2113,7 +2116,7 @@ macro_rules! for_all_meta_rpc {
             ,{ ddl_client, drop_schema, DropSchemaRequest, DropSchemaResponse }
             ,{ ddl_client, drop_index, DropIndexRequest, DropIndexResponse }
             ,{ ddl_client, drop_function, DropFunctionRequest, DropFunctionResponse }
-            ,{ ddl_client, replace_table_plan, ReplaceTablePlanRequest, ReplaceTablePlanResponse }
+            ,{ ddl_client, replace_job_plan, ReplaceJobPlanRequest, ReplaceJobPlanResponse }
             ,{ ddl_client, alter_source, AlterSourceRequest, AlterSourceResponse }
             ,{ ddl_client, risectl_list_state_tables, RisectlListStateTablesRequest, RisectlListStateTablesResponse }
             ,{ ddl_client, get_ddl_progress, GetDdlProgressRequest, GetDdlProgressResponse }


Test patch:
diff --git a/src/frontend/src/test_utils.rs b/src/frontend/src/test_utils.rs
index 23ae1c8cfa0cc..fc36500415815 100644
--- a/src/frontend/src/test_utils.rs
+++ b/src/frontend/src/test_utils.rs
@@ -46,7 +46,7 @@ use risingwave_pb::common::WorkerNode;
 use risingwave_pb::ddl_service::alter_owner_request::Object;
 use risingwave_pb::ddl_service::{
     alter_name_request, alter_set_schema_request, alter_swap_rename_request,
-    create_connection_request, DdlProgress, PbTableJobType, ReplaceTablePlan, TableJobType,
+    create_connection_request, DdlProgress, PbTableJobType, ReplaceJobPlan, TableJobType,
 };
 use risingwave_pb::hummock::write_limits::WriteLimit;
 use risingwave_pb::hummock::{
@@ -343,7 +343,7 @@ impl CatalogWriter for MockCatalogWriter {
         &self,
         sink: PbSink,
         graph: StreamFragmentGraph,
-        _affected_table_change: Option<ReplaceTablePlan>,
+        _affected_table_change: Option<ReplaceJobPlan>,
         _dependencies: HashSet<ObjectId>,
     ) -> Result<()> {
         self.create_sink_inner(sink, graph)
@@ -485,7 +485,7 @@ impl CatalogWriter for MockCatalogWriter {
         &self,
         sink_id: u32,
         cascade: bool,
-        _target_table_change: Option<ReplaceTablePlan>,
+        _target_table_change: Option<ReplaceJobPlan>,
     ) -> Result<()> {
         if cascade {
             return Err(ErrorCode::NotSupported(