sql-parser: CREATE REPLACEMENT MATERIALIZED VIEW

This commit changes the SQL syntax for replacement materialized views from
  `CREATE MATERIALIZED VIEW <name> REPLACING <target> AS <query>`
to
  `CREATE REPLACEMENT MATERIALIZED VIEW <name> FOR <target> AS <query>`

Author: teskje

doc/developer/design/20251111_replacement_materialized_views.md

@@ -62,7 +62,7 @@ A replacement materialized view allows users to stage the change of definition a
 The user can then inspect the replacement, and decide to apply or discard it.
 
 We add the following SQL syntax:
-* `CREATE MATERIALIZED VIEW replacement_name REPLACING mv_name AS SELECT ...`
+* `CREATE REPLACEMENT MATERIALIZED VIEW replacement_name FOR mv_name AS SELECT ...`
     Creates a replacement for the specified materialized view with the new definition.
     The usual properties for materialized views apply, such as the cluster and its options.
 * `ALTER MATERIALIZED VIEW mv_name APPLY REPLACEMENT replacement_name`
@@ -225,10 +225,3 @@ process, you are responsible for getting answers to these open
 questions. All open questions should be answered by the time a design
 document is merged.
 -->
-
-### SQL syntax
-
-The current proposal uses the term `REPLACING` in the `CREATE MATERIALIZED VIEW` statement.
-An alternative is to use `AS REPLACEMENT FOR mv_name`, which might be more explicit.
-We could also consider the term `REPLACE`, but that might be confused with `CREATE OR REPLACE`.
-Alternatively, we could use `REPLACES` instead of `REPLACING`.

misc/python/materialize/parallel_workload/action.py

@@ -1048,7 +1048,7 @@ def run(self, exe: Executor) -> bool:
 
         tmp_mv = identifier(view.name() + "_" + threading.current_thread().getName())
         exe.execute(
-            f"CREATE MATERIALIZED VIEW {tmp_mv} REPLACING {identifier(view.name())} AS {view.get_select()}",
+            f"CREATE REPLACEMENT MATERIALIZED VIEW {tmp_mv} FOR {identifier(view.name())} AS {view.get_select()}",
         )
         time.sleep(self.rng.random())
         if self.rng.choice([True, False]):

src/adapter/src/coord/command_handler.rs

@@ -1381,7 +1381,7 @@ impl Coordinator {
                         if_exists: cmvs.if_exists,
                         name: cmvs.name,
                         columns: cmvs.columns,
-                        replacing: cmvs.replacing,
+                        replacement_for: cmvs.replacement_for,
                         in_cluster: cmvs.in_cluster,
                         query: cmvs.query,
                         with_options: cmvs.with_options,

src/catalog/src/durable/objects.rs

@@ -1333,9 +1333,9 @@ pub fn item_type(create_sql: &str) -> CatalogItemType {
     let mut tokens = create_sql.split_whitespace();
     assert_eq!(tokens.next(), Some("CREATE"));
 
-    // Read away TEMPORARY, if any.
+    // Read away item type modifiers, if any.
     let next_token = match tokens.next() {
-        Some("TEMPORARY") => tokens.next(),
+        Some("TEMPORARY") | Some("REPLACEMENT") => tokens.next(),
         token => token,
     };
 

src/catalog/src/memory/objects.rs

@@ -1474,7 +1474,7 @@ impl MaterializedView {
             if_exists: old_stmt.if_exists,
             name: old_stmt.name,
             columns: rpl_stmt.columns,
-            replacing: None,
+            replacement_for: None,
             in_cluster: rpl_stmt.in_cluster,
             query: rpl_stmt.query,
             as_of: rpl_stmt.as_of,

src/sql-parser/src/ast/defs/statement.rs

@@ -1387,7 +1387,7 @@ pub struct CreateMaterializedViewStatement<T: AstInfo> {
     pub if_exists: IfExistsBehavior,
     pub name: UnresolvedItemName,
     pub columns: Vec<Ident>,
-    pub replacing: Option<T::ItemName>,
+    pub replacement_for: Option<T::ItemName>,
     pub in_cluster: Option<T::ClusterName>,
     pub query: Query<T>,
     pub as_of: Option<u64>,
@@ -1400,6 +1400,9 @@ impl<T: AstInfo> AstDisplay for CreateMaterializedViewStatement<T> {
         if self.if_exists == IfExistsBehavior::Replace {
             f.write_str(" OR REPLACE");
         }
+        if self.replacement_for.is_some() {
+            f.write_str(" REPLACEMENT");
+        }
 
         f.write_str(" MATERIALIZED VIEW");
 
@@ -1416,8 +1419,8 @@ impl<T: AstInfo> AstDisplay for CreateMaterializedViewStatement<T> {
             f.write_str(")");
         }
 
-        if let Some(target) = &self.replacing {
-            f.write_str(" REPLACING ");
+        if let Some(target) = &self.replacement_for {
+            f.write_str(" FOR ");
             f.write_node(target);
         }
 

src/sql-parser/src/parser.rs

@@ -1959,6 +1959,8 @@ impl<'a> Parser<'a> {
                 .map_parser_err(StatementKind::CreateConnection)
         } else if self.peek_keywords(&[MATERIALIZED, VIEW])
             || self.peek_keywords(&[OR, REPLACE, MATERIALIZED, VIEW])
+            || self.peek_keywords(&[REPLACEMENT, MATERIALIZED, VIEW])
+            || self.peek_keywords(&[OR, REPLACE, REPLACEMENT, MATERIALIZED, VIEW])
         {
             self.parse_create_materialized_view()
                 .map_parser_err(StatementKind::CreateMaterializedView)
@@ -3855,17 +3857,20 @@ impl<'a> Parser<'a> {
         } else {
             IfExistsBehavior::Error
         };
+        let replacement = self.parse_keyword(REPLACEMENT);
         self.expect_keywords(&[MATERIALIZED, VIEW])?;
         if if_exists == IfExistsBehavior::Error && self.parse_if_not_exists()? {
             if_exists = IfExistsBehavior::Skip;
         }
 
         let name = self.parse_item_name()?;
         let columns = self.parse_parenthesized_column_list(Optional)?;
-        let replacing = self
-            .parse_keyword(REPLACING)
-            .then(|| self.parse_raw_name())
-            .transpose()?;
+        let replacement_for = if replacement {
+            self.expect_keyword(FOR)?;
+            Some(self.parse_raw_name()?)
+        } else {
+            None
+        };
         let in_cluster = self.parse_optional_in_cluster()?;
 
         let with_options = if self.parse_keyword(WITH) {
@@ -3886,7 +3891,7 @@ impl<'a> Parser<'a> {
                 if_exists,
                 name,
                 columns,
-                replacing,
+                replacement_for,
                 in_cluster,
                 query,
                 as_of,