Snowflake: parse SHOW STREAMS / DESCRIBE STREAM / ALTER STREAM COMMENT

Adds ShowStreams and AlterStream statements, DescribeObjectType::Stream,
the STREAMS keyword, and an EOF-anchored error for the invalid
DESCRIBE TABLE STREAM <name> shape (matching real Snowflake).

Author: sabir-akhadov-localstack

src/ast/mod.rs

@@ -5011,6 +5011,29 @@ pub enum Statement {
         show_options: ShowStatementOptions,
     },
     /// ```sql
+    /// SHOW [TERSE] STREAMS [LIKE '<pattern>']
+    ///     [IN { ACCOUNT | DATABASE <db> | SCHEMA <schema> }]
+    /// ```
+    /// See <https://docs.snowflake.com/en/sql-reference/sql/show-streams>
+    ShowStreams {
+        /// `true` when terse output format was requested.
+        terse: bool,
+        /// Additional options for `SHOW` statements.
+        show_options: ShowStatementOptions,
+    },
+    /// ```sql
+    /// ALTER STREAM [IF EXISTS] <name> { SET COMMENT = '<string>' | UNSET COMMENT }
+    /// ```
+    /// See <https://docs.snowflake.com/en/sql-reference/sql/alter-stream>
+    AlterStream {
+        /// `IF EXISTS` flag.
+        if_exists: bool,
+        /// Stream name.
+        name: ObjectName,
+        /// The alter action.
+        operation: AlterStreamOperation,
+    },
+    /// ```sql
     /// CREATE [OR REPLACE] EXTERNAL VOLUME [IF NOT EXISTS] <name>
     /// ```
     /// See <https://docs.snowflake.com/en/sql-reference/sql/create-external-volume>
@@ -7353,6 +7376,28 @@ impl fmt::Display for Statement {
                 )?;
                 Ok(())
             }
+            Statement::ShowStreams {
+                terse,
+                show_options,
+            } => {
+                write!(
+                    f,
+                    "SHOW {terse}STREAMS{show_options}",
+                    terse = if *terse { "TERSE " } else { "" },
+                )?;
+                Ok(())
+            }
+            Statement::AlterStream {
+                if_exists,
+                name,
+                operation,
+            } => {
+                write!(f, "ALTER STREAM")?;
+                if *if_exists {
+                    write!(f, " IF EXISTS")?;
+                }
+                write!(f, " {name} {operation}")
+            }
             Statement::CreateExternalVolume {
                 or_replace,
                 if_not_exists,
@@ -10337,6 +10382,8 @@ pub enum DescribeObjectType {
     Task,
     /// `STAGE`
     Stage,
+    /// `STREAM`
+    Stream,
 }
 
 impl fmt::Display for DescribeObjectType {
@@ -10349,6 +10396,7 @@ impl fmt::Display for DescribeObjectType {
             DescribeObjectType::Schema => "SCHEMA",
             DescribeObjectType::Task => "TASK",
             DescribeObjectType::Stage => "STAGE",
+            DescribeObjectType::Stream => "STREAM",
         })
     }
 }
@@ -13021,6 +13069,28 @@ impl fmt::Display for AlterTaskAction {
     }
 }
 
+/// Action for [`Statement::AlterStream`].
+///
+/// See <https://docs.snowflake.com/en/sql-reference/sql/alter-stream>.
+#[derive(Debug, Clone, PartialEq, PartialOrd, Eq, Ord, Hash)]
+#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+#[cfg_attr(feature = "visitor", derive(Visit, VisitMut))]
+pub enum AlterStreamOperation {
+    /// `SET COMMENT = '<string>'`
+    SetComment(String),
+    /// `UNSET COMMENT`
+    UnsetComment,
+}
+
+impl fmt::Display for AlterStreamOperation {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        match self {
+            AlterStreamOperation::SetComment(value) => write!(f, "SET COMMENT = '{value}'"),
+            AlterStreamOperation::UnsetComment => write!(f, "UNSET COMMENT"),
+        }
+    }
+}
+
 /// The `CATALOG_SOURCE` of a `CATALOG INTEGRATION`.
 #[derive(Debug, Clone, PartialEq, PartialOrd, Eq, Ord, Hash)]
 #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]

src/ast/spans.rs

@@ -542,8 +542,10 @@ impl Spanned for Statement {
             Statement::ShowAccounts { .. } => Span::empty(),
             Statement::CreateTask { .. } => Span::empty(),
             Statement::AlterTask { .. } => Span::empty(),
+            Statement::AlterStream { .. } => Span::empty(),
             Statement::ExecuteTask { .. } => Span::empty(),
             Statement::ShowTasks { .. } => Span::empty(),
+            Statement::ShowStreams { .. } => Span::empty(),
             Statement::CreateExternalVolume { .. } => Span::empty(),
             Statement::AlterExternalVolume { .. } => Span::empty(),
             Statement::DropExternalVolume { .. } => Span::empty(),

src/keywords.rs

@@ -1064,6 +1064,7 @@ define_keywords!(
     STORED,
     STRAIGHT_JOIN,
     STREAM,
+    STREAMS,
     STRICT,
     STRING,
     STRUCT,

src/parser/mod.rs

@@ -11459,6 +11459,7 @@ impl<'a> Parser<'a> {
             Keyword::WAREHOUSE,
             Keyword::ACCOUNT,
             Keyword::TASK,
+            Keyword::STREAM,
         ])?;
         match object_type {
             Keyword::SCHEMA => {
@@ -11510,9 +11511,10 @@ impl<'a> Parser<'a> {
             Keyword::WAREHOUSE => self.parse_alter_warehouse(),
             Keyword::ACCOUNT => self.parse_alter_account(),
             Keyword::TASK => self.parse_alter_task(),
+            Keyword::STREAM => self.parse_alter_stream(),
             // unreachable because expect_one_of_keywords used above
             unexpected_keyword => Err(ParserError::ParserError(
-                format!("Internal parser error: expected any of {{VIEW, TYPE, COLLATION, TABLE, INDEX, FUNCTION, AGGREGATE, ROLE, POLICY, CONNECTOR, ICEBERG, SCHEMA, USER, OPERATOR, WAREHOUSE, ACCOUNT, TASK}}, got {unexpected_keyword:?}"),
+                format!("Internal parser error: expected any of {{VIEW, TYPE, COLLATION, TABLE, INDEX, FUNCTION, AGGREGATE, ROLE, POLICY, CONNECTOR, ICEBERG, SCHEMA, USER, OPERATOR, WAREHOUSE, ACCOUNT, TASK, STREAM}}, got {unexpected_keyword:?}"),
             )),
         }
     }
@@ -11849,6 +11851,27 @@ impl<'a> Parser<'a> {
         })
     }
 
+    /// Parse `ALTER STREAM [IF EXISTS] <name> { SET COMMENT = '<s>' | UNSET COMMENT }`.
+    pub fn parse_alter_stream(&mut self) -> Result<Statement, ParserError> {
+        let if_exists = self.parse_keywords(&[Keyword::IF, Keyword::EXISTS]);
+        let name = self.parse_object_name(false)?;
+        let operation = if self.parse_keyword(Keyword::SET) {
+            self.expect_keyword(Keyword::COMMENT)?;
+            self.expect_token(&Token::Eq)?;
+            AlterStreamOperation::SetComment(self.parse_literal_string()?)
+        } else if self.parse_keyword(Keyword::UNSET) {
+            self.expect_keyword(Keyword::COMMENT)?;
+            AlterStreamOperation::UnsetComment
+        } else {
+            return self.expected("SET or UNSET after ALTER STREAM", self.peek_token());
+        };
+        Ok(Statement::AlterStream {
+            if_exists,
+            name,
+            operation,
+        })
+    }
+
     /// Parse a [Statement::AlterTable]
     pub fn parse_alter_table(&mut self, iceberg: bool) -> Result<Statement, ParserError> {
         let if_exists = self.parse_keywords(&[Keyword::IF, Keyword::EXISTS]);
@@ -14880,13 +14903,23 @@ impl<'a> Parser<'a> {
                             object_name,
                         });
                     }
+                    // `DESCRIBE TABLE STREAM <name>` is not valid Snowflake
+                    // syntax (a client such as sqlglot may emit it). Real
+                    // Snowflake consumes the whole statement and then reports an
+                    // unexpected end-of-input; mirror that by draining the name
+                    // and failing at EOF rather than at the second identifier.
+                    if self.parse_keywords(&[Keyword::TABLE, Keyword::STREAM]) {
+                        let _ = self.parse_object_name(false)?;
+                        return self.expected("end of statement", self.peek_token());
+                    }
                     if let Some(kw) = self.parse_one_of_keywords(&[
                         Keyword::TABLE,
                         Keyword::VIEW,
                         Keyword::DATABASE,
                         Keyword::SCHEMA,
                         Keyword::TASK,
                         Keyword::STAGE,
+                        Keyword::STREAM,
                     ]) {
                         let object_type = match kw {
                             Keyword::TABLE => DescribeObjectType::Table,
@@ -14895,6 +14928,7 @@ impl<'a> Parser<'a> {
                             Keyword::SCHEMA => DescribeObjectType::Schema,
                             Keyword::TASK => DescribeObjectType::Task,
                             Keyword::STAGE => DescribeObjectType::Stage,
+                            Keyword::STREAM => DescribeObjectType::Stream,
                             _ => return self.expected("a describe object type", self.peek_token()),
                         };
                         let object_name = self.parse_object_name(false)?;
@@ -16403,6 +16437,8 @@ impl<'a> Parser<'a> {
             Ok(self.parse_show_tables(terse, extended, full, external)?)
         } else if self.parse_keyword(Keyword::TASKS) {
             Ok(self.parse_show_tasks(terse)?)
+        } else if self.parse_keyword(Keyword::STREAMS) {
+            Ok(self.parse_show_streams(terse)?)
         } else if self.parse_keywords(&[Keyword::MATERIALIZED, Keyword::VIEWS]) {
             Ok(self.parse_show_views(terse, true)?)
         } else if self.parse_keyword(Keyword::VIEWS) {
@@ -16557,6 +16593,14 @@ impl<'a> Parser<'a> {
         })
     }
 
+    fn parse_show_streams(&mut self, terse: bool) -> Result<Statement, ParserError> {
+        let show_options = self.parse_show_stmt_options()?;
+        Ok(Statement::ShowStreams {
+            terse,
+            show_options,
+        })
+    }
+
     fn parse_show_views(
         &mut self,
         terse: bool,