fix pk check

e2e_test/source/cdc/cdc.share_stream.slt

@@ -27,10 +27,9 @@ statement error Should not create MATERIALIZED VIEW or SELECT directly on shared
 create materialized view mv as select * from mysql_mytest;
 
 statement error The upstream table name must contain database name prefix*
-create table products_test ( id INT,
+create table products_test ( id INT PRIMARY KEY,
    name STRING,
-   description STRING,
-   PRIMARY KEY (id)
+   description STRING
 ) from mysql_mytest table 'products';
 
 statement ok
@@ -233,12 +232,11 @@ CREATE TABLE IF NOT EXISTS postgres_all_types(
 
 statement error The upstream table name must contain schema name prefix*
 CREATE TABLE person_new (
-    id int,
+    id int PRIMARY KEY,
     name varchar,
     email_address varchar,
     credit_card varchar,
-    city varchar,
-    PRIMARY KEY (id)
+    city varchar
 ) FROM pg_source TABLE 'person';
 
 statement ok

src/frontend/src/handler/create_table.rs

@@ -743,6 +743,7 @@ pub(crate) fn gen_create_table_plan_for_cdc_source(
     with_version_column: Option<String>,
     include_column_options: IncludeOption,
 ) -> Result<(PlanRef, PbTable)> {
+    // cdc table must have primary key constraint or primary key column
     if !constraints.iter().any(|c| {
         matches!(
             c,
@@ -751,6 +752,10 @@ pub(crate) fn gen_create_table_plan_for_cdc_source(
                 ..
             }
         )
+    }) && !column_defs.iter().any(|col| {
+        col.options
+            .iter()
+            .any(|opt| matches!(opt.option, ColumnOption::Unique { is_primary: true }))
     }) {
         return Err(ErrorCode::NotSupported(
             "CDC table without primary key constraint is not supported".to_owned(),