adapter,sql: plan and sequence replacement MV application

teskje · teskje · commit 0005b5909c93 · 2025-11-26T12:06:28.000+01:00
diff --git a/src/adapter/src/catalog/transact.rs b/src/adapter/src/catalog/transact.rs
@@ -100,6 +100,10 @@ pub enum Op {
         typ: SqlColumnType,
         sql: RawDataType,
     },
+    AlterMaterializedViewApplyReplacement {
+        id: CatalogItemId,
+        replacement_id: CatalogItemId,
+    },
     CreateDatabase {
         name: String,
         owner_id: RoleId,
@@ -769,6 +773,30 @@ impl Catalog {
                 tx.update_item(id, new_entry.into())?;
                 storage_collections_to_register.insert(new_global_id, shard_id);
             }
+            Op::AlterMaterializedViewApplyReplacement { id, replacement_id } => {
+                let mut new_entry = state.get_entry(&id).clone();
+                let replacement = state.get_entry(&replacement_id);
+
+                let CatalogItem::MaterializedView(mv) = &mut new_entry.item else {
+                    return Err(AdapterError::internal(
+                        "ALTER MATERIALIZED VIEW ... APPLY REPLACEMENT",
+                        "id must refer to a materialized view",
+                    ));
+                };
+                let CatalogItem::MaterializedView(replacement_mv) = &replacement.item else {
+                    return Err(AdapterError::internal(
+                        "ALTER MATERIALIZED VIEW ... APPLY REPLACEMENT",
+                        "replacement_id must refer to a materialized view",
+                    ));
+                };
+
+                mv.apply_replacement(replacement_mv.clone());
+
+                tx.remove_item(replacement_id)?;
+                tx.update_item(id, new_entry.into())?;
+
+                // TODO(alter-mv): audit logging
+            }
             Op::CreateDatabase { name, owner_id } => {
                 let database_owner_privileges = vec![rbac::owner_privilege(
                     mz_sql::catalog::ObjectType::Database,
diff --git a/src/adapter/src/command.rs b/src/adapter/src/command.rs
@@ -718,6 +718,7 @@ impl ExecuteResponse {
             | AlterSource
             | AlterSink
             | AlterTableAddColumn
+            | AlterMaterializedViewApplyReplacement
             | AlterNetworkPolicy => &[AlteredObject],
             AlterDefaultPrivileges => &[AlteredDefaultPrivileges],
             AlterSetCluster => &[AlteredObject],
diff --git a/src/adapter/src/coord/appends.rs b/src/adapter/src/coord/appends.rs
@@ -1006,6 +1006,7 @@ pub(crate) fn waiting_on_startup_appends(
         | Plan::AlterRole(_)
         | Plan::AlterOwner(_)
         | Plan::AlterTableAddColumn(_)
+        | Plan::AlterMaterializedViewApplyReplacement(_)
         | Plan::Declare(_)
         | Plan::Fetch(_)
         | Plan::Close(_)
diff --git a/src/adapter/src/coord/catalog_serving.rs b/src/adapter/src/coord/catalog_serving.rs
@@ -128,6 +128,7 @@ pub fn auto_run_on_catalog_server<'a, 's, 'p>(
         | Plan::AlterRole(_)
         | Plan::AlterOwner(_)
         | Plan::AlterTableAddColumn(_)
+        | Plan::AlterMaterializedViewApplyReplacement(_)
         | Plan::Declare(_)
         | Plan::Fetch(_)
         | Plan::Close(_)
diff --git a/src/adapter/src/coord/ddl.rs b/src/adapter/src/coord/ddl.rs
@@ -1480,6 +1480,7 @@ impl Coordinator {
                 | Op::AlterRetainHistory { .. }
                 | Op::AlterNetworkPolicy { .. }
                 | Op::AlterAddColumn { .. }
+                | Op::AlterMaterializedViewApplyReplacement { .. }
                 | Op::UpdatePrivilege { .. }
                 | Op::UpdateDefaultPrivilege { .. }
                 | Op::GrantRole { .. }
diff --git a/src/adapter/src/coord/peek.rs b/src/adapter/src/coord/peek.rs
@@ -859,7 +859,7 @@ impl crate::coord::Coordinator {
         // If a dataflow was created, drop it once the peek command is sent.
         if let Some(index_id) = drop_dataflow {
             self.remove_compute_ids_from_timeline(vec![(compute_instance, index_id)]);
-            self.drop_indexes(vec![(compute_instance, index_id)]);
+            self.drop_compute_collections(vec![(compute_instance, index_id)]);
         }
 
         let persist_client = self.persist_client.clone();
diff --git a/src/adapter/src/coord/sequencer.rs b/src/adapter/src/coord/sequencer.rs
@@ -537,6 +537,12 @@ impl Coordinator {
                     let result = self.sequence_alter_table(&mut ctx, plan).await;
                     ctx.retire(result);
                 }
+                Plan::AlterMaterializedViewApplyReplacement(plan) => {
+                    let result = self
+                        .sequence_alter_materialized_view_apply_replacement(&mut ctx, plan)
+                        .await;
+                    ctx.retire(result);
+                }
                 Plan::AlterNetworkPolicy(plan) => {
                     let res = self
                         .sequence_alter_network_policy(ctx.session(), plan)
diff --git a/src/adapter/src/coord/sequencer/inner.rs b/src/adapter/src/coord/sequencer/inner.rs
@@ -58,7 +58,10 @@ use mz_sql::names::{
     Aug, ObjectId, QualifiedItemName, ResolvedDatabaseSpecifier, ResolvedIds, ResolvedItemName,
     SchemaSpecifier, SystemObjectId,
 };
-use mz_sql::plan::{ConnectionDetails, NetworkPolicyRule, StatementContext};
+use mz_sql::plan::{
+    AlterMaterializedViewApplyReplacementPlan, ConnectionDetails, NetworkPolicyRule,
+    StatementContext,
+};
 use mz_sql::pure::{PurifiedSourceExport, generate_subsource_statements};
 use mz_storage_types::sinks::StorageSinkDesc;
 use mz_storage_types::sources::{GenericSourceConnection, IngestionDescription, SourceExport};
@@ -5105,6 +5108,57 @@ impl Coordinator {
         Ok(ExecuteResponse::AlteredObject(ObjectType::Table))
     }
 
+    #[instrument]
+    pub(super) async fn sequence_alter_materialized_view_apply_replacement(
+        &mut self,
+        ctx: &mut ExecuteContext,
+        plan: AlterMaterializedViewApplyReplacementPlan,
+    ) -> Result<ExecuteResponse, AdapterError> {
+        const ERROR_CONTEXT: &str = "ALTER MATERIALIZED VIEW ... APPLY REPLACEMENT";
+
+        let AlterMaterializedViewApplyReplacementPlan { id, replacement_id } = plan;
+
+        // TODO(alter-mv): Wait until there is overlap between the old MV's write frontier and the
+        // new MV's as-of, to ensure no times are skipped.
+
+        let Some(old) = self.catalog().get_entry(&id).materialized_view() else {
+            return Err(AdapterError::internal(
+                ERROR_CONTEXT,
+                "id must refer to a materialized view",
+            ));
+        };
+        let Some(new) = self
+            .catalog()
+            .get_entry(&replacement_id)
+            .materialized_view()
+        else {
+            return Err(AdapterError::internal(
+                ERROR_CONTEXT,
+                "replacement_id must refer to a materialized view",
+            ));
+        };
+
+        let old_cluster_id = old.cluster_id;
+        let new_cluster_id = new.cluster_id;
+
+        let old_gid = old.global_id_writes();
+        let new_gid = new.global_id_writes();
+
+        let ops = vec![catalog::Op::AlterMaterializedViewApplyReplacement { id, replacement_id }];
+
+        self.catalog_transact_with_side_effects(Some(ctx), ops, move |coord, _ctx| {
+            Box::pin(async move {
+                // Cut over the MV computation, by shutting down the old dataflow and allowing the
+                // new dataflow to start writing.
+                coord.drop_compute_collections(vec![(old_cluster_id, old_gid)]);
+                coord.allow_writes(new_cluster_id, new_gid);
+            })
+        })
+        .await?;
+
+        Ok(ExecuteResponse::AlteredObject(ObjectType::MaterializedView))
+    }
+
     pub(super) async fn statistics_oracle(
         &self,
         session: &Session,
diff --git a/src/catalog/src/memory/objects.rs b/src/catalog/src/memory/objects.rs
@@ -752,6 +752,10 @@ impl mz_sql::catalog::CatalogItem for CatalogCollectionEntry {
         self.entry.writable_table_details()
     }
 
+    fn replacement_target(&self) -> Option<CatalogItemId> {
+        self.entry.replacement_target()
+    }
+
     fn type_details(&self) -> Option<&CatalogTypeDetails<IdReference>> {
         self.entry.type_details()
     }
@@ -1446,6 +1450,67 @@ impl MaterializedView {
         self.desc
             .at_version(RelationVersionSelector::Specific(*version))
     }
+
+    /// Apply the given replacement materialized view to this [`MaterializedView`].
+    pub fn apply_replacement(&mut self, replacement: Self) {
+        let target_id = replacement
+            .replacement_target
+            .expect("replacement has target");
+
+        fn parse(create_sql: &str) -> mz_sql::ast::CreateMaterializedViewStatement<Raw> {
+            let res = mz_sql::parse::parse(create_sql).unwrap_or_else(|e| {
+                panic!("invalid create_sql persisted in catalog: {e}\n{create_sql}");
+            });
+            if let Statement::CreateMaterializedView(cmvs) = res.into_element().ast {
+                cmvs
+            } else {
+                panic!("invalid MV create_sql persisted in catalog\n{create_sql}");
+            }
+        }
+
+        let old_stmt = parse(&self.create_sql);
+        let rpl_stmt = parse(&replacement.create_sql);
+        let new_stmt = mz_sql::ast::CreateMaterializedViewStatement {
+            if_exists: old_stmt.if_exists,
+            name: old_stmt.name,
+            columns: rpl_stmt.columns,
+            replacing: None,
+            in_cluster: rpl_stmt.in_cluster,
+            query: rpl_stmt.query,
+            as_of: rpl_stmt.as_of,
+            with_options: rpl_stmt.with_options,
+        };
+        let create_sql = new_stmt.to_ast_string_stable();
+
+        let mut collections = std::mem::take(&mut self.collections);
+        // Note: We can't use `self.desc.latest_version` here because a replacement doesn't
+        // necessary evolve the relation schema, so that version might be lower than the actual
+        // latest version.
+        let latest_version = collections.keys().max().expect("at least one version");
+        let new_version = latest_version.bump();
+        collections.insert(new_version, replacement.global_id_writes());
+
+        let mut resolved_ids = replacement.resolved_ids;
+        resolved_ids.remove_item(&target_id);
+        let mut dependencies = replacement.dependencies;
+        dependencies.0.remove(&target_id);
+
+        *self = Self {
+            create_sql,
+            collections,
+            raw_expr: replacement.raw_expr,
+            optimized_expr: replacement.optimized_expr,
+            desc: replacement.desc,
+            resolved_ids,
+            dependencies,
+            replacement_target: None,
+            cluster_id: replacement.cluster_id,
+            non_null_assertions: replacement.non_null_assertions,
+            custom_logical_compaction_window: replacement.custom_logical_compaction_window,
+            refresh_schedule: replacement.refresh_schedule,
+            initial_as_of: replacement.initial_as_of,
+        };
+    }
 }
 
 #[derive(Debug, Clone, Serialize)]
@@ -3369,6 +3434,14 @@ impl mz_sql::catalog::CatalogItem for CatalogEntry {
         }
     }
 
+    fn replacement_target(&self) -> Option<CatalogItemId> {
+        if let CatalogItem::MaterializedView(mv) = self.item() {
+            mv.replacement_target
+        } else {
+            None
+        }
+    }
+
     fn type_details(&self) -> Option<&CatalogTypeDetails<IdReference>> {
         if let CatalogItem::Type(Type { details, .. }) = self.item() {
             Some(details)
diff --git a/src/repr/src/relation.rs b/src/repr/src/relation.rs
@@ -1355,8 +1355,6 @@ impl VersionedRelationDesc {
     /// Returns this [`RelationDesc`] at the specified version.
     pub fn at_version(&self, version: RelationVersionSelector) -> RelationDesc {
         // Get all of the changes from the start, up to whatever version was requested.
-        //
-        // TODO(parkmycar): We should probably panic on unknown verisons?
         let up_to_version = match version {
             RelationVersionSelector::Latest => RelationVersion(u64::MAX),
             RelationVersionSelector::Specific(v) => v,
diff --git a/src/sql/src/catalog.rs b/src/sql/src/catalog.rs
@@ -836,6 +836,9 @@ pub trait CatalogItem {
     /// catalog item is a table that accepts writes.
     fn writable_table_details(&self) -> Option<&[Expr<Aug>]>;
 
+    /// The item this catalog item replaces, if any.
+    fn replacement_target(&self) -> Option<CatalogItemId>;
+
     /// Returns the type information associated with the catalog item, if the
     /// catalog item is a type.
     fn type_details(&self) -> Option<&CatalogTypeDetails<IdReference>>;
diff --git a/src/sql/src/plan.rs b/src/sql/src/plan.rs
@@ -193,6 +193,7 @@ pub enum Plan {
     AlterRole(AlterRolePlan),
     AlterOwner(AlterOwnerPlan),
     AlterTableAddColumn(AlterTablePlan),
+    AlterMaterializedViewApplyReplacement(AlterMaterializedViewApplyReplacementPlan),
     AlterNetworkPolicy(AlterNetworkPolicyPlan),
     Declare(DeclarePlan),
     Fetch(FetchPlan),
@@ -253,7 +254,10 @@ impl Plan {
             StatementKind::AlterTableAddColumn => {
                 &[PlanKind::AlterNoop, PlanKind::AlterTableAddColumn]
             }
-            StatementKind::AlterMaterializedViewApplyReplacement => todo!(),
+            StatementKind::AlterMaterializedViewApplyReplacement => &[
+                PlanKind::AlterNoop,
+                PlanKind::AlterMaterializedViewApplyReplacement,
+            ],
             StatementKind::Close => &[PlanKind::Close],
             StatementKind::Comment => &[PlanKind::Comment],
             StatementKind::Commit => &[PlanKind::CommitTransaction],
@@ -446,6 +450,9 @@ impl Plan {
                 ObjectType::NetworkPolicy => "alter network policy owner",
             },
             Plan::AlterTableAddColumn(_) => "alter table add column",
+            Plan::AlterMaterializedViewApplyReplacement(_) => {
+                "alter materialized view apply replacement"
+            }
             Plan::Declare(_) => "declare",
             Plan::Fetch(_) => "fetch",
             Plan::Close(_) => "close",
@@ -1325,6 +1332,12 @@ pub struct AlterTablePlan {
     pub raw_sql_type: RawDataType,
 }
 
+#[derive(Debug)]
+pub struct AlterMaterializedViewApplyReplacementPlan {
+    pub id: CatalogItemId,
+    pub replacement_id: CatalogItemId,
+}
+
 #[derive(Debug)]
 pub struct DeclarePlan {
     pub name: String,
diff --git a/src/sql/src/plan/statement/ddl.rs b/src/sql/src/plan/statement/ddl.rs
@@ -148,21 +148,21 @@ use crate::plan::with_options::{OptionalDuration, OptionalString, TryFromValue};
 use crate::plan::{
     AlterClusterPlan, AlterClusterPlanStrategy, AlterClusterRenamePlan,
     AlterClusterReplicaRenamePlan, AlterClusterSwapPlan, AlterConnectionPlan, AlterItemRenamePlan,
-    AlterNetworkPolicyPlan, AlterNoopPlan, AlterOptionParameter, AlterRetainHistoryPlan,
-    AlterRolePlan, AlterSchemaRenamePlan, AlterSchemaSwapPlan, AlterSecretPlan,
-    AlterSetClusterPlan, AlterSinkPlan, AlterSystemResetAllPlan, AlterSystemResetPlan,
-    AlterSystemSetPlan, AlterTablePlan, ClusterSchedule, CommentPlan, ComputeReplicaConfig,
-    ComputeReplicaIntrospectionConfig, ConnectionDetails, CreateClusterManagedPlan,
-    CreateClusterPlan, CreateClusterReplicaPlan, CreateClusterUnmanagedPlan, CreateClusterVariant,
-    CreateConnectionPlan, CreateContinualTaskPlan, CreateDatabasePlan, CreateIndexPlan,
-    CreateMaterializedViewPlan, CreateNetworkPolicyPlan, CreateRolePlan, CreateSchemaPlan,
-    CreateSecretPlan, CreateSinkPlan, CreateSourcePlan, CreateTablePlan, CreateTypePlan,
-    CreateViewPlan, DataSourceDesc, DropObjectsPlan, DropOwnedPlan, HirRelationExpr, Index,
-    MaterializedView, NetworkPolicyRule, NetworkPolicyRuleAction, NetworkPolicyRuleDirection, Plan,
-    PlanClusterOption, PlanNotice, PolicyAddress, QueryContext, ReplicaConfig, Secret, Sink,
-    Source, Table, TableDataSource, Type, VariableValue, View, WebhookBodyFormat,
-    WebhookHeaderFilters, WebhookHeaders, WebhookValidation, literal, plan_utils, query,
-    transform_ast,
+    AlterMaterializedViewApplyReplacementPlan, AlterNetworkPolicyPlan, AlterNoopPlan,
+    AlterOptionParameter, AlterRetainHistoryPlan, AlterRolePlan, AlterSchemaRenamePlan,
+    AlterSchemaSwapPlan, AlterSecretPlan, AlterSetClusterPlan, AlterSinkPlan,
+    AlterSystemResetAllPlan, AlterSystemResetPlan, AlterSystemSetPlan, AlterTablePlan,
+    ClusterSchedule, CommentPlan, ComputeReplicaConfig, ComputeReplicaIntrospectionConfig,
+    ConnectionDetails, CreateClusterManagedPlan, CreateClusterPlan, CreateClusterReplicaPlan,
+    CreateClusterUnmanagedPlan, CreateClusterVariant, CreateConnectionPlan,
+    CreateContinualTaskPlan, CreateDatabasePlan, CreateIndexPlan, CreateMaterializedViewPlan,
+    CreateNetworkPolicyPlan, CreateRolePlan, CreateSchemaPlan, CreateSecretPlan, CreateSinkPlan,
+    CreateSourcePlan, CreateTablePlan, CreateTypePlan, CreateViewPlan, DataSourceDesc,
+    DropObjectsPlan, DropOwnedPlan, HirRelationExpr, Index, MaterializedView, NetworkPolicyRule,
+    NetworkPolicyRuleAction, NetworkPolicyRuleDirection, Plan, PlanClusterOption, PlanNotice,
+    PolicyAddress, QueryContext, ReplicaConfig, Secret, Sink, Source, Table, TableDataSource, Type,
+    VariableValue, View, WebhookBodyFormat, WebhookHeaderFilters, WebhookHeaders,
+    WebhookValidation, literal, plan_utils, query, transform_ast,
 };
 use crate::session::vars::{
     self, ENABLE_CLUSTER_SCHEDULE_REFRESH, ENABLE_COLLECTION_PARTITION_BY,
@@ -7367,7 +7367,39 @@ pub fn plan_alter_materialized_view_apply_replacement(
     scx: &StatementContext,
     stmt: AlterMaterializedViewApplyReplacementStatement,
 ) -> Result<Plan, PlanError> {
-    todo!()
+    let AlterMaterializedViewApplyReplacementStatement {
+        if_exists,
+        name,
+        replacement_name,
+    } = stmt;
+
+    let object_type = ObjectType::MaterializedView;
+    let Some(mv) = resolve_item_or_type(scx, object_type, name.clone(), if_exists)? else {
+        scx.catalog.add_notice(PlanNotice::ObjectDoesNotExist {
+            name: name.to_ast_string_simple(),
+            object_type,
+        });
+        return Ok(Plan::AlterNoop(AlterNoopPlan { object_type }));
+    };
+
+    let replacement = resolve_item_or_type(scx, object_type, replacement_name, false)?
+        .expect("if_exists not set");
+
+    if replacement.replacement_target() != Some(mv.id()) {
+        return Err(PlanError::InvalidReplacement {
+            item_type: mv.item_type(),
+            item_name: scx.catalog.minimal_qualification(mv.name()),
+            replacement_type: replacement.item_type(),
+            replacement_name: scx.catalog.minimal_qualification(replacement.name()),
+        });
+    }
+
+    Ok(Plan::AlterMaterializedViewApplyReplacement(
+        AlterMaterializedViewApplyReplacementPlan {
+            id: mv.id(),
+            replacement_id: replacement.id(),
+        },
+    ))
 }
 
 pub fn describe_comment(
diff --git a/src/sql/src/rbac.rs b/src/sql/src/rbac.rs

Original file line number	Diff line number	Diff line change
`@@ -859,7 +859,7 @@ impl crate::coord::Coordinator {`
`859`	`859`	`// If a dataflow was created, drop it once the peek command is sent.`
`860`	`860`	`if let Some(index_id) = drop_dataflow {`
`861`	`861`	`self.remove_compute_ids_from_timeline(vec![(compute_instance, index_id)]);`
`862`		`- self.drop_indexes(vec![(compute_instance, index_id)]);`
	`862`	`+ self.drop_compute_collections(vec![(compute_instance, index_id)]);`
`863`	`863`	`}`
`864`	`864`
`865`	`865`	`let persist_client = self.persist_client.clone();`