do not use Option for definition to persist

Signed-off-by: Bugen Zhao <[email protected]>

src/frontend/src/handler/alter_table_column.rs

@@ -89,7 +89,7 @@ pub async fn get_replace_table_plan(
     session: &Arc<SessionImpl>,
     table_name: ObjectName,
     new_definition_for_plan: Statement,
-    new_definition_to_persist: Option<Statement>,
+    new_definition_to_persist: Statement,
     old_catalog: &Arc<TableCatalog>,
 ) -> Result<(
     Option<Source>,
@@ -98,11 +98,9 @@ pub async fn get_replace_table_plan(
     ColIndexMapping,
     TableJobType,
 )> {
-    let new_definition_to_persist = new_definition_to_persist
-        .as_ref()
-        .unwrap_or(&new_definition_for_plan);
     // Create handler args as if we're creating a new table with the altered definition.
-    let handler_args = HandlerArgs::new(session.clone(), new_definition_to_persist, Arc::from(""))?;
+    let handler_args =
+        HandlerArgs::new(session.clone(), &new_definition_to_persist, Arc::from(""))?;
 
     let col_id_gen = ColumnIdGenerator::new_alter(old_catalog);
     let Statement::CreateTable {
@@ -433,8 +431,8 @@ pub async fn handle_alter_table_column(
     let (source, table, graph, col_index_mapping, job_type) = get_replace_table_plan(
         &session,
         table_name,
-        definition,
-        raw_definition,
+        definition.clone(),
+        raw_definition.unwrap_or(definition),
         &original_catalog,
     )
     .await?;

src/frontend/src/handler/alter_table_with_sr.rs

@@ -27,7 +27,7 @@ use crate::error::{ErrorCode, Result};
 use crate::TableCatalog;
 
 fn get_format_encode_from_table(table: &TableCatalog) -> Result<Option<FormatEncodeOptions>> {
-    let stmt = table.create_sql_ast_purified()?;
+    let stmt = table.create_sql_ast()?;
     let Statement::CreateTable { format_encode, .. } = stmt else {
         unreachable!()
     };
@@ -57,14 +57,20 @@ pub async fn handle_refresh_schema(
         .into());
     }
 
-    // Not using the purified definition because we want to re-fetch the schema.
+    // HIGHTLIGHT: Not using the purified definition because we want to re-fetch the schema.
     let definition = original_table
         .create_sql_ast()
         .context("unable to parse original table definition")?;
 
     let (source, table, graph, col_index_mapping, job_type) = {
-        let result =
-            get_replace_table_plan(&session, table_name, definition, None, &original_table).await;
+        let result = get_replace_table_plan(
+            &session,
+            table_name,
+            definition.clone(),
+            definition,
+            &original_table,
+        )
+        .await;
         match result {
             Ok((source, table, graph, col_index_mapping, job_type)) => {
                 Ok((source, table, graph, col_index_mapping, job_type))

src/frontend/src/handler/create_sink.rs

@@ -520,6 +520,7 @@ pub(crate) async fn reparse_table_for_sink(
         .map(|format_encode| format_encode.into_v2_with_warning());
 
     // Create handler args as if we're creating a new table with the altered definition.
+    // Use `raw_definition` for persistence.
     let handler_args = HandlerArgs::new(session.clone(), &raw_definition, Arc::from(""))?;
     let col_id_gen = ColumnIdGenerator::new_alter(table_catalog);
     let Statement::CreateTable {

src/frontend/src/rpc/mod.rs

@@ -104,8 +104,8 @@ async fn get_new_table_plan(
     let (_, table, graph, col_index_mapping, job_type) = get_replace_table_plan(
         &session,
         table_name,
+        new_table_definition.clone(),
         new_table_definition,
-        None,
         &original_catalog,
     )
     .await?;