replace table keep vnode count

Signed-off-by: Bugen Zhao <i@bugenzhao.com>

src/frontend/src/handler/alter_table_column.rs

@@ -191,7 +191,7 @@ pub async fn get_replace_table_plan(
         panic!("unexpected statement type: {:?}", definition);
     };
 
-    let (mut graph, mut table, source, job_type) = generate_stream_graph_for_table(
+    let (mut graph, table, source, job_type) = generate_stream_graph_for_table(
         session,
         table_name,
         original_catalog,
@@ -241,7 +241,10 @@ pub async fn get_replace_table_plan(
         )?;
     }
 
+    // Set some fields ourselves so that the meta service does not need to maintain them.
+    let mut table = table;
     table.incoming_sinks = incoming_sink_ids.iter().copied().collect();
+    table.maybe_vnode_count = Some(original_catalog.vnode_count() as _);
 
     Ok((source, table, graph, col_index_mapping, job_type))
 }

src/meta/src/rpc/ddl_controller.rs

@@ -2149,8 +2149,8 @@ impl DdlController {
             old_table_fragments.assigned_parallelism,
         );
 
-        // TODO(var-vnode): fill vnode count for table catalog in `stream_job`,
-        // like what we do in `build_stream_job`.
+        // Note: no need to set `vnode_count` as it's already set by the frontend.
+        // See `get_replace_table_plan`.
 
         let ctx = ReplaceTableContext {
             old_table_fragments,