use purified for schema change

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

src/frontend/src/handler/alter_table_column.rs

@@ -55,7 +55,6 @@ pub async fn get_new_table_definition_for_cdc_table(
     );
 
     // Retrieve the original table definition.
-    // TODO(purify): use purified definition.
     let mut definition = original_catalog.create_sql_ast()?;
 
     // Clear the original columns field, so that we'll follow `new_columns` to generate a
@@ -260,8 +259,7 @@ pub async fn handle_alter_table_column(
     }
 
     // Retrieve the original table definition and parse it to AST.
-    // TODO(purify): use purified definition.
-    let mut definition = original_catalog.create_sql_ast()?;
+    let mut definition = original_catalog.create_sql_ast_purified()?;
     let Statement::CreateTable {
         columns,
         format_encode,

src/frontend/src/handler/alter_table_with_sr.rs

@@ -27,7 +27,6 @@ use crate::error::{ErrorCode, Result};
 use crate::TableCatalog;
 
 fn get_format_encode_from_table(table: &TableCatalog) -> Result<Option<FormatEncodeOptions>> {
-    // TODO(purify): use purified definition.
     let stmt = table.create_sql_ast()?;
     let Statement::CreateTable { format_encode, .. } = stmt else {
         unreachable!()
@@ -58,9 +57,8 @@ pub async fn handle_refresh_schema(
         .into());
     }
 
-    // Not using the purified definition because we want to re-fetch the schema.
     let definition = original_table
-        .create_sql_ast()
+        .create_sql_ast_purified()
         .context("unable to parse original table definition")?;
 
     let (source, table, graph, col_index_mapping, job_type) = {

src/frontend/src/handler/create_sink.rs

@@ -502,8 +502,7 @@ pub(crate) async fn reparse_table_for_sink(
     table_catalog: &Arc<TableCatalog>,
 ) -> Result<(StreamFragmentGraph, Table, Option<PbSource>)> {
     // Retrieve the original table definition and parse it to AST.
-    // TODO(purify): use purified definition.
-    let definition = table_catalog.create_sql_ast()?;
+    let definition = table_catalog.create_sql_ast_purified()?;
     let Statement::CreateTable {
         name,
         format_encode,