refactor(meta-service): upgrade snapshot storage(rotbl) (#18326)

* refactor(meta-service): upgrade snapshot storage(rotbl)

# `rotbl` crate Changelog

## Version 0.2.1

**What's Changed for Users:**

- **Storage is now pluggable** - You'll need to provide a storage instance when creating or opening Rotbl files. This change makes it possible to use different storage backends beyond just the filesystem
- **Codec functionality moved** - The encoding/decoding features have been moved to a separate `codeq` library, which means better code sharing across projects
- **Better async support** - Fixed thread safety issues so Rotbl works more reliably in async applications
- **Improved backwards compatibility** - Added extensive testing to ensure your existing data files will always work with newer versions

**New Capabilities:**

- **Flexible storage backends** - The new Storage trait means we can add support for cloud storage, databases, or other backends in the future
- **Better testing infrastructure** - Completely rebuilt our test system to work across different storage types
- **Runtime configuration** - Config values are now calculated when you need them rather than upfront, making the library more efficient

**What We Removed:**

- Some internal APIs that weren't being used
- Redundant configuration methods that were causing confusion

* chore: fix snapshot install bug

Author: drmingdrmer

Cargo.lock

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

Cargo.toml

@@ -22,6 +22,7 @@ use databend_common_meta_types::snapshot_db::DB;
 use databend_common_meta_types::sys_data::SysData;
 use futures::Stream;
 use futures_util::TryStreamExt;
+use rotbl::storage::impls::fs::FsStorage;
 use rotbl::v001::Rotbl;
 use rotbl::v001::RotblMeta;
 use rotbl::v001::SeqMarked;
@@ -30,40 +31,39 @@ use crate::leveled_store::leveled_map::LeveledMap;
 
 /// Builds a snapshot from series of key-value in `(String, SeqMarked)`
 pub(crate) struct DBBuilder {
-    path: String,
-    rotbl_builder: rotbl::v001::Builder,
+    storage_path: String,
+    rel_path: String,
+    rotbl_builder: rotbl::v001::Builder<FsStorage>,
 }
 
 impl DBBuilder {
     pub fn new<P: AsRef<Path>>(
-        path: P,
+        storage_path: P,
+        rel_path: &str,
         rotbl_config: rotbl::v001::Config,
     ) -> Result<Self, io::Error> {
-        let p = path.as_ref().to_str().ok_or_else(|| {
+        let storage = FsStorage::new(storage_path.as_ref().to_path_buf());
+
+        let base_path = storage_path.as_ref().to_str().ok_or_else(|| {
             io::Error::new(
                 io::ErrorKind::InvalidInput,
-                format!("invalid path: {:?}", path.as_ref()),
+                format!("invalid path: {:?}", storage_path.as_ref()),
             )
         })?;
 
-        let inner = rotbl::v001::Builder::new(rotbl_config, path.as_ref())?;
+        let rel_path = rel_path.to_string();
+
+        let inner = rotbl::v001::Builder::new(storage, rotbl_config, &rel_path)?;
 
         let b = Self {
-            path: p.to_string(),
+            storage_path: base_path.to_string(),
+            rel_path,
             rotbl_builder: inner,
         };
 
         Ok(b)
     }
 
-    /// This method is only used for test.
-    #[allow(dead_code)]
-    pub fn new_with_default_config<P: AsRef<Path>>(path: P) -> Result<Self, io::Error> {
-        let mut config = rotbl::v001::Config::default();
-        config.fill_default_values();
-        Self::new(path, config)
-    }
-
     /// Append a key-value pair to the builder, the keys must be sorted.
     pub fn append_kv(&mut self, k: String, v: SeqMarked) -> Result<(), io::Error> {
         self.rotbl_builder.append_kv(k, v)
@@ -82,8 +82,11 @@ impl DBBuilder {
         Ok(())
     }
 
-    /// Flush the data to disk and return a read-only table [`Rotbl`] instance.
-    pub fn flush(self, sys_data: SysData) -> Result<(String, Rotbl), io::Error> {
+    /// Flush the data to disk and return:
+    /// - storage path,
+    /// - relative path of the db,
+    /// - and a read-only table [`Rotbl`] instance.
+    pub fn flush(self, sys_data: SysData) -> Result<(String, String, Rotbl), io::Error> {
         let meta = serde_json::to_string(&sys_data).map_err(|e| {
             io::Error::new(
                 io::ErrorKind::InvalidData,
@@ -95,7 +98,7 @@ impl DBBuilder {
         let rotbl_meta = RotblMeta::new(0, meta);
 
         let r = self.rotbl_builder.commit(rotbl_meta)?;
-        Ok((self.path, r))
+        Ok((self.storage_path, self.rel_path, r))
     }
 
     /// Build a [`DB`] from a leveled map.
@@ -116,8 +119,8 @@ impl DBBuilder {
         self.append_kv_stream(strm).await?;
 
         let snapshot_id = make_snapshot_id(&sys_data);
-        let (path, r) = self.flush(sys_data)?;
-        let db = DB::new(path, snapshot_id, Arc::new(r))?;
+        let (storage_path, rel_path, r) = self.flush(sys_data)?;
+        let db = DB::new(storage_path, rel_path, snapshot_id, Arc::new(r))?;
 
         Ok(db)
     }

src/meta/raft-store/src/leveled_store/db_builder.rs

@@ -57,9 +57,8 @@ async fn test_db_map_api_ro() -> anyhow::Result<()> {
 
         let temp_dir = tempfile::tempdir()?;
         let path = temp_dir.path();
-        let path = path.join("temp-db");
 
-        let db_builder = DBBuilder::new_with_default_config(path)?;
+        let db_builder = DBBuilder::new(path, "temp-db", rotbl::v001::Config::default())?;
         db_builder
             .build_from_leveled_map(lm, |_| "1-1-1-1".to_string())
             .await?

src/meta/raft-store/src/leveled_store/db_map_api_ro_test.rs

@@ -13,28 +13,36 @@
 // limitations under the License.
 
 use std::io;
+use std::path::PathBuf;
 use std::sync::Arc;
 
 use databend_common_meta_types::snapshot_db::DB;
 use log::info;
 use openraft::SnapshotId;
+use rotbl::storage::impls::fs::FsStorage;
 use rotbl::v001::Rotbl;
 
 use crate::config::RaftConfig;
 use crate::sm_v003::open_snapshot::OpenSnapshot;
 
 impl OpenSnapshot for DB {
     fn open_snapshot(
-        path: impl ToString,
+        storage_path: impl ToString,
+        rel_path: impl ToString,
         snapshot_id: SnapshotId,
         raft_config: &RaftConfig,
     ) -> Result<Self, io::Error> {
+        let storage_path = storage_path.to_string();
+        let rel_path = rel_path.to_string();
+
+        let storage = FsStorage::new(PathBuf::from(&storage_path));
+
         let config = raft_config.to_rotbl_config();
-        let r = Rotbl::open(config, path.to_string())?;
+        let r = Rotbl::open(storage, config, &rel_path.to_string())?;
 
-        info!("Opened snapshot at {}", path.to_string());
+        info!("Opened snapshot at {storage_path}/{rel_path}");
 
-        let db = Self::new(path, snapshot_id, Arc::new(r))?;
+        let db = Self::new(storage_path, rel_path, snapshot_id, Arc::new(r))?;
         Ok(db)
     }
 }

src/meta/raft-store/src/leveled_store/db_open_snapshot_impl.rs

@@ -156,8 +156,12 @@ async fn test_compact() -> anyhow::Result<()> {
 
     let temp_dir = tempfile::tempdir()?;
     let path = temp_dir.path();
-    let path = path.join("temp-compacted");
-    compact(&mut lm, path.as_os_str().to_str().unwrap()).await?;
+    compact(
+        &mut lm,
+        path.as_os_str().to_str().unwrap(),
+        "temp-compacted",
+    )
+    .await?;
 
     let db = lm.persisted().unwrap();
 
@@ -208,8 +212,12 @@ async fn test_compact_expire_index() -> anyhow::Result<()> {
 
     let temp_dir = tempfile::tempdir()?;
     let path = temp_dir.path();
-    let path = path.join("temp-compacted");
-    compact(&mut lm, path.as_os_str().to_str().unwrap()).await?;
+    compact(
+        &mut lm,
+        path.as_os_str().to_str().unwrap(),
+        "temp-compacted",
+    )
+    .await?;
 
     let db = lm.persisted().unwrap();
 
@@ -359,8 +367,7 @@ async fn build_3_levels() -> anyhow::Result<(LeveledMap, impl Drop)> {
     // Move the bottom level to db
     let temp_dir = tempfile::tempdir()?;
     let path = temp_dir.path();
-    let path = path.join("temp-db");
-    move_bottom_to_db(&mut lm, path.to_str().unwrap()).await?;
+    move_bottom_to_db(&mut lm, path.to_str().unwrap(), "temp-db").await?;
 
     Ok((lm, temp_dir))
 }
@@ -393,13 +400,16 @@ async fn build_sm_with_expire() -> anyhow::Result<(SMV003, impl Drop)> {
 
     let temp_dir = tempfile::tempdir()?;
     let path = temp_dir.path();
-    let path = path.join("temp-db");
-    move_bottom_to_db(lm, path.to_str().unwrap()).await?;
+    move_bottom_to_db(lm, path.to_str().unwrap(), "temp-db").await?;
     Ok((sm, temp_dir))
 }
 
 /// Build a DB from the bottom level of the immutable levels.
-async fn move_bottom_to_db(lm: &mut LeveledMap, path: &str) -> Result<(), io::Error> {
+async fn move_bottom_to_db(
+    lm: &mut LeveledMap,
+    base_path: &str,
+    rel_path: &str,
+) -> Result<(), io::Error> {
     let mut immutables = lm.immutable_levels_ref().clone();
     let bottom = immutables.levels().remove(0);
     lm.replace_immutable_levels(immutables);
@@ -408,14 +418,14 @@ async fn move_bottom_to_db(lm: &mut LeveledMap, path: &str) -> Result<(), io::Er
     let mut lm2 = LeveledMap::default();
     lm2.replace_immutable_levels(bottom);
 
-    compact(&mut lm2, path).await?;
+    compact(&mut lm2, base_path, rel_path).await?;
 
     *lm.persisted_mut() = lm2.persisted_mut().clone();
     Ok(())
 }
 
-async fn compact(lm: &mut LeveledMap, path: &str) -> Result<(), io::Error> {
-    let db_builder = DBBuilder::new_with_default_config(path)?;
+async fn compact(lm: &mut LeveledMap, base_path: &str, rel_path: &str) -> Result<(), io::Error> {
+    let db_builder = DBBuilder::new(base_path, rel_path, rotbl::v001::Config::default())?;
 
     let db = db_builder
         .build_from_leveled_map(lm, |_sys_data| "1-1-1-1.snap".to_string())

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

@@ -18,8 +18,10 @@ use crate::config::RaftConfig;
 
 /// A trait for opening a snapshot.
 pub trait OpenSnapshot {
+    /// Open a snapshot at `<storage_path>/<rel_path>`
     fn open_snapshot(
-        path: impl ToString,
+        storage_path: impl ToString,
+        rel_path: impl ToString,
         snapshot_id: SnapshotId,
         raft_config: &RaftConfig,
     ) -> Result<Self, std::io::Error>

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

@@ -23,7 +23,9 @@ pub struct Received {
     pub format: String,
     pub vote: Vote,
     pub snapshot_meta: SnapshotMeta,
-    pub temp_path: String,
+
+    pub storage_path: String,
+    pub temp_rel_path: String,
 
     pub remote_addr: String,
 

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

@@ -33,7 +33,9 @@ use crate::sm_v003::received::Received;
 pub struct ReceiverV003 {
     remote_addr: String,
 
-    temp_path: String,
+    storage_path: String,
+
+    temp_rel_path: String,
 
     temp_file: Option<BufWriter<File>>,
 
@@ -49,13 +51,19 @@ pub struct ReceiverV003 {
 
 impl ReceiverV003 {
     /// Create a new snapshot receiver with an empty snapshot.
-    pub(crate) fn new(remote_addr: impl ToString, temp_path: impl ToString, temp_f: File) -> Self {
+    pub(crate) fn new(
+        remote_addr: impl ToString,
+        storage_path: impl ToString,
+        temp_rel_path: impl ToString,
+        temp_f: File,
+    ) -> Self {
         let remote_addr = remote_addr.to_string();
-        info!("Begin receiving snapshot v2 stream from: {}", remote_addr);
+        info!("Begin receiving snapshot v003 stream from: {}", remote_addr);
 
         ReceiverV003 {
             remote_addr,
-            temp_path: temp_path.to_string(),
+            storage_path: storage_path.to_string(),
+            temp_rel_path: temp_rel_path.to_string(),
             temp_file: Some(BufWriter::with_capacity(64 * 1024 * 1024, temp_f)),
             on_recv: None,
             n_received: 0,
@@ -118,27 +126,26 @@ impl ReceiverV003 {
         chunk: SnapshotChunkRequestV003,
     ) -> Result<Option<Received>, io::Error> {
         let remote_addr = self.remote_addr.clone();
-        let temp_path = self.temp_path.clone();
 
         fn invalid_input<E>(e: E) -> io::Error
         where E: Into<Box<dyn std::error::Error + Send + Sync>> {
             io::Error::new(io::ErrorKind::InvalidInput, e)
         }
 
+        // 1. update stat
+        self.update_stat(&chunk);
+
         // Add context info to io::Error
         let ctx = |e: io::Error, context: &str| -> io::Error {
             io::Error::new(
                 e.kind(),
                 format!(
-                    "{} while:(ReceiverV003::receive(): {}; remote_addr: {}; temp_path: {})",
-                    e, context, remote_addr, temp_path
+                    "{} while:(ReceiverV003::receive(): {}; remote_addr: {}; temp_path: {}/{})",
+                    e, context, remote_addr, self.storage_path, self.temp_rel_path
                 ),
             )
         };
 
-        // 1. update stat
-        self.update_stat(&chunk);
-
         // 2. write chunk to local snapshot_data
         {
             let f = self.temp_file.as_mut().ok_or_else(|| {
@@ -163,8 +170,8 @@ impl ReceiverV003 {
             };
 
             info!(
-                "snapshot from {} is completely received, format: {}, vote: {:?}, meta: {:?}, size: {}; path: {}",
-                self.remote_addr, format, vote, snapshot_meta, self.size_received, self.temp_path
+                "snapshot from {} is completely received, format: {}, vote: {:?}, meta: {:?}, size: {}; path: {}/{}",
+                self.remote_addr, format, vote, snapshot_meta, self.size_received, self.storage_path, self.temp_rel_path
             );
 
             if format != "rotbl::v001" {
@@ -198,7 +205,8 @@ impl ReceiverV003 {
                 format,
                 vote,
                 snapshot_meta,
-                temp_path: self.temp_path.clone(),
+                storage_path: self.storage_path.clone(),
+                temp_rel_path: self.temp_rel_path.clone(),
                 remote_addr: self.remote_addr.clone(),
                 n_received: self.n_received,
                 size_received: self.size_received,
@@ -214,15 +222,15 @@ impl ReceiverV003 {
         debug!(
             len = data_len,
             total_len = self.size_received;
-            "received {}-th snapshot chunk from {}; path: {}",
-            self.n_received, self.remote_addr, self.temp_path
+            "received {}-th snapshot chunk from {}; path: {}/{}",
+            self.n_received, self.remote_addr, self.storage_path, self.temp_rel_path
         );
 
         if self.n_received % 100 == 0 {
             info!(
                 total_len = self.size_received;
-                "received {}-th snapshot chunk from {}; path: {}",
-                self.n_received, self.remote_addr, self.temp_path
+                "received {}-th snapshot chunk from {}; path: {}/{}",
+                self.n_received, self.remote_addr, self.storage_path, self.temp_rel_path
             );
         }