fix schema check fow allowed operaions

Author: StrikeW

src/frontend/src/handler/alter_table_column.rs

@@ -97,22 +97,24 @@ pub async fn get_new_table_definition_for_cdc_table(
         "source schema should be None for CDC table"
     );
 
-    let orig_column_map: HashMap<String, ColumnDef> = HashMap::from_iter(
-        original_columns
+    let orig_column_catalog: HashMap<String, ColumnCatalog> = HashMap::from_iter(
+        original_catalog
+            .columns()
             .iter()
-            .map(|col| (col.name.real_value(), col.clone())),
+            .map(|col| (col.name().to_string(), col.clone())),
     );
 
     // update the original columns with new version columns
     let mut new_column_defs = vec![];
-    for col in new_columns {
-        // if the column exists in the original definitoins, use the original column definition.
+    for new_col in new_columns {
+        // if the column exists in the original catalog, use it to construct the column definition.
         // since we don't support altering the column type right now
-        if let Some(original_col) = orig_column_map.get(col.name()) {
-            new_column_defs.push(original_col.clone());
+        if let Some(original_col) = orig_column_catalog.get(new_col.name()) {
+            let ty = to_ast_data_type(original_col.data_type())?;
+            new_column_defs.push(ColumnDef::new(original_col.name().into(), ty, None, vec![]));
         } else {
-            let ty = to_ast_data_type(col.data_type())?;
-            new_column_defs.push(ColumnDef::new(col.name().into(), ty, None, vec![]));
+            let ty = to_ast_data_type(new_col.data_type())?;
+            new_column_defs.push(ColumnDef::new(new_col.name().into(), ty, None, vec![]));
         }
     }
     *original_columns = new_column_defs;

src/frontend/src/handler/create_table.rs

@@ -1187,6 +1187,7 @@ async fn derive_schema_for_cdc_table(
                     .iter_mut()
                     .zip_eq_fast(new_version_columns.into_iter())
                 {
+                    assert_eq!(col.name(), new_version_col.name());
                     col.column_desc.generated_or_default_column =
                         new_version_col.column_desc.generated_or_default_column;
                 }

src/meta/service/src/ddl_service.rs

@@ -19,6 +19,7 @@ use anyhow::anyhow;
 use rand::seq::SliceRandom;
 use rand::thread_rng;
 use risingwave_common::catalog::ColumnCatalog;
+use risingwave_common::types::DataType;
 use risingwave_common::util::column_index_mapping::ColIndexMapping;
 use risingwave_connector::sink::catalog::SinkId;
 use risingwave_meta::manager::{EventLogManagerRef, MetadataManager};
@@ -969,40 +970,41 @@ impl DdlService for DdlServiceImpl {
             for table in tables {
                 // Since we only support `ADD` and `DROP` column, we check whether the new columns and the original columns
                 // is a subset of the other.
-                let original_column_names: HashSet<String> = HashSet::from_iter(
-                    table
-                        .columns
-                        .iter()
-                        .map(|col| ColumnCatalog::from(col.clone()).column_desc.name),
-                );
-                let new_column_names: HashSet<String> = HashSet::from_iter(
-                    table_change
-                        .columns
-                        .iter()
-                        .map(|col| ColumnCatalog::from(col.clone()).column_desc.name),
-                );
-                if !(original_column_names.is_subset(&new_column_names)
-                    || original_column_names.is_superset(&new_column_names))
+                let original_columns: HashSet<(String, DataType)> =
+                    HashSet::from_iter(table.columns.iter().map(|col| {
+                        let col = ColumnCatalog::from(col.clone());
+                        let data_type = col.data_type().clone();
+                        (col.column_desc.name, data_type)
+                    }));
+                let new_columns: HashSet<(String, DataType)> =
+                    HashSet::from_iter(table_change.columns.iter().map(|col| {
+                        let col = ColumnCatalog::from(col.clone());
+                        let data_type = col.data_type().clone();
+                        (col.column_desc.name, data_type)
+                    }));
+
+                if !(original_columns.is_subset(&new_columns)
+                    || original_columns.is_superset(&new_columns))
                 {
                     tracing::warn!(target: "auto_schema_change",
                                     table_id = table.id,
                                     cdc_table_id = table.cdc_table_id,
                                     upstraem_ddl = table_change.upstream_ddl,
-                                    original_columns = ?original_column_names,
-                                    new_columns = ?new_column_names,
+                                    original_columns = ?original_columns,
+                                    new_columns = ?new_columns,
                                     "New columns should be a subset or superset of the original columns, since only `ADD COLUMN` and `DROP COLUMN` is supported");
                     return Err(Status::invalid_argument(
                         "New columns should be a subset or superset of the original columns",
                     ));
                 }
                 // skip the schema change if there is no change to original columns
-                if original_column_names == new_column_names {
+                if original_columns == new_columns {
                     tracing::warn!(target: "auto_schema_change",
                                    table_id = table.id,
                                    cdc_table_id = table.cdc_table_id,
                                    upstraem_ddl = table_change.upstream_ddl,
-                                    original_columns = ?original_column_names,
-                                    new_columns = ?new_column_names,
+                                    original_columns = ?original_columns,
+                                    new_columns = ?new_columns,
                                    "No change to columns, skipping the schema change");
                     continue;
                 }