refactor(storage): simplify build_compaction_group_info (#7999)

This PR does several trivial refactor/fix:
- Refactor: simplify build_compaction_group_info, as member_table_ids is now included in hummock version.  Note that even after this PR, [this comment](https://github.com/risingwavelabs/risingwave/blob/d30a96e2150a568d906922708f8ae0e3cd376a4b/src/storage/src/hummock/compactor/shared_buffer_compact.rs#L63) stands correct because compute node's hummock version may have not been updated to contain the new table id.
- Fix: fix entry leaking of read_version_mapping_guard.
- Fix: add a notification for backup manager.

Approved-By: Little-Wallace
Approved-By: Li0k

src/meta/src/backup_restore/backup_manager.rs

@@ -189,6 +189,14 @@ impl<S: MetaStore> BackupManager<S> {
     /// Deletes existent backups from backup storage.
     pub async fn delete_backups(&self, ids: &[MetaSnapshotId]) -> MetaResult<()> {
         self.backup_store.delete(ids).await?;
+        self.env
+            .notification_manager()
+            .notify_hummock_without_version(
+                Operation::Update,
+                Info::MetaBackupManifestId(MetaBackupManifestId {
+                    id: self.backup_store.manifest().manifest_id,
+                }),
+            );
         Ok(())
     }
 

src/storage/hummock_sdk/src/compaction_group/hummock_version_ext.rs

@@ -343,13 +343,8 @@ impl HummockVersionUpdateExt for HummockVersion {
     fn build_compaction_group_info(&self) -> HashMap<TableId, CompactionGroupId> {
         let mut ret = HashMap::new();
         for (compaction_group_id, levels) in &self.levels {
-            if let Some(ref l0) = levels.l0 {
-                for sub_level in l0.get_sub_levels() {
-                    update_compaction_group_info(sub_level, *compaction_group_id, &mut ret);
-                }
-            }
-            for level in levels.get_levels() {
-                update_compaction_group_info(level, *compaction_group_id, &mut ret);
+            for table_id in &levels.member_table_ids {
+                ret.insert(TableId::new(*table_id), *compaction_group_id);
             }
         }
         ret
@@ -375,18 +370,6 @@ impl HummockVersionUpdateExt for HummockVersion {
     }
 }
 
-fn update_compaction_group_info(
-    level: &Level,
-    compaction_group_id: CompactionGroupId,
-    compaction_group_info: &mut HashMap<TableId, CompactionGroupId>,
-) {
-    for table_info in level.get_table_infos() {
-        table_info.get_table_ids().iter().for_each(|table_id| {
-            compaction_group_info.insert(TableId::new(*table_id), compaction_group_id);
-        });
-    }
-}
-
 pub trait HummockLevelsExt {
     fn get_level0(&self) -> &OverlappingLevel;
     fn get_level(&self, idx: usize) -> &Level;

src/storage/src/hummock/event_handler/hummock_event_handler.rs

@@ -544,15 +544,18 @@ impl HummockEventHandler {
                                 instance_id
                             );
                             let mut read_version_mapping_guard = self.read_version_mapping.write();
-                            read_version_mapping_guard
+                            let entry = read_version_mapping_guard
                                 .get_mut(&table_id)
                                 .unwrap_or_else(|| {
                                     panic!(
                                         "DestroyHummockInstance table_id {} instance_id {} fail",
                                         table_id, instance_id
                                     )
-                                })
-                                .remove(&instance_id).unwrap_or_else(|| panic!("DestroyHummockInstance inexist instance table_id {} instance_id {}",  table_id, instance_id));
+                                });
+                            entry.remove(&instance_id).unwrap_or_else(|| panic!("DestroyHummockInstance inexist instance table_id {} instance_id {}",  table_id, instance_id));
+                            if entry.is_empty() {
+                                read_version_mapping_guard.remove(&table_id);
+                            }
                         }
                     }
                 }