Merge branch 'main' into refactor/physical_plan_trait

Author: zhang2014

Cargo.lock

@@ -467,7 +467,7 @@ reqwest-hickory-resolver = "0.2"
 ringbuffer = "0.14.2"
 rmp-serde = "1.1.1"
 roaring = { version = "^0.10", features = ["serde"] }
-rotbl = { version = "0.2.7", features = [] }
+rotbl = { version = "0.2.8", features = [] }
 rust_decimal = "1.26"
 rustix = "0.38.37"
 rustls = { version = "0.23.27", features = ["ring", "tls12"], default-features = false }

Cargo.toml

@@ -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/raft-store/src/leveled_store/leveled_map/mod.rs

@@ -19,7 +19,6 @@ use databend_common_meta_types::snapshot_db::DB;
 use databend_common_meta_types::sys_data::SysData;
 use futures::Stream;
 use futures_util::TryStreamExt;
-use log::debug;
 use log::info;
 use rotbl::v001::SeqMarked;
 use tokio::sync::mpsc;
@@ -93,7 +92,7 @@ impl WriterV003 {
         mut kv_rx: mpsc::Receiver<WriteEntry<(String, SeqMarked), (MetaSnapshotId, SysData)>>,
     ) -> Result<DB, io::Error> {
         while let Some(ent) = kv_rx.blocking_recv() {
-            debug!(entry :? =(&ent); "write kv");
+            // debug!(entry :? =(&ent); "write kv");
 
             let (k, v) = match ent {
                 WriteEntry::Data(ent) => ent,

src/meta/raft-store/src/sm_v003/writer_v003.rs

@@ -58,8 +58,8 @@ where T: fmt::Display
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         write!(
             f,
-            "StateMachineLock-Write-Guard({}): {}",
-            self.purpose, self.inner
+            "StateMachineLock-Write-Guard: {} {}",
+            self.inner, self.purpose,
         )
     }
 }
@@ -68,8 +68,8 @@ impl<T: ?Sized> Drop for WriteGuard<'_, T> {
     fn drop(&mut self) {
         let elapsed = self.acquired.elapsed();
         info!(
-            "StateMachineLock-Write-Guard({}) released after {:?}",
-            self.purpose, elapsed
+            "StateMachineLock-Write-Release: total: {:?}; {}",
+            elapsed, self.purpose,
         );
     }
 }
@@ -105,8 +105,8 @@ where T: fmt::Display
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         write!(
             f,
-            "StateMachineLock-Read-Guard({}): {}",
-            self.purpose, self.inner
+            "StateMachineLock-Read-Guard: {} {}",
+            self.inner, self.purpose
         )
     }
 }
@@ -115,8 +115,8 @@ impl<T: ?Sized> Drop for ReadGuard<'_, T> {
     fn drop(&mut self) {
         let elapsed = self.acquired.elapsed();
         info!(
-            "StateMachineLock-Read-Guard({}) released after {:?}",
-            self.purpose, elapsed
+            "StateMachineLock-Read-Release: total: {:?}; {}",
+            elapsed, self.purpose,
         );
     }
 }

src/meta/service/src/store/lock_guards.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-Elapsed: 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-Elapsed: 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");