feat: allow generated column in cdc table

src/common/src/catalog/column.rs

@@ -464,6 +464,10 @@ impl ColumnCatalog {
         }
     }
 
+    pub fn is_rw_sys_column(&self) -> bool {
+        self.column_desc.system_column.is_some()
+    }
+
     pub fn rw_timestamp_column() -> Self {
         Self {
             column_desc: rw_timestamp_column_desc(),

src/frontend/src/handler/create_source.rs

@@ -735,7 +735,7 @@ pub(crate) fn bind_all_columns(
             // TODO(yuhao): https://github.com/risingwavelabs/risingwave/issues/12209
             Err(RwError::from(ProtocolError(
                     format!("User-defined schema from SQL is not allowed with FORMAT {} ENCODE {}. \
-                    Please refer to https://www.risingwave.dev/docs/current/sql-create-source/ for more information.", format_encode.format, format_encode.row_encode))))
+                    Please refer to https://www.risingwave.dev/docs/current/sql-create-source/ for more information.", source_schema.format, source_schema.row_encode))))
         }
     } else {
         if wildcard_idx.is_some() {

src/frontend/src/handler/create_table.rs

@@ -829,12 +829,18 @@ pub(crate) fn gen_create_table_plan_for_cdc_table(
 
     let (options, secret_refs) = cdc_with_options.into_parts();
 
+    let non_generated_column_descs = columns
+        .iter()
+        .filter_map(|c| (!c.is_generated()).then(|| c.column_desc.clone()))
+        .collect_vec();
+    let non_generated_column_num = non_generated_column_descs.len();
+
     let cdc_table_desc = CdcTableDesc {
         table_id,
         source_id: source.id.into(), // id of cdc source streaming job
         external_table_name: external_table_name.clone(),
         pk: table_pk,
-        columns: columns.iter().map(|c| c.column_desc.clone()).collect(),
+        columns: non_generated_column_descs,
         stream_key: pk_column_indices,
         connect_properties: options,
         secret_refs,
@@ -852,7 +858,7 @@ pub(crate) fn gen_create_table_plan_for_cdc_table(
     );
 
     let scan_node: PlanRef = logical_scan.into();
-    let required_cols = FixedBitSet::with_capacity(columns.len());
+    let required_cols = FixedBitSet::with_capacity(non_generated_column_num);
     let plan_root = PlanRoot::new_with_logical_plan(
         scan_node,
         RequiredDist::Any,
@@ -1134,18 +1140,6 @@ fn sanity_check_for_cdc_table(
     constraints: &Vec<TableConstraint>,
     source_watermarks: &Vec<SourceWatermark>,
 ) -> Result<()> {
-    for c in column_defs {
-        for op in &c.options {
-            if let ColumnOption::GeneratedColumns(_) = op.option {
-                return Err(ErrorCode::NotSupported(
-                    "generated column defined on the table created from a CDC source".into(),
-                    "Remove the generated column in the column list".into(),
-                )
-                .into());
-            }
-        }
-    }
-
     // wildcard cannot be used with column definitions
     if wildcard_idx.is_some() && !column_defs.is_empty() {
         return Err(ErrorCode::NotSupported(

src/meta/service/src/ddl_service.rs

@@ -920,6 +920,30 @@ impl DdlService for DdlServiceImpl {
         };
 
         for table_change in schema_change.table_changes {
+            for c in &table_change.columns {
+                let c = ColumnCatalog::from(c.clone());
+
+                let invalid_col_type = |column_type: &str, c: &ColumnCatalog| {
+                    tracing::warn!(target: "auto_schema_change",
+                      cdc_table_id = table_change.cdc_table_id,
+                    upstraem_ddl = table_change.upstream_ddl,
+                        "invalid column type from cdc table change");
+                    return Err(Status::invalid_argument(format!(
+                        "invalid column type: {} from cdc table change, column: {:?}",
+                        column_type, c
+                    )));
+                };
+                if c.is_generated() {
+                    return invalid_col_type("generated column", &c);
+                }
+                if c.is_rw_sys_column() {
+                    return invalid_col_type("rw system column", &c);
+                }
+                if c.is_hidden {
+                    return invalid_col_type("hidden column", &c);
+                }
+            }
+
             // get the table catalog corresponding to the cdc table
             let tables: Vec<Table> = self
                 .metadata_manager
@@ -930,10 +954,14 @@ impl DdlService for DdlServiceImpl {
                 // 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_columns: HashSet<(String, DataType)> =
-                    HashSet::from_iter(table.columns.iter().map(|col| {
+                    HashSet::from_iter(table.columns.iter().filter_map(|col| {
                         let col = ColumnCatalog::from(col.clone());
                         let data_type = col.data_type().clone();
-                        (col.column_desc.name, data_type)
+                        if col.is_generated() {
+                            None
+                        } else {
+                            Some((col.column_desc.name, data_type))
+                        }
                     }));
                 let new_columns: HashSet<(String, DataType)> =
                     HashSet::from_iter(table_change.columns.iter().map(|col| {