fix(meta): unpin snapshot on MV creation failure

src/meta/src/barrier/mod.rs

@@ -822,6 +822,10 @@ where
             // If failed, enter recovery mode.
             self.set_status(BarrierManagerStatus::Recovering).await;
             *tracker = CreateMviewProgressTracker::new();
+            self.snapshot_manager
+                .unpin_all()
+                .await
+                .expect("unpin meta's snapshots");
             let new_epoch = self.recovery(state.in_flight_prev_epoch).await;
             state.in_flight_prev_epoch = new_epoch;
             state

src/meta/src/barrier/snapshot.rs

@@ -82,4 +82,11 @@ impl<S: MetaStore> SnapshotManager<S> {
 
         Ok(())
     }
+
+    pub async fn unpin_all(&self) -> MetaResult<()> {
+        let mut snapshots = self.snapshots.lock().await;
+        self.hummock_manager.unpin_snapshot(META_NODE_ID).await?;
+        snapshots.clear();
+        Ok(())
+    }
 }

src/meta/src/hummock/manager/compaction.rs

@@ -20,7 +20,6 @@ use risingwave_hummock_sdk::{CompactionGroupId, HummockCompactionTaskId, Hummock
 use risingwave_pb::hummock::{CompactTaskAssignment, CompactionConfig};
 
 use crate::hummock::compaction::CompactStatus;
-use crate::hummock::error::Result;
 use crate::hummock::manager::read_lock;
 use crate::hummock::HummockManager;
 use crate::model::BTreeMapTransaction;
@@ -41,10 +40,10 @@ impl Compaction {
     pub fn cancel_assigned_tasks_for_context_ids(
         &mut self,
         context_ids: &[HummockContextId],
-    ) -> Result<(
+    ) -> (
         BTreeMapTransaction<'_, CompactionGroupId, CompactStatus>,
         BTreeMapTransaction<'_, HummockCompactionTaskId, CompactTaskAssignment>,
-    )> {
+    ) {
         let mut compact_statuses = BTreeMapTransaction::new(&mut self.compaction_statuses);
         let mut compact_task_assignment =
             BTreeMapTransaction::new(&mut self.compact_task_assignment);
@@ -84,7 +83,7 @@ impl Compaction {
                 compact_task_assignment.remove(task_id);
             }
         }
-        Ok((compact_statuses, compact_task_assignment))
+        (compact_statuses, compact_task_assignment)
     }
 }
 
@@ -179,9 +178,7 @@ mod tests {
         );
 
         // irrelevant context id
-        let (compact_status, assignment) = compaction
-            .cancel_assigned_tasks_for_context_ids(&[22])
-            .unwrap();
+        let (compact_status, assignment) = compaction.cancel_assigned_tasks_for_context_ids(&[22]);
         compact_status.commit_memory();
         assignment.commit_memory();
         assert_eq!(
@@ -203,9 +200,7 @@ mod tests {
         );
 
         // target context id
-        let (compact_status, assignment) = compaction
-            .cancel_assigned_tasks_for_context_ids(&[11])
-            .unwrap();
+        let (compact_status, assignment) = compaction.cancel_assigned_tasks_for_context_ids(&[11]);
         compact_status.commit_memory();
         assignment.commit_memory();
         assert_eq!(

src/meta/src/hummock/manager/compaction_group_manager.rs

@@ -434,11 +434,12 @@ impl<S: MetaStore> CompactionGroupManagerInner<S> {
         let mut remove_some_group = false;
         let mut compaction_groups = BTreeMapTransaction::new(&mut self.compaction_groups);
         for table_id in table_ids {
-            let compaction_group_id = self
-                .index
-                .get(table_id)
-                .cloned()
-                .ok_or(Error::InvalidCompactionGroupMember(*table_id))?;
+            let compaction_group_id = match self.index.get(table_id).cloned() {
+                None => {
+                    continue;
+                }
+                Some(id) => id,
+            };
             let mut compaction_group = compaction_groups
                 .get_mut(compaction_group_id)
                 .ok_or(Error::InvalidCompactionGroup(compaction_group_id))?;

src/meta/src/hummock/manager/context.rs

@@ -29,6 +29,7 @@ use crate::hummock::manager::{
     commit_multi_var, read_lock, start_measure_real_process_timer, write_lock,
 };
 use crate::hummock::HummockManager;
+use crate::manager::META_NODE_ID;
 use crate::model::{BTreeMapTransaction, ValTransaction};
 use crate::storage::{MetaStore, Transaction};
 
@@ -54,7 +55,7 @@ where
         let mut compaction_guard = write_lock!(self, compaction).await;
         let compaction = compaction_guard.deref_mut();
         let (compact_statuses, compact_task_assignment) =
-            compaction.cancel_assigned_tasks_for_context_ids(context_ids.as_ref())?;
+            compaction.cancel_assigned_tasks_for_context_ids(context_ids.as_ref());
         for context_id in context_ids.as_ref() {
             self.compactor_manager
                 .purge_heartbeats_for_context(*context_id);
@@ -187,4 +188,8 @@ where
         .await;
         Ok(())
     }
+
+    pub async fn release_meta_context(&self) -> Result<()> {
+        self.release_contexts([META_NODE_ID]).await
+    }
 }

src/meta/src/hummock/manager/mod.rs

@@ -292,7 +292,7 @@ where
         instance.release_invalid_contexts().await?;
         instance.cancel_unassigned_compaction_task().await?;
         // Release snapshots pinned by meta on restarting.
-        instance.release_contexts([META_NODE_ID]).await?;
+        instance.release_meta_context().await?;
         Ok(instance)
     }