fix(parser): cdc source cannot define columns and constraints (#16636)

risingwavelabs · May 10, 2024 · 2dd6b73 · 2dd6b73
1 parent e5c9c48
commit 2dd6b73
Showing 1 changed file with 7 additions and 3 deletions.
diff --git a/src/sqlparser/src/ast/statement.rs b/src/sqlparser/src/ast/statement.rs
@@ -402,9 +402,13 @@ 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 && (!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)?;