Snowflake: parse CREATE STREAM ON VIEW + StreamSourceKind discriminator

src/ast/mod.rs

@@ -4945,7 +4945,7 @@ pub enum Statement {
         name: ObjectName,
     },
     /// ```sql
-    /// CREATE [OR REPLACE] STREAM [IF NOT EXISTS] <name> ON TABLE <table>
+    /// CREATE [OR REPLACE] STREAM [IF NOT EXISTS] <name> ON { TABLE | VIEW } <source>
     /// ```
     CreateStream {
         /// `OR REPLACE` flag.
@@ -4954,7 +4954,10 @@ pub enum Statement {
         if_not_exists: bool,
         /// Stream name.
         name: ObjectName,
-        /// The source table the stream tracks (`ON TABLE <table>`).
+        /// Whether the source is a table (`ON TABLE`) or a view (`ON VIEW`).
+        source_kind: StreamSourceKind,
+        /// The source object the stream tracks (the `<source>` after
+        /// `ON TABLE`/`ON VIEW`).
         source_table: ObjectName,
     },
     /// ```sql
@@ -7408,11 +7411,12 @@ impl fmt::Display for Statement {
                 or_replace,
                 if_not_exists,
                 name,
+                source_kind,
                 source_table,
             } => {
                 write!(
                     f,
-                    "CREATE {or_replace}STREAM {if_not_exists}{name} ON TABLE {source_table}",
+                    "CREATE {or_replace}STREAM {if_not_exists}{name} ON {source_kind} {source_table}",
                     or_replace = if *or_replace { "OR REPLACE " } else { "" },
                     if_not_exists = if *if_not_exists { "IF NOT EXISTS " } else { "" },
                 )
@@ -10439,6 +10443,28 @@ impl fmt::Display for ObjectType {
     }
 }
 
+#[derive(Debug, Copy, Clone, PartialEq, PartialOrd, Eq, Ord, Hash)]
+#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+#[cfg_attr(feature = "visitor", derive(Visit, VisitMut))]
+/// The kind of object a Snowflake stream tracks, i.e. whether it was created
+/// with `ON TABLE <name>` or `ON VIEW <name>`.
+/// <https://docs.snowflake.com/en/sql-reference/sql/create-stream>
+pub enum StreamSourceKind {
+    /// `ON TABLE <name>`.
+    Table,
+    /// `ON VIEW <name>`.
+    View,
+}
+
+impl fmt::Display for StreamSourceKind {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        f.write_str(match self {
+            StreamSourceKind::Table => "TABLE",
+            StreamSourceKind::View => "VIEW",
+        })
+    }
+}
+
 #[derive(Debug, Copy, Clone, PartialEq, PartialOrd, Eq, Ord, Hash)]
 #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
 #[cfg_attr(feature = "visitor", derive(Visit, VisitMut))]

src/parser/mod.rs

@@ -5628,16 +5628,23 @@ impl<'a> Parser<'a> {
         })
     }
 
-    /// `CREATE [OR REPLACE] STREAM [IF NOT EXISTS] <name> ON TABLE <table>`
+    /// `CREATE [OR REPLACE] STREAM [IF NOT EXISTS] <name> ON { TABLE | VIEW } <source>`
     fn parse_create_stream(&mut self, or_replace: bool) -> Result<Statement, ParserError> {
         let if_not_exists = self.parse_keywords(&[Keyword::IF, Keyword::NOT, Keyword::EXISTS]);
         let name = self.parse_object_name(false)?;
-        self.expect_keywords(&[Keyword::ON, Keyword::TABLE])?;
+        self.expect_keyword(Keyword::ON)?;
+        let source_kind = if self.parse_keyword(Keyword::VIEW) {
+            StreamSourceKind::View
+        } else {
+            self.expect_keyword(Keyword::TABLE)?;
+            StreamSourceKind::Table
+        };
         let source_table = self.parse_object_name(false)?;
         Ok(Statement::CreateStream {
             or_replace,
             if_not_exists,
             name,
+            source_kind,
             source_table,
         })
     }

tests/sqlparser_snowflake.rs

@@ -71,6 +71,38 @@ fn parse_sf_create_secure_view_and_materialized_view() {
     }
 }
 
+#[test]
+fn parse_sf_create_stream_on_table_and_view() {
+    for (sql, expected_kind) in [
+        ("CREATE STREAM s ON TABLE t", StreamSourceKind::Table),
+        ("CREATE STREAM s ON VIEW v", StreamSourceKind::View),
+        (
+            "CREATE OR REPLACE STREAM s ON TABLE t",
+            StreamSourceKind::Table,
+        ),
+        (
+            "CREATE OR REPLACE STREAM s ON VIEW v",
+            StreamSourceKind::View,
+        ),
+        (
+            "CREATE STREAM IF NOT EXISTS s ON TABLE t",
+            StreamSourceKind::Table,
+        ),
+        (
+            "CREATE STREAM IF NOT EXISTS s ON VIEW v",
+            StreamSourceKind::View,
+        ),
+    ] {
+        match snowflake().verified_stmt(sql) {
+            Statement::CreateStream { source_kind, .. } => {
+                assert_eq!(source_kind, expected_kind);
+            }
+            _ => unreachable!(),
+        }
+        assert_eq!(snowflake().verified_stmt(sql).to_string(), sql);
+    }
+}
+
 #[test]
 fn test_snowflake_create_or_replace_table() {
     let sql = "CREATE OR REPLACE TABLE my_table (a number)";