Merge pull request #9 from MaterializeInc/create_sources_add_like

CREATE SOURCES should support filter

Author: JLDLaughlin

src/ast/mod.rs

@@ -411,6 +411,7 @@ pub enum Statement {
     },
     /// `CREATE SOURCES`
     CreateSources {
+        like: Option<String>,
         url: String,
         schema_registry: String,
         with_options: Vec<SqlOption>,
@@ -601,13 +602,18 @@ impl fmt::Display for Statement {
                 Ok(())
             }
             Statement::CreateSources {
+                like,
                 url,
                 schema_registry,
                 with_options,
             } => {
+                write!(f, "CREATE SOURCES ")?;
+                if let Some(like) = like {
+                    write!(f, "LIKE '{}' ", value::escape_single_quote_string(like),)?;
+                }
                 write!(
                     f,
-                    "CREATE SOURCES FROM {} USING SCHEMA REGISTRY {}",
+                    "FROM {} USING SCHEMA REGISTRY {}",
                     Value::SingleQuotedString(url.clone()).to_string(),
                     Value::SingleQuotedString(schema_registry.clone()).to_string(),
                 )?;

src/ast/visit.rs

@@ -318,11 +318,12 @@ pub trait Visit<'ast> {
 
     fn visit_create_sources(
         &mut self,
+        like: Option<&'ast String>,
         url: &'ast String,
         schema_registry: &'ast String,
         with_options: &'ast Vec<SqlOption>,
     ) {
-        visit_create_sources(self, url, schema_registry, with_options)
+        visit_create_sources(self, like, url, schema_registry, with_options)
     }
 
     fn visit_source_schema(&mut self, source_schema: &'ast SourceSchema) {
@@ -541,10 +542,11 @@ pub fn visit_statement<'ast, V: Visit<'ast> + ?Sized>(visitor: &mut V, statement
             with_options,
         } => visitor.visit_create_source(name, url, schema, with_options),
         Statement::CreateSources {
+            like,
             url,
             schema_registry,
             with_options,
-        } => visitor.visit_create_sources(url, schema_registry, with_options),
+        } => visitor.visit_create_sources(like.as_ref(), url, schema_registry, with_options),
         Statement::CreateSink {
             name,
             from,
@@ -1157,10 +1159,14 @@ pub fn visit_create_source<'ast, V: Visit<'ast> + ?Sized>(
 
 pub fn visit_create_sources<'ast, V: Visit<'ast> + ?Sized>(
     visitor: &mut V,
+    like: Option<&'ast String>,
     url: &'ast String,
     schema_registry: &'ast String,
     with_options: &'ast Vec<SqlOption>,
 ) {
+    if let Some(like) = like {
+        visitor.visit_literal_string(like);
+    }
     visitor.visit_literal_string(url);
     visitor.visit_literal_string(schema_registry);
     for option in with_options {

src/parser.rs

@@ -904,18 +904,29 @@ impl Parser {
     }
 
     pub fn parse_create_sources(&mut self) -> Result<Statement, ParserError> {
+        // Need to get the LIKE if it exists, otherwise keep moving.
+        let like = self.parse_like_filter()?;
         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 {
+            like,
             url,
             schema_registry,
             with_options,
         })
     }
 
+    fn parse_like_filter(&mut self) -> Result<Option<String>, ParserError> {
+        if self.parse_keyword("LIKE") {
+            Ok(Some(self.parse_literal_string()?))
+        } else {
+            Ok(None)
+        }
+    }
+
     pub fn parse_create_sink(&mut self) -> Result<Statement, ParserError> {
         let name = self.parse_object_name()?;
         self.expect_keyword("FROM")?;

tests/sqlparser_common.rs

@@ -2202,10 +2202,34 @@ fn parse_create_sources() {
     let sql = "CREATE SOURCES FROM 'kafka://whatever' USING SCHEMA REGISTRY 'http://foo.bar:8081'";
     match verified_stmt(sql) {
         Statement::CreateSources {
+            like,
             url,
             schema_registry,
             with_options,
         } => {
+            assert!(like.is_none());
+            assert_eq!("kafka://whatever", url);
+            assert_eq!("http://foo.bar:8081", schema_registry);
+            assert!(with_options.is_empty());
+        }
+        _ => assert!(false),
+    }
+}
+
+#[test]
+fn parse_create_sources_with_like_regex() {
+    let sql = "CREATE SOURCES LIKE '%foo%' FROM 'kafka://whatever' USING SCHEMA REGISTRY 'http://foo.bar:8081'";
+    match verified_stmt(sql) {
+        Statement::CreateSources {
+            like,
+            url,
+            schema_registry,
+            with_options,
+        } => {
+            match like {
+                Some(value) => assert_eq!("%foo%", value),
+                None => unimplemented!(),
+            }
             assert_eq!("kafka://whatever", url);
             assert_eq!("http://foo.bar:8081", schema_registry);
             assert!(with_options.is_empty());