feat(sql): Remove the ROW FORMAT keyword for CREATE statement of nati…

…ve CDC connectors (#7133)

Remove the ROW FORMAT keyword for CREATE statement of native CDC connectors, which must be debezium json format.

```
create materialized source orders (
order_id int,
order_date timestamp,
customer_name string,
price decimal,
product_id int,
order_status smallint,
PRIMARY KEY (order_id)
) with (
connector = 'mysql-cdc',
hostname = '127.0.0.1',
...
) row format debezium_json <-- removed
```

Approved-By: neverchanje

e2e_test/source/cdc/cdc.load.slt

@@ -14,7 +14,7 @@ create materialized source products ( id INT,
  database.name = 'mydb',
  table.name = 'products',
  server.id = '5085'
-) row format debezium_json;
+);
 
 statement ok
 create materialized view products_cnt as select count(*) as cnt from products;
@@ -37,7 +37,7 @@ create materialized source orders (
  database.name = 'mydb',
  table.name = 'orders',
  server.id = '5086'
-) row format debezium_json;
+);
 
 statement ok
 create materialized view orders_cnt as select count(*) as cnt from orders;
@@ -60,7 +60,7 @@ create materialized source shipments (
  schema.name = 'public',
  table.name = 'shipments',
  slot.name = 'shipments'
-) row format debezium_json;
+);
 
 statement ok
 create materialized view shipments_cnt as select count(*) as cnt from shipments;
@@ -87,4 +87,4 @@ create materialized source mytable (
  database.name = 'mydb',
  table.name = 'mytable',
  server.id = '5087'
-) row format debezium_json;
+);

e2e_test/source/cdc/cdc.validate.mysql.slt

@@ -15,7 +15,7 @@ create materialized source products ( id INT,
  database.name = 'mydb',
  table.name = 'products',
  server.id = '5085'
-) row format debezium_json;
+);
 
 # invalid password
 statement error
@@ -32,7 +32,7 @@ create materialized source products ( id INT,
  database.name = 'mydb',
  table.name = 'products',
  server.id = '5085'
-) row format debezium_json;
+);
 
 # invalid database name
 statement error
@@ -49,7 +49,7 @@ create materialized source products ( id INT,
  database.name = 'mdb',
  table.name = 'products',
  server.id = '5085'
-) row format debezium_json;
+);
 
 # invalid table name
 statement error
@@ -66,7 +66,7 @@ create materialized source products ( id INT,
  database.name = 'mydb',
  table.name = 'prdcts',
  server.id = '5085'
-) row format debezium_json;
+);
 
 # invalid primary key
 statement error
@@ -87,7 +87,7 @@ create materialized source orders (
  database.name = 'mydb',
  table.name = 'orders',
  server.id = '5086'
-) row format debezium_json;
+);
 
 # column data type mismatch
 statement error
@@ -108,7 +108,7 @@ create materialized source orders (
  database.name = 'mydb',
  table.name = 'orders',
  server.id = '5086'
-) row format debezium_json;
+);
 
 # column name mismatch
 statement error
@@ -129,4 +129,4 @@ create materialized source orders (
  database.name = 'mydb',
  table.name = 'orders',
  server.id = '5086'
-) row format debezium_json;
+);

e2e_test/source/cdc/cdc.validate.postgres.slt

@@ -19,7 +19,7 @@ create materialized source shipments (
  schema.name = 'public',
  table.name = 'shipments',
  slot.name = 'shipments'
-) row format debezium_json;
+);
 
 
 # invalid password
@@ -41,7 +41,7 @@ create materialized source shipments (
  schema.name = 'public',
  table.name = 'shipments',
  slot.name = 'shipments'
-) row format debezium_json;
+);
 
 # invalid table name
 statement error
@@ -62,7 +62,7 @@ create materialized source shipments (
  schema.name = 'public',
  table.name = 'shipment',
  slot.name = 'shipments'
-) row format debezium_json;
+);
 
 
 # invalid primary key
@@ -84,7 +84,7 @@ create materialized source shipments (
  schema.name = 'public',
  table.name = 'shipments',
  slot.name = 'shipments'
-) row format debezium_json;
+);
 
 
 # column name mismatch
@@ -106,7 +106,7 @@ create materialized source shipments (
  schema.name = 'public',
  table.name = 'shipments',
  slot.name = 'shipments'
-) row format debezium_json;
+);
 
 # column data type mismatch
 statement error
@@ -127,4 +127,4 @@ create materialized source shipments (
  schema.name = 'public',
  table.name = 'shipments',
  slot.name = 'shipments'
-) row format debezium_json;
+);

src/sqlparser/src/ast/statement.rs

@@ -267,8 +267,18 @@ impl ParseTo for CreateSourceStatement {
         let (columns, constraints) = p.parse_columns()?;
 
         impl_parse_to!(with_properties: WithProperties, p);
-        impl_parse_to!([Keyword::ROW, Keyword::FORMAT], p);
-        impl_parse_to!(source_schema: SourceSchema, p);
+        let option = with_properties
+            .0
+            .iter()
+            .find(|&opt| opt.name.real_value() == "connector");
+        // row format for cdc source must be debezium json
+        let source_schema = if let Some(opt) = option && opt.value.to_string().contains("-cdc") {
+            SourceSchema::DebeziumJson
+        } else {
+            impl_parse_to!([Keyword::ROW, Keyword::FORMAT], p);
+            SourceSchema::parse_to(p)?
+        };
+
         Ok(Self {
             if_not_exists,
             columns,

src/sqlparser/src/lib.rs

@@ -32,6 +32,7 @@
 
 #![cfg_attr(not(feature = "std"), no_std)]
 #![feature(lint_reasons)]
+#![feature(let_chains)]
 #![expect(clippy::derive_partial_eq_without_eq)]
 #![expect(clippy::doc_markdown)]
 #![expect(clippy::upper_case_acronyms)]