LAV-2521: Support CREATE STREAM … CLONE (#2957)

* LAV-2521: support cloning streams

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

* LAV-2521: complete stream clone parity

Assign clone ownership to the executing role and cover reverse drop independence plus AT/BEFORE STREAM suffix rejection with Cloud snapshots.

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

* LAV-2521: use snapshot runner for clone errors

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

* LAV-2521: record clone errors with snapshot runner

Recapture the clone error envelopes through run_snapshot_test and strip transport-only source-location fields that the emulator does not emit.

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

@@ -5084,6 +5084,8 @@ pub enum Statement {
         if_not_exists: bool,
         /// Stream name.
         name: ObjectName,
+        /// Whether this is `CREATE STREAM <name> CLONE <source>`.
+        clone: bool,
         /// 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
@@ -8539,14 +8541,20 @@ impl fmt::Display for Statement {
                 or_replace,
                 if_not_exists,
                 name,
+                clone,
                 source_kind,
                 source_table,
                 at_before,
                 append_only,
             } => {
+                let source_prefix = if *clone {
+                    "CLONE ".to_string()
+                } else {
+                    format!("ON {source_kind} ")
+                };
                 write!(
                     f,
-                    "CREATE {or_replace}STREAM {if_not_exists}{name} ON {source_kind} {source_table}",
+                    "CREATE {or_replace}STREAM {if_not_exists}{name} {source_prefix}{source_table}",
                     or_replace = if *or_replace { "OR REPLACE " } else { "" },
                     if_not_exists = if *if_not_exists { "IF NOT EXISTS " } else { "" },
                 )?;

src/parser/mod.rs

@@ -5705,12 +5705,16 @@ 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)?;
-        self.expect_keyword(Keyword::ON)?;
-        let source_kind = if self.parse_keyword(Keyword::VIEW) {
+        let clone = self.parse_keyword(Keyword::CLONE);
+        if !clone {
+            self.expect_keyword(Keyword::ON)?;
+        }
+        let source_kind = if clone || self.parse_keyword(Keyword::TABLE) {
+            StreamSourceKind::Table
+        } else if self.parse_keyword(Keyword::VIEW) {
             StreamSourceKind::View
         } else {
-            self.expect_keyword(Keyword::TABLE)?;
-            StreamSourceKind::Table
+            return self.expected("TABLE or VIEW", self.peek_token());
         };
         let source_table = self.parse_object_name(false)?;
         // Optional `{ AT | BEFORE } ( <key> => <expr> )` clause, kept whole as
@@ -5739,6 +5743,7 @@ impl<'a> Parser<'a> {
             or_replace,
             if_not_exists,
             name,
+            clone,
             source_kind,
             source_table,
             at_before,