adapter: enable dropping of replaced materialized views

Once materialized views can have multiple historical storage collections
associated with them, we'll need to make sure to drop all of those when
dropping a materialized view. There is always only a single compute
collection to drop.

This commit ensures we'll do the right thing and also simplifies the
item dropping code a bit in the process: Instead of keeping track of
indexes/MVs/CTs to drop, we instead only track compute and storage
collections to drop. Doing so allows deleting a bunch of duplicate code.

Author: teskje

src/adapter/src/coord/catalog_implications.rs

@@ -159,9 +159,7 @@ impl Coordinator {
         let mut sources_to_drop = vec![];
         let mut replication_slots_to_drop: Vec<(PostgresConnection, String)> = vec![];
         let mut storage_sink_gids_to_drop = vec![];
-        let mut indexes_to_drop = vec![];
-        let mut materialized_views_to_drop = vec![];
-        let mut continual_tasks_to_drop = vec![];
+        let mut compute_gids_to_drop = vec![];
         let mut view_gids_to_drop = vec![];
         let mut secrets_to_drop = vec![];
         let mut vpc_endpoints_to_drop = vec![];
@@ -305,7 +303,7 @@ impl Coordinator {
                     );
                 }
                 CatalogImplication::Index(CatalogImplicationKind::Dropped(index, full_name)) => {
-                    indexes_to_drop.push((index.cluster_id, index.global_id()));
+                    compute_gids_to_drop.push((index.cluster_id, index.global_id()));
                     dropped_item_names.insert(index.global_id(), full_name);
                 }
                 CatalogImplication::MaterializedView(CatalogImplicationKind::Added(mv)) => {
@@ -325,7 +323,8 @@ impl Coordinator {
                     mv,
                     full_name,
                 )) => {
-                    materialized_views_to_drop.push((mv.cluster_id, mv.global_id_writes()));
+                    compute_gids_to_drop.push((mv.cluster_id, mv.global_id_writes()));
+                    sources_to_drop.extend(mv.global_ids().map(|gid| (catalog_id, gid)));
                     dropped_item_names.insert(mv.global_id_writes(), full_name);
                 }
                 CatalogImplication::View(CatalogImplicationKind::Added(view)) => {
@@ -358,7 +357,8 @@ impl Coordinator {
                     ct,
                     _full_name,
                 )) => {
-                    continual_tasks_to_drop.push((catalog_id, ct.cluster_id, ct.global_id()));
+                    compute_gids_to_drop.push((ct.cluster_id, ct.global_id()));
+                    sources_to_drop.push((catalog_id, ct.global_id()));
                 }
                 CatalogImplication::Secret(CatalogImplicationKind::Added(secret)) => {
                     tracing::debug!(?secret, "not handling AddSecret in here yet");
@@ -524,9 +524,7 @@ impl Coordinator {
             .map(|(_, gid)| *gid)
             .chain(tables_to_drop.iter().map(|(_, gid)| *gid))
             .chain(storage_sink_gids_to_drop.iter().copied())
-            .chain(indexes_to_drop.iter().map(|(_, id)| *id))
-            .chain(materialized_views_to_drop.iter().map(|(_, id)| *id))
-            .chain(continual_tasks_to_drop.iter().map(|(_, _, gid)| *gid))
+            .chain(compute_gids_to_drop.iter().map(|(_, gid)| *gid))
             .chain(view_gids_to_drop.iter().copied())
             .collect();
 
@@ -592,27 +590,14 @@ impl Coordinator {
             }
         }
 
-        let storage_ids_to_drop: BTreeSet<_> = sources_to_drop
+        let storage_gids_to_drop: BTreeSet<_> = sources_to_drop
             .iter()
             .map(|(_id, gid)| gid)
             .chain(storage_sink_gids_to_drop.iter())
             .chain(tables_to_drop.iter().map(|(_id, gid)| gid))
-            .chain(materialized_views_to_drop.iter().map(|(_, id)| id))
-            .chain(continual_tasks_to_drop.iter().map(|(_, _, gid)| gid))
             .copied()
             .collect();
 
-        let compute_ids_to_drop: BTreeSet<_> = indexes_to_drop
-            .iter()
-            .copied()
-            .chain(materialized_views_to_drop.iter().copied())
-            .chain(
-                continual_tasks_to_drop
-                    .iter()
-                    .map(|(_, cluster_id, gid)| (*cluster_id, *gid)),
-            )
-            .collect();
-
         // Gather resources that we have to remove from timeline state and
         // pre-check if any Timelines become empty, when we drop the specified
         // storage and compute resources.
@@ -624,13 +609,13 @@ impl Coordinator {
             let mut id_bundle = CollectionIdBundle::default();
 
             for storage_id in read_holds.storage_ids() {
-                if storage_ids_to_drop.contains(&storage_id) {
+                if storage_gids_to_drop.contains(&storage_id) {
                     id_bundle.storage_ids.insert(storage_id);
                 }
             }
 
             for (instance_id, id) in read_holds.compute_ids() {
-                if compute_ids_to_drop.contains(&(instance_id, id))
+                if compute_gids_to_drop.contains(&(instance_id, id))
                     || clusters_to_drop.contains(&instance_id)
                 {
                     id_bundle
@@ -713,16 +698,8 @@ impl Coordinator {
                 }
             }
 
-            if !indexes_to_drop.is_empty() {
-                self.drop_indexes(indexes_to_drop);
-            }
-
-            if !materialized_views_to_drop.is_empty() {
-                self.drop_materialized_views(materialized_views_to_drop);
-            }
-
-            if !continual_tasks_to_drop.is_empty() {
-                self.drop_continual_tasks(continual_tasks_to_drop);
+            if !compute_gids_to_drop.is_empty() {
+                self.drop_compute_collections(compute_gids_to_drop);
             }
 
             if !vpc_endpoints_to_drop.is_empty() {

src/adapter/src/coord/ddl.rs

@@ -739,9 +739,9 @@ impl Coordinator {
             .unwrap_or_terminate("cannot fail to drop sinks");
     }
 
-    pub(crate) fn drop_indexes(&mut self, indexes: Vec<(ClusterId, GlobalId)>) {
+    pub(crate) fn drop_compute_collections(&mut self, collections: Vec<(ClusterId, GlobalId)>) {
         let mut by_cluster: BTreeMap<_, Vec<_>> = BTreeMap::new();
-        for (cluster_id, gid) in indexes {
+        for (cluster_id, gid) in collections {
             by_cluster.entry(cluster_id).or_default().push(gid);
         }
         for (cluster_id, gids) in by_cluster {
@@ -755,58 +755,6 @@ impl Coordinator {
         }
     }
 
-    /// A convenience method for dropping materialized views.
-    pub(crate) fn drop_materialized_views(&mut self, mviews: Vec<(ClusterId, GlobalId)>) {
-        let mut by_cluster: BTreeMap<_, Vec<_>> = BTreeMap::new();
-        let mut mv_gids = Vec::new();
-        for (cluster_id, gid) in mviews {
-            by_cluster.entry(cluster_id).or_default().push(gid);
-            mv_gids.push(gid);
-        }
-
-        // Drop compute sinks.
-        for (cluster_id, ids) in by_cluster {
-            let compute = &mut self.controller.compute;
-            // A cluster could have been dropped, so verify it exists.
-            if compute.instance_exists(cluster_id) {
-                compute
-                    .drop_collections(cluster_id, ids)
-                    .unwrap_or_terminate("cannot fail to drop collections");
-            }
-        }
-
-        // Drop storage resources.
-        let storage_metadata = self.catalog.state().storage_metadata();
-        self.controller
-            .storage
-            .drop_sources(storage_metadata, mv_gids)
-            .unwrap_or_terminate("cannot fail to drop sources");
-    }
-
-    /// A convenience method for dropping continual tasks.
-    pub(crate) fn drop_continual_tasks(&mut self, cts: Vec<(CatalogItemId, ClusterId, GlobalId)>) {
-        let mut by_cluster: BTreeMap<_, Vec<_>> = BTreeMap::new();
-        let mut source_ids = Vec::new();
-        for (item_id, cluster_id, gid) in cts {
-            by_cluster.entry(cluster_id).or_default().push(gid);
-            source_ids.push((item_id, gid));
-        }
-
-        // Drop compute sinks.
-        for (cluster_id, ids) in by_cluster {
-            let compute = &mut self.controller.compute;
-            // A cluster could have been dropped, so verify it exists.
-            if compute.instance_exists(cluster_id) {
-                compute
-                    .drop_collections(cluster_id, ids)
-                    .unwrap_or_terminate("cannot fail to drop collections");
-            }
-        }
-
-        // Drop storage sources.
-        self.drop_sources(source_ids)
-    }
-
     pub(crate) fn drop_vpc_endpoints_in_background(&self, vpc_endpoints: Vec<CatalogItemId>) {
         let cloud_resource_controller = Arc::clone(self.cloud_resource_controller
             .as_ref()

src/adapter/src/coord/peek.rs

@@ -861,7 +861,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();