Repository: risingwavelabs/risingwave (language: rust)

Issue:
Support ALTER SOURCE SET PARALLELISM
part of #16003 


We've supported scaling in https://github.com/risingwavelabs/risingwave/pull/16825, but it can only be scaled under adaptive parallelism, since we lack a command for `ALTER SOURCE SET PARALLELISM` now.


Discussion / hints:
This issue has been open for 60 days with no activity.

If you think it is still relevant today, and needs to be done *in the near future*, you can comment to update the status, or just manually remove the `no-issue-activity` label.

You can also confidently close this issue as not planned to keep our backlog clean.
Don't worry if you think the issue is still valuable to continue in the future.
It's searchable and can be reopened when it's time. 😄

Fix patch:
diff --git a/proto/ddl_service.proto b/proto/ddl_service.proto
index 3351de5448b2c..0c4740cafef19 100644
--- a/proto/ddl_service.proto
+++ b/proto/ddl_service.proto
@@ -267,6 +267,7 @@ message AlterSetSchemaResponse {
 }
 
 message AlterParallelismRequest {
+  // job id
   uint32 table_id = 1;
   meta.TableParallelism parallelism = 2;
   bool deferred = 3;
diff --git a/src/ctl/src/cmd_impl/meta/cluster_info.rs b/src/ctl/src/cmd_impl/meta/cluster_info.rs
index 7c8ece6459a2c..737e5a851915e 100644
--- a/src/ctl/src/cmd_impl/meta/cluster_info.rs
+++ b/src/ctl/src/cmd_impl/meta/cluster_info.rs
@@ -86,7 +86,11 @@ pub async fn source_split_info(context: &CtlContext, ignore_id: bool) -> anyhow:
         } else {
             println!("Table #{}", table_fragment.table_id);
         }
-        for fragment in table_fragment.fragments.values() {
+        for fragment in table_fragment
+            .fragments
+            .values()
+            .sorted_by_key(|f| f.fragment_id)
+        {
             let fragment_type_mask = fragment.fragment_type_mask;
             if fragment_type_mask & FragmentTypeFlag::Source as u32 == 0
                 && fragment_type_mask & FragmentTypeFlag::SourceScan as u32 == 0
@@ -112,7 +116,7 @@ pub async fn source_split_info(context: &CtlContext, ignore_id: bool) -> anyhow:
                     "SourceScan"
                 }
             );
-            for actor in &fragment.actors {
+            for actor in fragment.actors.iter().sorted_by_key(|a| a.actor_id) {
                 if let Some((split_count, splits)) = actor_splits_map.get(&actor.actor_id) {
                     println!(
                         "\t\tActor{} ({} splits): [{}]",
diff --git a/src/frontend/src/catalog/catalog_service.rs b/src/frontend/src/catalog/catalog_service.rs
index 583eaa5c1ef2d..3d29a130274db 100644
--- a/src/frontend/src/catalog/catalog_service.rs
+++ b/src/frontend/src/catalog/catalog_service.rs
@@ -215,7 +215,7 @@ pub trait CatalogWriter: Send + Sync {
 
     async fn alter_parallelism(
         &self,
-        table_id: u32,
+        job_id: u32,
         parallelism: PbTableParallelism,
         deferred: bool,
     ) -> Result<()>;
@@ -560,12 +560,12 @@ impl CatalogWriter for CatalogWriterImpl {
 
     async fn alter_parallelism(
         &self,
-        table_id: u32,
+        job_id: u32,
         parallelism: PbTableParallelism,
         deferred: bool,
     ) -> Result<()> {
         self.meta_client
-            .alter_parallelism(table_id, parallelism, deferred)
+            .alter_parallelism(job_id, parallelism, deferred)
             .await
             .map_err(|e| anyhow!(e))?;
 
diff --git a/src/frontend/src/handler/alter_parallelism.rs b/src/frontend/src/handler/alter_parallelism.rs
index 4fd76b93b3a1a..ed64d811013f3 100644
--- a/src/frontend/src/handler/alter_parallelism.rs
+++ b/src/frontend/src/handler/alter_parallelism.rs
@@ -44,7 +44,7 @@ pub async fn handle_alter_parallelism(
     let user_name = &session.user_name();
     let schema_path = SchemaPath::new(schema_name.as_deref(), &search_path, user_name);
 
-    let table_id = {
+    let job_id = {
         let reader = session.env().catalog_reader().read_guard();
 
         match stmt_type {
@@ -76,6 +76,21 @@ pub async fn handle_alter_parallelism(
                 session.check_privilege_for_drop_alter(schema_name, &**table)?;
                 table.id.table_id()
             }
+            StatementType::ALTER_SOURCE => {
+                let (source, schema_name) =
+                    reader.get_source_by_name(db_name, schema_path, &real_table_name)?;
+
+                if !source.info.is_shared() {
+                    return Err(ErrorCode::InvalidInputSyntax(
+                        "cannot alter parallelism of non-shared source.\nUse `ALTER MATERIALIZED VIEW SET PARALLELISM` to alter the materialized view using the source instead."
+                        .to_owned()
+                    )
+                    .into());
+                }
+
+                session.check_privilege_for_drop_alter(schema_name, &**source)?;
+                source.id
+            }
             StatementType::ALTER_SINK => {
                 let (sink, schema_name) =
                     reader.get_sink_by_name(db_name, schema_path, &real_table_name)?;
@@ -97,7 +112,7 @@ pub async fn handle_alter_parallelism(
 
     let catalog_writer = session.catalog_writer()?;
     catalog_writer
-        .alter_parallelism(table_id, target_parallelism, deferred)
+        .alter_parallelism(job_id, target_parallelism, deferred)
         .await?;
 
     if deferred {
diff --git a/src/frontend/src/handler/mod.rs b/src/frontend/src/handler/mod.rs
index 960d21b995532..3c1e2493bb683 100644
--- a/src/frontend/src/handler/mod.rs
+++ b/src/frontend/src/handler/mod.rs
@@ -1035,6 +1035,19 @@ pub async fn handle(
                 )
                 .await
             }
+            AlterSourceOperation::SetParallelism {
+                parallelism,
+                deferred,
+            } => {
+                alter_parallelism::handle_alter_parallelism(
+                    handler_args,
+                    name,
+                    parallelism,
+                    StatementType::ALTER_SOURCE,
+                    deferred,
+                )
+                .await
+            }
         },
         Statement::AlterFunction {
             name,
diff --git a/src/meta/service/src/ddl_service.rs b/src/meta/service/src/ddl_service.rs
index 30cb5ccec05f8..6d40654d6ce28 100644
--- a/src/meta/service/src/ddl_service.rs
+++ b/src/meta/service/src/ddl_service.rs
@@ -903,12 +903,12 @@ impl DdlService for DdlServiceImpl {
     ) -> Result<Response<AlterParallelismResponse>, Status> {
         let req = request.into_inner();
 
-        let table_id = req.get_table_id();
+        let job_id = req.get_table_id();
         let parallelism = *req.get_parallelism()?;
         let deferred = req.get_deferred();
         self.ddl_controller
             .reschedule_streaming_job(
-                table_id,
+                job_id,
                 JobRescheduleTarget {
                     parallelism: JobParallelismTarget::Update(TableParallelism::from(parallelism)),
                     resource_group: JobResourceGroupTarget::Keep,
diff --git a/src/meta/src/rpc/ddl_controller.rs b/src/meta/src/rpc/ddl_controller.rs
index 09f504dd50d98..bd317f76ebf93 100644
--- a/src/meta/src/rpc/ddl_controller.rs
+++ b/src/meta/src/rpc/ddl_controller.rs
@@ -394,7 +394,7 @@ impl DdlController {
     #[tracing::instrument(skip(self), level = "debug")]
     pub async fn reschedule_streaming_job(
         &self,
-        table_id: u32,
+        job_id: u32,
         target: JobRescheduleTarget,
         mut deferred: bool,
     ) -> MetaResult<()> {
@@ -407,7 +407,7 @@ impl DdlController {
         }
 
         self.stream_manager
-            .reschedule_streaming_job(table_id, target, deferred)
+            .reschedule_streaming_job(job_id, target, deferred)
             .await
     }
 
diff --git a/src/meta/src/stream/stream_manager.rs b/src/meta/src/stream/stream_manager.rs
index 2b4d35c08cca7..98802153b5846 100644
--- a/src/meta/src/stream/stream_manager.rs
+++ b/src/meta/src/stream/stream_manager.rs
@@ -613,7 +613,7 @@ impl GlobalStreamManager {
 
     pub(crate) async fn reschedule_streaming_job(
         &self,
-        table_id: u32,
+        job_id: u32,
         target: JobRescheduleTarget,
         deferred: bool,
     ) -> MetaResult<()> {
@@ -633,7 +633,7 @@ impl GlobalStreamManager {
                 if related_jobs.contains(&job) {
                     bail!(
                         "Cannot alter the job {} because the related job {} is currently being created",
-                        table_id,
+                        job_id,
                         job.table_id
                     );
                 }
@@ -648,10 +648,10 @@ impl GlobalStreamManager {
         let database_id = DatabaseId::new(
             self.metadata_manager
                 .catalog_controller
-                .get_object_database_id(table_id as ObjectId)
+                .get_object_database_id(job_id as ObjectId)
                 .await? as _,
         );
-        let table_id = TableId::new(table_id);
+        let job_id = TableId::new(job_id);
 
         let worker_nodes = self
             .metadata_manager
@@ -668,7 +668,7 @@ impl GlobalStreamManager {
             .sum::<usize>();
         let max_parallelism = self
             .metadata_manager
-            .get_job_max_parallelism(table_id)
+            .get_job_max_parallelism(job_id)
             .await?;
 
         if let JobParallelismTarget::Update(parallelism) = parallelism_change {
@@ -697,12 +697,12 @@ impl GlobalStreamManager {
         }
 
         let table_parallelism_assignment = match &parallelism_change {
-            JobParallelismTarget::Update(parallelism) => HashMap::from([(table_id, *parallelism)]),
+            JobParallelismTarget::Update(parallelism) => HashMap::from([(job_id, *parallelism)]),
             JobParallelismTarget::Refresh => HashMap::new(),
         };
         let resource_group_assignment = match &resource_group_change {
             JobResourceGroupTarget::Update(target) => {
-                HashMap::from([(table_id.table_id() as ObjectId, target.clone())])
+                HashMap::from([(job_id.table_id() as ObjectId, target.clone())])
             }
             JobResourceGroupTarget::Keep => HashMap::new(),
         };
@@ -710,7 +710,7 @@ impl GlobalStreamManager {
         if deferred {
             tracing::debug!(
                 "deferred mode enabled for job {}, set the parallelism directly to parallelism {:?}, resource group {:?}",
-                table_id,
+                job_id,
                 parallelism_change,
                 resource_group_change,
             );
@@ -728,7 +728,7 @@ impl GlobalStreamManager {
                 .scale_controller
                 .generate_job_reschedule_plan(JobReschedulePolicy {
                     targets: HashMap::from([(
-                        table_id.table_id,
+                        job_id.table_id,
                         JobRescheduleTarget {
                             parallelism: parallelism_change,
                             resource_group: resource_group_change,
@@ -740,7 +740,7 @@ impl GlobalStreamManager {
             if reschedule_plan.reschedules.is_empty() {
                 tracing::debug!(
                     "empty reschedule plan generated for job {}, set the parallelism directly to {:?}",
-                    table_id,
+                    job_id,
                     reschedule_plan.post_updates
                 );
                 self.scale_controller
diff --git a/src/sqlparser/src/ast/ddl.rs b/src/sqlparser/src/ast/ddl.rs
index 2133c1c5396b3..e84f874f7f0e5 100644
--- a/src/sqlparser/src/ast/ddl.rs
+++ b/src/sqlparser/src/ast/ddl.rs
@@ -208,14 +208,33 @@ pub enum AlterSubscriptionOperation {
 #[derive(Debug, Clone, PartialEq, Eq, Hash)]
 #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
 pub enum AlterSourceOperation {
-    RenameSource { source_name: ObjectName },
-    AddColumn { column_def: ColumnDef },
-    ChangeOwner { new_owner_name: Ident },
-    SetSchema { new_schema_name: ObjectName },
-    FormatEncode { format_encode: FormatEncodeOptions },
+    RenameSource {
+        source_name: ObjectName,
+    },
+    AddColumn {
+        column_def: ColumnDef,
+    },
+    ChangeOwner {
+        new_owner_name: Ident,
+    },
+    SetSchema {
+        new_schema_name: ObjectName,
+    },
+    FormatEncode {
+        format_encode: FormatEncodeOptions,
+    },
     RefreshSchema,
-    SetSourceRateLimit { rate_limit: i32 },
-    SwapRenameSource { target_source: ObjectName },
+    SetSourceRateLimit {
+        rate_limit: i32,
+    },
+    SwapRenameSource {
+        target_source: ObjectName,
+    },
+    /// `SET PARALLELISM TO <parallelism> [ DEFERRED ]`
+    SetParallelism {
+        parallelism: SetVariableValue,
+        deferred: bool,
+    },
 }
 
 #[derive(Debug, Clone, PartialEq, Eq, Hash)]
@@ -326,7 +345,7 @@ impl fmt::Display for AlterTableOperation {
             } => {
                 write!(
                     f,
-                    "SET PARALLELISM TO {} {}",
+                    "SET PARALLELISM TO {}{}",
                     parallelism,
                     if *deferred { " DEFERRED" } else { "" }
                 )
@@ -365,7 +384,7 @@ impl fmt::Display for AlterIndexOperation {
             } => {
                 write!(
                     f,
-                    "SET PARALLELISM TO {} {}",
+                    "SET PARALLELISM TO {}{}",
                     parallelism,
                     if *deferred { " DEFERRED" } else { "" }
                 )
@@ -392,7 +411,7 @@ impl fmt::Display for AlterViewOperation {
             } => {
                 write!(
                     f,
-                    "SET PARALLELISM TO {} {}",
+                    "SET PARALLELISM TO {}{}",
                     parallelism,
                     if *deferred { " DEFERRED" } else { "" }
                 )
@@ -437,7 +456,7 @@ impl fmt::Display for AlterSinkOperation {
             } => {
                 write!(
                     f,
-                    "SET PARALLELISM TO {} {}",
+                    "SET PARALLELISM TO {}{}",
                     parallelism,
                     if *deferred { " DEFERRED" } else { "" }
                 )
@@ -500,6 +519,17 @@ impl fmt::Display for AlterSourceOperation {
             AlterSourceOperation::SwapRenameSource { target_source } => {
                 write!(f, "SWAP WITH {}", target_source)
             }
+            AlterSourceOperation::SetParallelism {
+                parallelism,
+                deferred,
+            } => {
+                write!(
+                    f,
+                    "SET PARALLELISM TO {}{}",
+                    parallelism,
+                    if *deferred { " DEFERRED" } else { "" }
+                )
+            }
         }
     }
 }
diff --git a/src/sqlparser/src/parser.rs b/src/sqlparser/src/parser.rs
index 2120d0fd76418..eadbeb569c2ed 100644
--- a/src/sqlparser/src/parser.rs
+++ b/src/sqlparser/src/parser.rs
@@ -3619,8 +3619,22 @@ impl Parser<'_> {
                 }
             } else if let Some(rate_limit) = self.parse_alter_source_rate_limit(false)? {
                 AlterSourceOperation::SetSourceRateLimit { rate_limit }
+            } else if self.parse_keyword(Keyword::PARALLELISM) {
+                if self.expect_keyword(Keyword::TO).is_err()
+                    && self.expect_token(&Token::Eq).is_err()
+                {
+                    return self.expected("TO or = after ALTER SOURCE SET PARALLELISM");
+                }
+
+                let value = self.parse_set_variable()?;
+                let deferred = self.parse_keyword(Keyword::DEFERRED);
+
+                AlterSourceOperation::SetParallelism {
+                    parallelism: value,
+                    deferred,
+                }
             } else {
-                return self.expected("SCHEMA or SOURCE_RATE_LIMIT after SET");
+                return self.expected("SCHEMA, SOURCE_RATE_LIMIT or PARALLELISM after SET");
             }
         } else if self.peek_nth_any_of_keywords(0, &[Keyword::FORMAT]) {
             let format_encode = self.parse_schema()?.unwrap();
@@ -3634,9 +3648,7 @@ impl Parser<'_> {
             let target_source = self.parse_object_name()?;
             AlterSourceOperation::SwapRenameSource { target_source }
         } else {
-            return self.expected(
-                "RENAME, ADD COLUMN, OWNER TO, SET or SOURCE_RATE_LIMIT after ALTER SOURCE",
-            );
+            return self.expected("RENAME, ADD COLUMN, OWNER TO or SET after ALTER SOURCE");
         };
 
         Ok(Statement::AlterSource {
diff --git a/src/stream/src/executor/source/source_backfill_executor.rs b/src/stream/src/executor/source/source_backfill_executor.rs
index 2a325422551fb..a137744fe39d4 100644
--- a/src/stream/src/executor/source/source_backfill_executor.rs
+++ b/src/stream/src/executor/source/source_backfill_executor.rs
@@ -14,7 +14,6 @@
 
 use std::cmp::Ordering;
 use std::collections::{HashMap, HashSet};
-use std::sync::Once;
 use std::time::Instant;
 
 use anyhow::anyhow;
@@ -442,23 +441,7 @@ impl<S: StateStore> SourceBackfillExecutorInner<S> {
             .state_store()
             .clone();
         let table_id = self.backfill_state_store.state_store().table_id().into();
-        static STATE_TABLE_INITIALIZED: Once = Once::new();
-        tokio::spawn(async move {
-            // This is for self.backfill_finished() to be safe.
-            // We wait for 1st epoch's curr, i.e., the 2nd epoch's prev.
-            let epoch = first_epoch.curr;
-            tracing::info!("waiting for epoch: {}", epoch);
-            state_store
-                .try_wait_epoch(
-                    HummockReadEpoch::Committed(epoch),
-                    TryWaitEpochOptions { table_id },
-                )
-                .await
-                .expect("failed to wait epoch");
-            STATE_TABLE_INITIALIZED.call_once(|| ());
-            tracing::info!("finished waiting for epoch: {}", epoch);
-        });
-
+        let mut state_table_initialized = false;
         {
             let source_backfill_row_count = self
                 .metrics
@@ -651,6 +634,7 @@ impl<S: StateStore> SourceBackfillExecutorInner<S> {
                                     );
 
                                     let barrier_epoch = barrier.epoch;
+                                    let is_checkpoint = barrier.is_checkpoint();
                                     // yield barrier after reporting progress
                                     yield Message::Barrier(barrier);
 
@@ -665,11 +649,24 @@ impl<S: StateStore> SourceBackfillExecutorInner<S> {
                                         .await?;
                                     }
 
+                                    if !state_table_initialized && is_checkpoint {
+                                        // This is for self.backfill_finished() to be safe: wait until this actor can read all actors' written data.
+                                        // We wait for 2nd epoch
+                                        let epoch = barrier_epoch.prev;
+                                        tracing::info!("waiting for epoch: {}", epoch);
+                                        state_store
+                                            .try_wait_epoch(
+                                                HummockReadEpoch::Committed(epoch),
+                                                TryWaitEpochOptions { table_id },
+                                            )
+                                            .await?;
+                                        tracing::info!("finished waiting for epoch: {}", epoch);
+                                        state_table_initialized = true;
+                                    }
                                     // After we reported finished, we still don't exit the loop.
                                     // Because we need to handle split migration.
-                                    if STATE_TABLE_INITIALIZED.is_completed()
-                                        && self.backfill_finished(&backfill_stage.states).await?
-                                    {
+                                    if self.backfill_finished(&backfill_stage.states).await? {
+                                        tracing::info!("source backfill finished");
                                         break 'backfill_loop;
                                     }
                                 } else {
@@ -869,7 +866,7 @@ impl<S: StateStore> SourceBackfillExecutorInner<S> {
     ///
     /// Note: at the beginning, the actor will only read the state written by itself.
     /// It needs to _wait until it can read all actors' written data_.
-    /// i.e., wait for the first checkpoint has been available.
+    /// i.e., wait for the second checkpoint has been available.
     ///
     /// See <https://github.com/risingwavelabs/risingwave/issues/18300> for more details.
     async fn backfill_finished(&self, states: &BackfillStates) -> StreamExecutorResult<bool> {


Test patch:
diff --git a/e2e_test/source_inline/kafka/alter/alter_parallelism.slt b/e2e_test/source_inline/kafka/alter/alter_parallelism.slt
new file mode 100644
index 0000000000000..ea378be113a01
--- /dev/null
+++ b/e2e_test/source_inline/kafka/alter/alter_parallelism.slt
@@ -0,0 +1,199 @@
+# Note: this test should use 4 parallelism, depending on the risedev profile.
+# TODO: test-specific risedev profile https://github.com/risingwavelabs/risingwave/issues/20525
+
+control substitution on
+
+system ok
+rpk topic create alter_parallelism -p 10
+
+system ok
+cat<<EOF | rpk topic produce alter_parallelism -f "%p %v\n" -p 0
+0 {"x":10}
+0 {"x":1}
+0 {"x":2}
+EOF
+
+statement ok
+create source s_kafka(x int)
+WITH (
+    ${RISEDEV_KAFKA_WITH_OPTIONS_COMMON},
+    topic = 'alter_parallelism'
+)
+FORMAT PLAIN ENCODE JSON;
+
+statement ok
+create materialized view mv_kafka as select * from s_kafka;
+
+query ???
+select name, flags, parallelism from rw_fragments JOIN rw_relations ON rw_fragments.table_id = rw_relations.id order by name;
+----
+mv_kafka	{MVIEW,SOURCE_SCAN}	4
+s_kafka	{SOURCE}	4
+
+# split assignment is non-deterministic. so commented out.
+# system ok
+# risectl meta source-split-info
+# ----
+# Table #26
+# 	Fragment #1 (Source)
+# 		Actor #1   (3 splits): [1,7,8]
+# 		Actor #2   (2 splits): [0,5]
+# 		Actor #3   (2 splits): [2,6]
+# 		Actor #4   (3 splits): [3,4,9]
+# Table #28
+# 	Fragment #2 (SourceScan)
+# 		Actor #5   (3 splits): [1,7,8]
+# 		Actor #6   (2 splits): [0,5]
+# 		Actor #7   (2 splits): [2,6]
+# 		Actor #8   (3 splits): [3,4,9]
+
+
+system ok
+cat<<EOF | rpk topic produce alter_parallelism -f "%p %v\n" -p 0
+1 {"x":100}
+1 {"x":11}
+1 {"x":22}
+EOF
+
+statement ok
+ALTER MATERIALIZED VIEW mv_kafka SET PARALLELISM TO 2;
+
+# scale down
+# ALTER MV should have no effect, because of NoShuffle
+query ???
+select name, flags, parallelism from rw_fragments JOIN rw_relations ON rw_fragments.table_id = rw_relations.id order by name;
+----
+mv_kafka	{MVIEW,SOURCE_SCAN}	4
+s_kafka	{SOURCE}	4
+
+# ALTER SOURCE will change both the source and the mv parallelism.
+statement ok
+alter source s_kafka set parallelism = 2;
+
+query ???
+select name, flags, parallelism from rw_fragments JOIN rw_relations ON rw_fragments.table_id = rw_relations.id order by name;
+----
+mv_kafka	{MVIEW,SOURCE_SCAN}	2
+s_kafka	{SOURCE}	2
+
+# system ok
+# risectl meta source-split-info
+# ----
+# Table #26
+# 	Fragment #1 (Source)
+# 		Actor #1   (5 splits): [1,2,5,6,9]
+# 		Actor #2   (5 splits): [0,3,4,7,8]
+# Table #28
+# 	Fragment #2 (SourceScan)
+# 		Actor #5   (5 splits): [1,2,5,6,9]
+# 		Actor #6   (5 splits): [0,3,4,7,8]
+
+
+system ok
+cat<<EOF | rpk topic produce alter_parallelism -f "%p %v\n" -p 0
+2 {"x":1000}
+2 {"x":111}
+2 {"x":222}
+EOF
+
+statement ok
+create materialized view mv_kafka_2 as select * from s_kafka;
+
+query ???
+select name, flags, parallelism from rw_fragments JOIN rw_relations ON rw_fragments.table_id = rw_relations.id order by name;
+----
+mv_kafka	{MVIEW,SOURCE_SCAN}	2
+mv_kafka_2	{MVIEW,SOURCE_SCAN}	2
+s_kafka	{SOURCE}	2
+
+
+# scale up
+statement ok
+alter source s_kafka set parallelism = 5;
+
+query ???
+select name, flags, parallelism from rw_fragments JOIN rw_relations ON rw_fragments.table_id = rw_relations.id order by name;
+----
+mv_kafka	{MVIEW,SOURCE_SCAN}	5
+mv_kafka_2	{MVIEW,SOURCE_SCAN}	5
+s_kafka	{SOURCE}	5
+
+# system ok
+# risectl meta source-split-info
+# ----
+# Table #26
+# 	Fragment #1 (Source)
+# 		Actor #1   (2 splits): [2,9]
+# 		Actor #2   (2 splits): [0,6]
+# 		Actor #14  (2 splits): [3,7]
+# 		Actor #15  (2 splits): [1,8]
+# 		Actor #16  (2 splits): [4,5]
+# Table #28
+# 	Fragment #2 (SourceScan)
+# 		Actor #5   (2 splits): [2,9]
+# 		Actor #6   (2 splits): [0,6]
+# 		Actor #11  (2 splits): [3,7]
+# 		Actor #12  (2 splits): [1,8]
+# 		Actor #13  (2 splits): [4,5]
+# Table #30
+# 	Fragment #3 (SourceScan)
+# 		Actor #9   (2 splits): [2,9]
+# 		Actor #10  (2 splits): [0,6]
+# 		Actor #17  (2 splits): [3,7]
+# 		Actor #18  (2 splits): [1,8]
+# 		Actor #19  (2 splits): [4,5]
+
+
+query I
+select * from mv_kafka order by x;
+----
+1
+2
+10
+11
+22
+100
+111
+222
+1000
+
+
+query I
+select * from mv_kafka_2 order by x;
+----
+1
+2
+10
+11
+22
+100
+111
+222
+1000
+
+statement ok
+drop source s_kafka cascade;
+
+statement ok
+SET streaming_use_shared_source = false;
+
+statement ok
+create source s_kafka_non_shared(x int)
+WITH (
+    ${RISEDEV_KAFKA_WITH_OPTIONS_COMMON},
+    topic = 'alter_parallelism'
+)
+FORMAT PLAIN ENCODE JSON;
+
+statement error
+alter source s_kafka_non_shared set parallelism = 5;
+----
+db error: ERROR: Failed to run the query
+
+Caused by:
+  Invalid input syntax: cannot alter parallelism of non-shared source.
+Use `ALTER MATERIALIZED VIEW SET PARALLELISM` to alter the materialized view using the source instead.
+
+
+system ok
+rpk topic delete alter_parallelism;
diff --git a/e2e_test/source_inline/kafka/alter_table_drop_connector.slt b/e2e_test/source_inline/kafka/alter/alter_table_drop_connector.slt
similarity index 100%
rename from e2e_test/source_inline/kafka/alter_table_drop_connector.slt
rename to e2e_test/source_inline/kafka/alter/alter_table_drop_connector.slt
diff --git a/e2e_test/source_inline/kafka/shared_source.slt.serial b/e2e_test/source_inline/kafka/shared_source.slt.serial
index 1559962f8b878..7bfe393d43966 100644
--- a/e2e_test/source_inline/kafka/shared_source.slt.serial
+++ b/e2e_test/source_inline/kafka/shared_source.slt.serial
@@ -247,87 +247,6 @@ internal_table.mjs --name s0 --type source
 3,"{""split_info"": {""partition"": 3, ""start_offset"": 13, ""stop_offset"": null, ""topic"": ""shared_source""}, ""split_type"": ""kafka""}"
 
 
-# # Note: the parallelism depends on the risedev profile.
-# # So scale tests below are commented out.
-
-# query ???
-# select name, flags, parallelism from rw_fragments JOIN rw_relations ON rw_fragments.table_id = rw_relations.id order by name;
-# ----
-# mv_1	{MVIEW,SOURCE_SCAN}	5
-# mv_2	{MVIEW,SOURCE_SCAN}	5
-# s0	{SOURCE}	5
-
-
-# system ok
-# risectl meta source-split-info --ignore-id
-# ----
-# Table
-# 	Fragment (Source)
-# 		Actor (1 splits): [0]
-# 		Actor (1 splits): [2]
-# 		Actor (1 splits): [3]
-# 		Actor (1 splits): [1]
-# 		Actor (0 splits): []
-# Table
-# 	Fragment (SourceScan)
-# 		Actor (1 splits): [0] <- Upstream Actor #1055: [0]
-# 		Actor (1 splits): [2] <- Upstream Actor #1056: [2]
-# 		Actor (1 splits): [3] <- Upstream Actor #1057: [3]
-# 		Actor (1 splits): [1] <- Upstream Actor #1058: [1]
-# 		Actor (0 splits): [] <- Upstream Actor #1059: []
-# Table
-# 	Fragment (SourceScan)
-# 		Actor (1 splits): [0] <- Upstream Actor #1055: [0]
-# 		Actor (1 splits): [2] <- Upstream Actor #1056: [2]
-# 		Actor (1 splits): [3] <- Upstream Actor #1057: [3]
-# 		Actor (1 splits): [1] <- Upstream Actor #1058: [1]
-# 		Actor (0 splits): [] <- Upstream Actor #1059: []
-
-
-# # scale down
-# statement ok
-# ALTER MATERIALIZED VIEW mv_1 SET PARALLELISM TO 2;
-
-# # should have no effect, because of NoShuffle
-# # TODO: support ALTER SOURCE SET PARALLELISM, then we can
-# query ???
-# select name, flags, parallelism from rw_fragments JOIN rw_relations ON rw_fragments.table_id = rw_relations.id order by name;
-# ----
-# mv_1	{MVIEW,SOURCE_SCAN}	5
-# mv_2	{MVIEW,SOURCE_SCAN}	5
-# s0	{SOURCE}	5
-
-# system ok
-# risectl meta source-split-info --ignore-id
-# ----
-# Table
-# 	Fragment (Source)
-# 		Actor (1 splits): [0]
-# 		Actor (1 splits): [2]
-# 		Actor (1 splits): [3]
-# 		Actor (1 splits): [1]
-# 		Actor (0 splits): []
-# Table
-# 	Fragment (SourceScan)
-# 		Actor (1 splits): [0] <- Upstream Actor #1055: [0]
-# 		Actor (1 splits): [2] <- Upstream Actor #1056: [2]
-# 		Actor (1 splits): [3] <- Upstream Actor #1057: [3]
-# 		Actor (1 splits): [1] <- Upstream Actor #1058: [1]
-# 		Actor (0 splits): [] <- Upstream Actor #1059: []
-# Table
-# 	Fragment (SourceScan)
-# 		Actor (1 splits): [0] <- Upstream Actor #1055: [0]
-# 		Actor (1 splits): [2] <- Upstream Actor #1056: [2]
-# 		Actor (1 splits): [3] <- Upstream Actor #1057: [3]
-# 		Actor (1 splits): [1] <- Upstream Actor #1058: [1]
-# 		Actor (0 splits): [] <- Upstream Actor #1059: []
-
-
-# # Manual test: change the parallelism of the compute node, kill and restart, and check
-# # risedev ctl meta source-split-info --ignore-id
-# # risedev psql -c "select name, flags, parallelism from rw_fragments JOIN rw_relations ON rw_fragments.table_id = rw_relations.id order by name;"
-
-
 # Test: rate limit and resume won't lose data
 
 statement ok