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

@@ -1080,7 +1080,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

@@ -2101,7 +2101,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,
@@ -647,6 +648,7 @@ impl Coordinator {
                     collections,
                     resolved_ids,
                     dependencies,
+                    replacement_target,
                     cluster_id,
                     non_null_assertions,
                     custom_logical_compaction_window: compaction_window,
@@ -687,17 +689,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 +727,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;

src/catalog/src/memory/objects.rs

@@ -1389,6 +1389,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

@@ -1813,6 +1813,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/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,
 };
@@ -2720,6 +2720,27 @@ pub fn plan_create_materialized_view(
     scx: &StatementContext,
     mut stmt: CreateMaterializedViewStatement<Aug>,
 ) -> Result<Plan, PlanError> {
+    // 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 {
+            sql_bail!(
+                "cannot replace {} {} because it is not a materialized view",
+                target.item_type(),
+                scx.catalog.minimal_qualification(target.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 cluster_id =
         crate::plan::statement::resolve_cluster_for_materialized_view(scx.catalog, &stmt)?;
     stmt.in_cluster = Some(ResolvedClusterName {
@@ -2962,12 +2983,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 +3012,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 +3264,7 @@ pub fn plan_create_continual_task(
             expr,
             dependencies,
             column_names,
+            replacement_target: None,
             cluster_id,
             non_null_assertions: Vec::new(),
             compaction_window: None,