LAV-2522: Support COPY GRANTS for CREATE STREAM (#2967)

* LAV-2522: support COPY GRANTS for streams

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

* LAV-2522: reproject copied stream grants to database roles

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

@@ -5100,6 +5100,8 @@ pub enum Statement {
         /// The validated `APPEND_ONLY` property, following the `{ AT | BEFORE }`
         /// clause. `None` when omitted (which Snowflake treats as `FALSE`).
         append_only: Option<bool>,
+        /// Whether eligible grants are copied from the replaced or cloned stream.
+        copy_grants: bool,
     },
     /// ```sql
     /// ALTER WAREHOUSE [IF EXISTS] [<name>] <operation>
@@ -8546,6 +8548,7 @@ impl fmt::Display for Statement {
                 source_table,
                 at_before,
                 append_only,
+                copy_grants,
             } => {
                 let source_prefix = if *clone {
                     "CLONE ".to_string()
@@ -8568,6 +8571,9 @@ impl fmt::Display for Statement {
                         if *append_only { "TRUE" } else { "FALSE" }
                     )?;
                 }
+                if *copy_grants {
+                    write!(f, " COPY GRANTS")?;
+                }
                 Ok(())
             }
             Statement::AlterWarehouse {

src/parser/mod.rs

@@ -5705,7 +5705,8 @@ impl<'a> Parser<'a> {
     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)?;
-        let clone = self.parse_keyword(Keyword::CLONE);
+        let leading_copy_grants = self.parse_keywords(&[Keyword::COPY, Keyword::GRANTS]);
+        let clone = !leading_copy_grants && self.parse_keyword(Keyword::CLONE);
         if !clone {
             self.expect_keyword(Keyword::ON)?;
         }
@@ -5717,6 +5718,8 @@ impl<'a> Parser<'a> {
             return self.expected("TABLE or VIEW", self.peek_token());
         };
         let source_table = self.parse_object_name(false)?;
+        let copy_grants = leading_copy_grants
+            || self.parse_keywords(&[Keyword::COPY, Keyword::GRANTS]);
         // Optional `{ AT | BEFORE } ( <key> => <expr> )` clause, kept whole as
         // the function-call expression (the same shape `TableVersion::Function`
         // uses for a table-version anchor).
@@ -5748,6 +5751,7 @@ impl<'a> Parser<'a> {
             source_table,
             at_before,
             append_only,
+            copy_grants,
         })
     }