Merge pull request #4 from umanwizard/create_sources

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

@@ -816,6 +816,15 @@ impl Parser {
         }
     }
 
+    /// Bail out if the following tokens are not the excpected sequence of keywords,
+    /// or consume them if they are
+    pub fn expect_keywords(&mut self, expected: &[&'static str]) -> Result<(), ParserError> {
+        for kw in expected {
+            self.expect_keyword(kw)?;
+        }
+        Ok(())
+    }
+
     /// Consume the next token if it matches the expected token, otherwise return false
     #[must_use]
     pub fn consume_token(&mut self, expected: &Token) -> bool {
@@ -846,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") {
@@ -869,10 +880,7 @@ impl Parser {
         } else {
             SourceSchema::Raw(self.parse_literal_string()?)
         };
-        let mut with_options = vec![];
-        if self.parse_keyword("WITH") {
-            with_options = self.parse_with_options()?;
-        }
+        let with_options = self.parse_with_options()?;
         Ok(Statement::CreateSource {
             name,
             url,
@@ -881,16 +889,26 @@ 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")?;
         let from = self.parse_object_name()?;
         self.expect_keyword("INTO")?;
         let url = self.parse_literal_string()?;
-        let mut with_options = vec![];
-        if self.parse_keyword("WITH") {
-            with_options = self.parse_with_options()?;
-        }
+        let with_options = self.parse_with_options()?;
         Ok(Statement::CreateSink {
             name,
             from,
@@ -928,11 +946,7 @@ impl Parser {
         // ANSI SQL and Postgres support RECURSIVE here, but we don't support it either.
         let name = self.parse_object_name()?;
         let columns = self.parse_parenthesized_column_list(Optional)?;
-        let with_options = if self.parse_keyword("WITH") {
-            self.parse_with_options()?
-        } else {
-            vec![]
-        };
+        let with_options = self.parse_with_options()?;
         self.expect_keyword("AS")?;
         let query = Box::new(self.parse_query()?);
         // Optional `WITH [ CASCADED | LOCAL ] CHECK OPTION` is widely supported here.
@@ -983,11 +997,7 @@ impl Parser {
         // parse optional column list (schema)
         let (columns, constraints) = self.parse_columns()?;
 
-        let with_options = if self.parse_keyword("WITH") {
-            self.parse_with_options()?
-        } else {
-            vec![]
-        };
+        let with_options = self.parse_with_options()?;
 
         Ok(Statement::CreateTable {
             name: table_name,
@@ -1135,19 +1145,23 @@ impl Parser {
     }
 
     pub fn parse_with_options(&mut self) -> Result<Vec<SqlOption>, ParserError> {
-        self.expect_token(&Token::LParen)?;
-        let mut options = vec![];
-        loop {
-            let name = self.parse_identifier()?;
-            self.expect_token(&Token::Eq)?;
-            let value = self.parse_value()?;
-            options.push(SqlOption { name, value });
-            if !self.consume_token(&Token::Comma) {
-                break;
+        if self.parse_keyword("WITH") {
+            self.expect_token(&Token::LParen)?;
+            let mut options = vec![];
+            loop {
+                let name = self.parse_identifier()?;
+                self.expect_token(&Token::Eq)?;
+                let value = self.parse_value()?;
+                options.push(SqlOption { name, value });
+                if !self.consume_token(&Token::Comma) {
+                    break;
+                }
             }
+            self.expect_token(&Token::RParen)?;
+            Ok(options)
+        } else {
+            Ok(vec![])
         }
-        self.expect_token(&Token::RParen)?;
-        Ok(options)
     }
 
     pub fn parse_alter(&mut self) -> Result<Statement, ParserError> {

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')";