fix(meta-service): eliminate 250ms lock delay during snapshot compaction (#18542)

Move expensive compactor drop operation outside state machine write lock
to prevent blocking other operations for ~250ms during snapshot building.

Changes:
- Change replace_with_compacted to take compactor by reference
- Move compactor drop to async spawn_blocking after lock release

The compactor contains large data structures (ImmutableLevels, DB handles)
whose destruction involves significant memory deallocation and resource
cleanup, causing multi-hundred millisecond delays when done under lock.

Author: drmingdrmer

src/meta/raft-store/src/leveled_store/leveled_map/mod.rs

@@ -146,9 +146,11 @@ impl LeveledMap {
         self.immutable_levels = b;
     }
 
-    /// Replace several bottom immutable levels and persisted level
-    /// with the compacted data.
-    pub fn replace_with_compacted(&mut self, mut compactor: Compactor, db: DB) {
+    /// Replace bottom immutable levels and persisted level with compacted data.
+    ///
+    /// **Important**: Do not drop the compactor within this function when called
+    /// under a state machine lock, as dropping may take ~250ms.
+    pub fn replace_with_compacted(&mut self, compactor: &mut Compactor, db: DB) {
         let len = compactor.immutable_levels.len();
         let corresponding_index = compactor
             .immutable_levels
@@ -169,7 +171,7 @@ impl LeveledMap {
 
         self.persisted = Some(db);
 
-        info!("compaction finished replacing the db");
+        info!("replace_with_compacted: finished replacing the db");
     }
 
     /// Get a singleton `Compactor` instance specific to `self`.

src/meta/service/src/store/store_inner.rs

@@ -188,10 +188,10 @@ impl RaftStoreInner {
         let g = self
             .state_machine
             .read()
-            .with_timing(|_output, total, busy| {
+            .with_timing(|_output, total, _busy| {
                 info!(
-                    "StateMachineLock-Read-Acquire: total: {:?}, busy: {:?}; {}",
-                    total, busy, &for_what
+                    "StateMachineLock-Read-Acquire: total: {:?}; {}",
+                    total, &for_what
                 );
             })
             .await;
@@ -208,10 +208,10 @@ impl RaftStoreInner {
         let g = self
             .state_machine
             .write()
-            .with_timing(|_output, total, busy| {
+            .with_timing(|_output, total, _busy| {
                 info!(
-                    "StateMachineLock-Write-Acquire: total: {:?}, busy: {:?}; {}",
-                    total, busy, &for_what
+                    "StateMachineLock-Write-Acquire: total: {:?}; {}",
+                    total, &for_what
                 );
             })
             .await;
@@ -313,9 +313,22 @@ impl RaftStoreInner {
                 .get_state_machine_write("do_build_snapshot-replace-compacted")
                 .await;
             sm.levels_mut()
-                .replace_with_compacted(compactor, db.clone());
+                .replace_with_compacted(&mut compactor, db.clone());
+            info!("do_build_snapshot-replace_with_compacted returned");
         }
 
+        // Consume and drop compactor, dropping it may involve blocking IO
+        databend_common_base::runtime::spawn_blocking(move || {
+            let drop_start = std::time::Instant::now();
+            drop(compactor);
+            info!(
+                "do_build_snapshot-compactor-drop completed in {:?}",
+                drop_start.elapsed()
+            );
+        })
+        .await
+        .expect("compactor drop task should not panic");
+
         // Clean old snapshot
         ss_store.new_loader().clean_old_snapshots().await?;
         info!("do_build_snapshot clean_old_snapshots complete");