parse CREATE SOURCES

Author: umanwizard

src/ast/mod.rs

@@ -397,6 +397,12 @@ pub enum Statement {
         schema: SourceSchema,
         with_options: Vec<SqlOption>,
     },
+    /// CREATE SOURCES
+    CreateSources {
+        url: String,
+        schema_registry: String,
+        with_options: Vec<SqlOption>,
+    },
     /// CREATE SINK
     CreateSink {
         name: ObjectName,
@@ -558,6 +564,22 @@ impl fmt::Display for Statement {
                 }
                 Ok(())
             }
+            Statement::CreateSources {
+                url,
+                schema_registry,
+                with_options,
+            } => {
+                write!(
+                    f,
+                    "CREATE SOURCES FROM {} USING SCHEMA REGISTRY {}",
+                    Value::SingleQuotedString(url.clone()).to_string(),
+                    Value::SingleQuotedString(schema_registry.clone()).to_string(),
+                )?;
+                if !with_options.is_empty() {
+                    write!(f, " WITH ({})", display_comma_separated(with_options))?;
+                }
+                Ok(())
+            }
             Statement::CreateSink {
                 name,
                 from,

src/ast/visit.rs

@@ -316,6 +316,15 @@ pub trait Visit<'ast> {
         visit_create_source(self, name, url, schema, with_options)
     }
 
+    fn visit_create_sources(
+        &mut self,
+        url: &'ast String,
+        schema_registry: &'ast String,
+        with_options: &'ast Vec<SqlOption>,
+    ) {
+        visit_create_sources(self, url, schema_registry, with_options)
+    }
+
     fn visit_source_schema(&mut self, source_schema: &'ast SourceSchema) {
         visit_source_schema(self, source_schema)
     }
@@ -504,6 +513,11 @@ pub fn visit_statement<'ast, V: Visit<'ast> + ?Sized>(visitor: &mut V, statement
             schema,
             with_options,
         } => visitor.visit_create_source(name, url, schema, with_options),
+        Statement::CreateSources {
+            url,
+            schema_registry,
+            with_options,
+        } => visitor.visit_create_sources(url, schema_registry, with_options),
         Statement::CreateSink {
             name,
             from,
@@ -1103,6 +1117,19 @@ pub fn visit_create_source<'ast, V: Visit<'ast> + ?Sized>(
     }
 }
 
+pub fn visit_create_sources<'ast, V: Visit<'ast> + ?Sized>(
+    visitor: &mut V,
+    url: &'ast String,
+    schema_registry: &'ast String,
+    with_options: &'ast Vec<SqlOption>,
+) {
+    visitor.visit_literal_string(url);
+    visitor.visit_literal_string(schema_registry);
+    for option in with_options {
+        visitor.visit_option(option);
+    }
+}
+
 fn visit_source_schema<'ast, V: Visit<'ast> + ?Sized>(
     visitor: &mut V,
     source_schema: &'ast SourceSchema,

src/parser.rs

@@ -855,6 +855,8 @@ impl Parser {
             self.parse_create_view()
         } else if self.parse_keyword("SOURCE") {
             self.parse_create_source()
+        } else if self.parse_keyword("SOURCES") {
+            self.parse_create_sources()
         } else if self.parse_keyword("SINK") {
             self.parse_create_sink()
         } else if self.parse_keyword("EXTERNAL") {
@@ -887,6 +889,19 @@ impl Parser {
         })
     }
 
+    pub fn parse_create_sources(&mut self) -> Result<Statement, ParserError> {
+        self.expect_keyword("FROM")?;
+        let url = self.parse_literal_string()?;
+        self.expect_keywords(&["USING", "SCHEMA", "REGISTRY"])?;
+        let schema_registry = self.parse_literal_string()?;
+        let with_options = self.parse_with_options()?;
+        Ok(Statement::CreateSources {
+            url,
+            schema_registry,
+            with_options,
+        })
+    }
+
     pub fn parse_create_sink(&mut self) -> Result<Statement, ParserError> {
         let name = self.parse_object_name()?;
         self.expect_keyword("FROM")?;

tests/sqlparser_common.rs

@@ -2216,6 +2216,23 @@ fn parse_create_source_registry() {
     }
 }
 
+#[test]
+fn parse_create_sources() {
+    let sql = "CREATE SOURCES FROM 'kafka://whatever' USING SCHEMA REGISTRY 'http://foo.bar:8081'";
+    match verified_stmt(sql) {
+        Statement::CreateSources {
+            url,
+            schema_registry,
+            with_options,
+        } => {
+            assert_eq!("kafka://whatever", url);
+            assert_eq!("http://foo.bar:8081", schema_registry);
+            assert!(with_options.is_empty());
+        }
+        _ => assert!(false)
+    }
+}
+
 #[test]
 fn parse_create_sink() {
     let sql = "CREATE SINK foo FROM bar INTO 'baz' WITH (name = 'val')";