cdc parser

risingwavelabs · May 8, 2024 · efad6a3 · efad6a3
1 parent 39491e6
commit efad6a3
Showing 1 changed file with 9 additions and 3 deletions.
diff --git a/src/sqlparser/src/ast/statement.rs b/src/sqlparser/src/ast/statement.rs
@@ -376,9 +376,15 @@ impl ParseTo for CreateSourceStatement {
             .iter()
             .find(|&opt| opt.name.real_value() == UPSTREAM_SOURCE_KEY);
         let connector: String = option.map(|opt| opt.value.to_string()).unwrap_or_default();
-        // The format of cdc source job is fixed to `FORMAT PLAIN ENCODE JSON`
-        let cdc_source_job =
-            connector.contains("-cdc") && columns.is_empty() && constraints.is_empty();
+        let cdc_source_job = connector.contains("-cdc");
+        if cdc_source_job {
+            if !columns.is_empty() || !constraints.is_empty() {
+                return Err(ParserError::ParserError(
+                    "CDC source cannot define columns and constraints".to_string(),
+                ));
+            }
+        }
+
         // row format for nexmark source must be native
         // default row format for datagen source is native
         let source_schema = p.parse_source_schema_with_connector(&connector, cdc_source_job)?;