adapter,sql: plan and sequence replacement MV creation

Author: teskje

src/adapter/src/catalog/state.rs

@@ -1375,6 +1375,7 @@ impl CatalogState {
                     desc,
                     resolved_ids,
                     dependencies,
+                    replacement_target: materialized_view.replacement_target,
                     cluster_id: materialized_view.cluster_id,
                     non_null_assertions: materialized_view.non_null_assertions,
                     custom_logical_compaction_window: materialized_view.compaction_window,

src/adapter/src/catalog/transact.rs

@@ -1130,7 +1130,15 @@ impl Catalog {
                         storage_collections_to_create.insert(source.global_id());
                     }
                     CatalogItem::MaterializedView(mv) => {
-                        storage_collections_to_create.insert(mv.global_id_writes());
+                        let mv_gid = mv.global_id_writes();
+                        if let Some(target_id) = mv.replacement_target {
+                            let target_gid = state.get_entry(&target_id).latest_global_id();
+                            let shard_id =
+                                state.storage_metadata().get_collection_shard(target_gid)?;
+                            storage_collections_to_register.insert(mv_gid, shard_id);
+                        } else {
+                            storage_collections_to_create.insert(mv_gid);
+                        }
                     }
                     CatalogItem::ContinualTask(ct) => {
                         storage_collections_to_create.insert(ct.global_id());

src/adapter/src/continual_task.rs

@@ -39,6 +39,7 @@ pub fn ct_item_from_plan(
                 expr: mut raw_expr,
                 dependencies,
                 column_names: _,
+                replacement_target: _,
                 non_null_assertions: _,
                 compaction_window: _,
                 refresh_schedule: _,

src/adapter/src/coord.rs

@@ -2117,7 +2117,12 @@ impl Coordinator {
                     }
 
                     self.ship_dataflow(df_desc, mview.cluster_id, None).await;
-                    self.allow_writes(mview.cluster_id, mview.global_id_writes());
+
+                    // If this is a replacement MV, it must remain read-only until the replacement
+                    // gets applied.
+                    if mview.replacement_target.is_none() {
+                        self.allow_writes(mview.cluster_id, mview.global_id_writes());
+                    }
                 }
                 CatalogItem::Sink(sink) => {
                     policies_to_set

src/adapter/src/coord/sequencer/inner/create_materialized_view.rs

@@ -561,6 +561,7 @@ impl Coordinator {
                             mut create_sql,
                             expr: raw_expr,
                             dependencies,
+                            replacement_target,
                             cluster_id,
                             non_null_assertions,
                             compaction_window,
@@ -577,6 +578,23 @@ impl Coordinator {
             global_lir_plan,
             ..
         } = stage;
+
+        // Validate the replacement target, if one is given.
+        // TODO(alter-mv): Could we do this already in planning?
+        if let Some(target_id) = replacement_target {
+            let Some(target) = self.catalog().get_entry(&target_id).materialized_view() else {
+                return Err(AdapterError::internal(
+                    "create materialized view",
+                    "replacement target not a materialized view",
+                ));
+            };
+
+            // For now, we don't support schema evolution for materialized views.
+            if &target.desc.latest() != global_lir_plan.desc() {
+                return Err(AdapterError::Unstructured(anyhow!("incompatible schemas")));
+            }
+        }
+
         // Timestamp selection
         let id_bundle = dataflow_import_id_bundle(global_lir_plan.df_desc(), cluster_id);
 
@@ -593,8 +611,12 @@ impl Coordinator {
             &read_holds_owned
         };
 
-        let (dataflow_as_of, storage_as_of, until) =
-            self.select_timestamps(id_bundle, refresh_schedule.as_ref(), read_holds)?;
+        let (dataflow_as_of, storage_as_of, until) = self.select_timestamps(
+            id_bundle,
+            refresh_schedule.as_ref(),
+            read_holds,
+            replacement_target,
+        )?;
 
         tracing::info!(
             dataflow_as_of = ?dataflow_as_of,
@@ -647,6 +669,7 @@ impl Coordinator {
                     collections,
                     resolved_ids,
                     dependencies,
+                    replacement_target,
                     cluster_id,
                     non_null_assertions,
                     custom_logical_compaction_window: compaction_window,
@@ -687,17 +710,26 @@ impl Coordinator {
 
                     let storage_metadata = coord.catalog.state().storage_metadata();
 
+                    let mut collection_desc =
+                        CollectionDescription::for_other(output_desc, Some(storage_as_of));
+                    let mut allow_writes = true;
+
+                    // If this MV is intended to replace another one, we need to start it in
+                    // read-only mode, targeting the shard of the replacement target.
+                    if let Some(target_id) = replacement_target {
+                        let target_gid = coord.catalog.get_entry(&target_id).latest_global_id();
+                        collection_desc.primary = Some(target_gid);
+                        allow_writes = false;
+                    }
+
                     // Announce the creation of the materialized view source.
                     coord
                         .controller
                         .storage
                         .create_collections(
                             storage_metadata,
                             None,
-                            vec![(
-                                global_id,
-                                CollectionDescription::for_other(output_desc, Some(storage_as_of)),
-                            )],
+                            vec![(global_id, collection_desc)],
                         )
                         .await
                         .unwrap_or_terminate("cannot fail to append");
@@ -716,7 +748,10 @@ impl Coordinator {
                             notice_builtin_updates_fut,
                         )
                         .await;
-                    coord.allow_writes(cluster_id, global_id);
+
+                    if allow_writes {
+                        coord.allow_writes(cluster_id, global_id);
+                    }
                 })
             })
             .await;
@@ -746,6 +781,7 @@ impl Coordinator {
         id_bundle: CollectionIdBundle,
         refresh_schedule: Option<&RefreshSchedule>,
         read_holds: &ReadHolds<mz_repr::Timestamp>,
+        replacement_target: Option<CatalogItemId>,
     ) -> Result<
         (
             Antichain<mz_repr::Timestamp>,
@@ -809,6 +845,24 @@ impl Coordinator {
             .and_then(|r| r.try_step_forward());
         let until = Antichain::from_iter(until_ts);
 
+        // If this is a replacement MV, ensure that `storage_as_of` > the `since` of the target
+        // storage collection. The storage controller requires the `since` of a storage collection
+        // to always be greater than the `since`s of the collections it depends on.
+        if let Some(target_id) = replacement_target {
+            let target_gid = self.catalog().get_entry(&target_id).latest_global_id();
+            let frontiers = self
+                .controller
+                .storage_collections
+                .collection_frontiers(target_gid)
+                .expect("replacement target exists");
+            let lower_bound = frontiers
+                .read_capabilities
+                .iter()
+                .map(|t| t.step_forward())
+                .collect();
+            storage_as_of.join_assign(&lower_bound);
+        }
+
         Ok((dataflow_as_of, storage_as_of, until))
     }
 

src/catalog/src/memory/objects.rs

@@ -1390,6 +1390,8 @@ pub struct MaterializedView {
     pub resolved_ids: ResolvedIds,
     /// All of the catalog objects that are referenced by this view.
     pub dependencies: DependencyIds,
+    /// ID of the materialized view this materialized view is intended to replace.
+    pub replacement_target: Option<CatalogItemId>,
     /// Cluster that this materialized view runs on.
     pub cluster_id: ClusterId,
     /// Column indexes that we assert are not `NULL`.

src/sql/src/plan.rs

@@ -1816,6 +1816,7 @@ pub struct MaterializedView {
     pub dependencies: DependencyIds,
     /// Columns of this view.
     pub column_names: Vec<ColumnName>,
+    pub replacement_target: Option<CatalogItemId>,
     /// Cluster this materialized view will get installed on.
     pub cluster_id: ClusterId,
     pub non_null_assertions: Vec<usize>,

src/sql/src/plan/error.rs

@@ -305,6 +305,12 @@ pub enum PlanError {
     },
     /// AS OF or UP TO should be an expression that is castable and simplifiable to a non-null mz_timestamp value.
     InvalidAsOfUpTo,
+    InvalidReplacement {
+        item_type: CatalogItemType,
+        item_name: PartialItemName,
+        replacement_type: CatalogItemType,
+        replacement_name: PartialItemName,
+    },
     // TODO(benesch): eventually all errors should be structured.
     Unstructured(String),
 }
@@ -845,7 +851,10 @@ impl fmt::Display for PlanError {
                 write!(f, "cursor {} does not exist", name.quoted())
             }
             Self::CopyFromTargetTableDropped { target_name: name } => write!(f, "COPY FROM's target table {} was dropped", name.quoted()),
-            Self::InvalidAsOfUpTo => write!(f, "AS OF or UP TO should be castable to a (non-null) mz_timestamp value")
+            Self::InvalidAsOfUpTo => write!(f, "AS OF or UP TO should be castable to a (non-null) mz_timestamp value"),
+            Self::InvalidReplacement { item_type, item_name, replacement_type, replacement_name } => {
+                write!(f, "cannot replace {item_type} {item_name} with {replacement_type} {replacement_name}")
+            }
         }
     }
 }

src/sql/src/plan/statement/ddl.rs

@@ -130,7 +130,7 @@ use crate::catalog::{
 use crate::iceberg::IcebergSinkConfigOptionExtracted;
 use crate::kafka_util::{KafkaSinkConfigOptionExtracted, KafkaSourceConfigOptionExtracted};
 use crate::names::{
-    Aug, CommentObjectId, DatabaseId, ObjectId, PartialItemName, QualifiedItemName,
+    Aug, CommentObjectId, DatabaseId, DependencyIds, ObjectId, PartialItemName, QualifiedItemName,
     ResolvedClusterName, ResolvedColumnReference, ResolvedDataType, ResolvedDatabaseSpecifier,
     ResolvedItemName, ResolvedNetworkPolicyName, SchemaSpecifier, SystemObjectId,
 };
@@ -2733,6 +2733,28 @@ pub fn plan_create_materialized_view(
     let partial_name = normalize::unresolved_item_name(stmt.name)?;
     let name = scx.allocate_qualified_name(partial_name.clone())?;
 
+    // Validate the replacement target, if one is given.
+    let replacement_target = if let Some(target_name) = &stmt.replacing {
+        let target = scx.get_item_by_resolved_name(target_name)?;
+        if target.item_type() != CatalogItemType::MaterializedView {
+            return Err(PlanError::InvalidReplacement {
+                item_type: target.item_type(),
+                item_name: scx.catalog.minimal_qualification(target.name()),
+                replacement_type: CatalogItemType::MaterializedView,
+                replacement_name: partial_name,
+            });
+        }
+        if target.id().is_system() {
+            sql_bail!(
+                "cannot replace {} because it is required by the database system",
+                scx.catalog.minimal_qualification(target.name()),
+            );
+        }
+        Some(target.id())
+    } else {
+        None
+    };
+
     let query::PlannedRootQuery {
         expr,
         mut desc,
@@ -2962,12 +2984,16 @@ pub fn plan_create_materialized_view(
                 .collect()
         })
         .unwrap_or_default();
-    let dependencies = expr
+    let mut dependencies: BTreeSet<_> = expr
         .depends_on()
         .into_iter()
         .map(|gid| scx.catalog.resolve_item_id(&gid))
         .collect();
 
+    if let Some(id) = replacement_target {
+        dependencies.insert(id);
+    }
+
     // Check for an object in the catalog with this same name
     let full_name = scx.catalog.resolve_full_name(&name);
     let partial_name = PartialItemName::from(full_name.clone());
@@ -2987,8 +3013,9 @@ pub fn plan_create_materialized_view(
         materialized_view: MaterializedView {
             create_sql,
             expr,
-            dependencies,
+            dependencies: DependencyIds(dependencies),
             column_names,
+            replacement_target,
             cluster_id,
             non_null_assertions,
             compaction_window,
@@ -3238,6 +3265,7 @@ pub fn plan_create_continual_task(
             expr,
             dependencies,
             column_names,
+            replacement_target: None,
             cluster_id,
             non_null_assertions: Vec::new(),
             compaction_window: None,