LAV-2523: Support CREATE OR ALTER STREAM (#3002)

* LAV-2523: support create or alter table streams

Preserve existing stream offsets while enforcing immutable source, APPEND_ONLY, and SHOW_INITIAL_ROWS settings, and apply COMMENT create-or-alter semantics. Add Cloud-captured coverage for creation, mutation, duplicate properties, collisions, and unsupported clauses.\n\n## Deviations\n\n- Masked volatile SHOW/DESCRIBE owner and stale_after fields using the established test_append_only_show_and_describe masks in tests/queries/test_streams.py; snapshots were captured against Snowflake Cloud.

Co-authored-by: Sabir Akhadov <sabir.akhadov@localstack.cloud>

* LAV-2523: preserve COPY GRANTS error precedence

Report the clause-order-specific unexpected token for CREATE OR ALTER STREAM so both COPY GRANTS placements match Snowflake.

Co-authored-by: Sabir Akhadov <sabir.akhadov@localstack.cloud>

* LAV-2523: derive COPY GRANTS error positions

Derive CREATE OR ALTER STREAM COPY GRANTS error metadata from the COPY, GRANTS, and ON token spans. Add Cloud-captured coverage for varying identifier lengths and multiline statements.

Co-authored-by: Sabir Akhadov <sabir.akhadov@localstack.cloud>

---------

Co-authored-by: spiral <spiral@localhost>
Co-authored-by: Sabir Akhadov <sabir.akhadov@localstack.cloud>

Author: 3 people

src/ast/mod.rs

@@ -5080,6 +5080,8 @@ pub enum Statement {
     CreateStream {
         /// `OR REPLACE` flag.
         or_replace: bool,
+        /// `OR ALTER` flag.
+        or_alter: bool,
         /// `IF NOT EXISTS` flag.
         if_not_exists: bool,
         /// Stream name.
@@ -8567,6 +8569,7 @@ impl fmt::Display for Statement {
             }
             Statement::CreateStream {
                 or_replace,
+                or_alter,
                 if_not_exists,
                 name,
                 clone,
@@ -8584,8 +8587,14 @@ impl fmt::Display for Statement {
                 };
                 write!(
                     f,
-                    "CREATE {or_replace}STREAM {if_not_exists}{name} {source_prefix}{source_table}",
-                    or_replace = if *or_replace { "OR REPLACE " } else { "" },
+                    "CREATE {modifier}STREAM {if_not_exists}{name} {source_prefix}{source_table}",
+                    modifier = if *or_replace {
+                        "OR REPLACE "
+                    } else if *or_alter {
+                        "OR ALTER "
+                    } else {
+                        ""
+                    },
                     if_not_exists = if *if_not_exists { "IF NOT EXISTS " } else { "" },
                 )?;
                 if let Some(at_before) = at_before {

src/parser/mod.rs

@@ -5628,7 +5628,7 @@ impl<'a> Parser<'a> {
         } else if self.parse_keyword(Keyword::SEQUENCE) {
             self.parse_create_sequence(or_replace, or_alter, temporary)
         } else if self.parse_keyword(Keyword::STREAM) {
-            self.parse_create_stream(or_replace)
+            self.parse_create_stream(or_replace, or_alter)
         } else if self.parse_keyword(Keyword::PIPE) {
             self.parse_create_pipe(or_replace)
         } else if or_replace {
@@ -5702,7 +5702,11 @@ impl<'a> Parser<'a> {
 
     /// `CREATE [OR REPLACE] STREAM [IF NOT EXISTS] <name> ON { TABLE | VIEW } <source>`
     ///   `[ { AT | BEFORE } ( <key> => <expr> ) ]`
-    fn parse_create_stream(&mut self, or_replace: bool) -> Result<Statement, ParserError> {
+    fn parse_create_stream(
+        &mut self,
+        or_replace: bool,
+        or_alter: bool,
+    ) -> Result<Statement, ParserError> {
         let if_not_exists = self.parse_keywords(&[Keyword::IF, Keyword::NOT, Keyword::EXISTS]);
         let name = self.parse_object_name(false)?;
         let leading_copy_grants = self.parse_keywords(&[Keyword::COPY, Keyword::GRANTS]);
@@ -5758,6 +5762,7 @@ impl<'a> Parser<'a> {
         }
         Ok(Statement::CreateStream {
             or_replace,
+            or_alter,
             if_not_exists,
             name,
             clone,