adapter,storage: fix replacement collection_metadata leak (#36172)

This commit fixes a leak of storage `collection_metadata` entries when
dropping replacement MVs without applying them. The previous code
intentionally avoided adding the IDs of dropped replacements to
`storage_collections_to_drop`, to avoid finalizing the shard, but
causing the leak in doing so. The solution implemented here is to make
`prepare_state` smarter and only finalize the shard if the dropped
collection is a primary.

### Motivation

Fixes MaterializeInc/database-issues#11322

Author: teskje

src/adapter/src/catalog/transact.rs

@@ -1714,11 +1714,7 @@ impl Catalog {
                 for item_id in delta.items {
                     let entry = state.get_entry(&item_id);
 
-                    // Drop associated storage collections, unless the dropped item is a
-                    // replacement, in which case the replacement target owns the storage
-                    // collection.
-                    if entry.item().is_storage_collection() && entry.replacement_target().is_none()
-                    {
+                    if entry.item().is_storage_collection() {
                         storage_collections_to_drop.extend(entry.global_ids());
                     }
 

src/storage-client/src/storage_collections.rs

@@ -1644,11 +1644,18 @@ impl StorageCollections for StorageCollectionsImpl {
         // Delete the metadata for any dropped collections.
         let dropped_mappings = txn.delete_collection_metadata(ids_to_drop);
 
-        let dropped_shards = dropped_mappings
-            .into_iter()
-            .map(|(_id, shard)| shard)
-            .collect();
-
+        // Only finalize the shards of dropped collections that don't have a primary.
+        // Otherwise the shard might still be in use by the primary.
+        let mut dropped_shards = BTreeSet::new();
+        {
+            let collections = self.collections.lock().expect("poisoned");
+            for (id, shard) in dropped_mappings {
+                let coll = collections.get(&id).expect("must exist");
+                if coll.primary.is_none() {
+                    dropped_shards.insert(shard);
+                }
+            }
+        }
         txn.insert_unfinalized_shards(dropped_shards)?;
 
         // Reconcile any shards we've successfully finalized with the shard

test/cluster/mzcompose.py

@@ -6431,6 +6431,53 @@ def workflow_github_10086(c: Composition) -> None:
     assert not upper_empty
 
 
+def workflow_github_11322(c: Composition) -> None:
+    """
+    Regression test for database-issues#11322, in which dropping a
+    replacement MV leaks a `collection_metadata` entry.
+    """
+
+    def storage_collection_metadata() -> dict[str, str]:
+        port = c.port("materialized", 6878)
+        resp = requests.get(f"http://localhost:{port}/api/catalog/dump")
+        resp.raise_for_status()
+        return resp.json()["storage_metadata"]["collection_metadata"]
+
+    c.up("materialized")
+
+    # Create a materialized view and a replacement.
+    c.sql("""
+        CREATE TABLE t (a int);
+        CREATE MATERIALIZED VIEW mv AS SELECT * FROM t;
+        CREATE REPLACEMENT MATERIALIZED VIEW rp FOR mv AS SELECT * FROM t;
+        SELECT * FROM mv;
+        """)
+
+    [(mv_id,)] = c.sql_query("SELECT id FROM mz_materialized_views WHERE name = 'mv'")
+    [(rp_id,)] = c.sql_query("SELECT id FROM mz_materialized_views WHERE name = 'rp'")
+
+    collection_metadata = storage_collection_metadata()
+    mv_shard = collection_metadata[mv_id]
+    rp_shard = collection_metadata[rp_id]
+    assert mv_shard == rp_shard
+
+    # Drop the replacement without applying it.
+    c.sql("DROP MATERIALIZED VIEW rp")
+
+    # Verify replacement's collection metadata has been removed.
+    collection_metadata = storage_collection_metadata()
+    assert collection_metadata[mv_id] == mv_shard
+    assert rp_id not in collection_metadata
+
+    c.kill("materialized")
+    c.up("materialized")
+
+    # Verify replacement's collection metadata has been removed durably.
+    collection_metadata = storage_collection_metadata()
+    assert collection_metadata[mv_id] == mv_shard
+    assert rp_id not in collection_metadata
+
+
 def workflow_test_github_10102(c: Composition) -> None:
     """
     Regression test for database-issues#10102: