Extend sqlparser semantics to support protobuf schemas

The format = protobuf_descriptor and message_name = "message_name"
arguments can be WITH args

Author: ruchirK

src/ast/mod.rs

@@ -745,7 +745,7 @@ impl fmt::Display for Statement {
                     Some(schema) => {
                         write!(f, " USING SCHEMA ")?;
                         match schema {
-                            SourceSchema::Raw(schema) => {
+                            SourceSchema::RawOrPath(schema) => {
                                 write!(f, "{}", Value::SingleQuotedString(schema.clone()))?;
                             }
                             SourceSchema::Registry(url) => {
@@ -1031,8 +1031,9 @@ impl fmt::Display for Function {
 /// Specifies the schema associated with a given Kafka topic.
 #[derive(Debug, Clone, PartialEq, Eq, Hash)]
 pub enum SourceSchema {
-    /// The schema is specified directly in the contained string.
-    Raw(String),
+    /// The schema is specified directly in the contained string
+    /// or its a path to a file
+    RawOrPath(String),
     /// The schema is available in a Confluent-compatible schema registry that
     /// is accessible at the specified URL.
     Registry(String),

src/ast/visit_macro.rs

@@ -1292,7 +1292,7 @@ macro_rules! make_visitor {
             source_schema: &'ast $($mut)* SourceSchema,
         ) {
             match source_schema {
-                SourceSchema::Raw(schema) => visitor.visit_literal_string(schema),
+                SourceSchema::RawOrPath(schema) => visitor.visit_literal_string(schema),
                 SourceSchema::Registry(url) => visitor.visit_literal_string(url),
             }
         }

src/parser.rs

@@ -1238,7 +1238,7 @@ impl Parser {
             let schema = if self.parse_keyword("REGISTRY") {
                 SourceSchema::Registry(self.parse_literal_string()?)
             } else {
-                SourceSchema::Raw(self.parse_literal_string()?)
+                SourceSchema::RawOrPath(self.parse_literal_string()?)
             };
             Some(schema)
         } else {

tests/sqlparser_common.rs

@@ -2878,7 +2878,7 @@ fn parse_create_source_raw_schema() {
         } => {
             assert_eq!("foo", name.to_string());
             assert_eq!("bar", url);
-            assert_eq!(SourceSchema::Raw("baz".into()), schema.unwrap());
+            assert_eq!(SourceSchema::RawOrPath("baz".into()), schema.unwrap());
             assert_eq!(
                 with_options,
                 vec![SqlOption {
@@ -2891,6 +2891,37 @@ fn parse_create_source_raw_schema() {
     }
 }
 
+#[test]
+fn parse_create_source_path_schema_multiple_args() {
+    let sql = "CREATE SOURCE foo FROM 'bar' USING SCHEMA 'path' WITH (format = 'someformat', message_name = 'somemessage')";
+    match verified_stmt(sql) {
+        Statement::CreateSource {
+            name,
+            url,
+            schema,
+            with_options,
+        } => {
+            assert_eq!("foo", name.to_string());
+            assert_eq!("bar", url);
+            assert_eq!(SourceSchema::RawOrPath("path".into()), schema.unwrap());
+            assert_eq!(
+                with_options,
+                vec![
+                    SqlOption {
+                        name: "format".into(),
+                        value: Value::SingleQuotedString("someformat".into())
+                    },
+                    SqlOption {
+                        name: "message_name".into(),
+                        value: Value::SingleQuotedString("somemessage".into())
+                    },
+                ]
+            );
+        }
+        _ => assert!(false),
+    }
+}
+
 #[test]
 fn parse_create_source_registry() {
     let sql = "CREATE SOURCE foo FROM 'bar' USING SCHEMA REGISTRY 'http://localhost:8081'";