refactor(meta-service): remove temp-snapshot during building snaphsot (#18344)

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.2.1", features = [] }
+rotbl = { version = "0.2.3", features = [] }
 rust_decimal = "1.26"
 rustix = "0.38.37"
 rustls = { version = "0.23.27", features = ["ring", "tls12"], default-features = false }

Cargo.toml

@@ -58,10 +58,10 @@ pub async fn import_v004(
     let writer = snapshot_store.new_writer()?;
     let (tx, join_handle) = writer.spawn_writer_thread("import_v004");
 
-    let sys_data = Arc::new(Mutex::new(SysData::default()));
+    let sys_data_holder = Arc::new(Mutex::new(SysData::default()));
 
     let mut converter = SMEntryV002ToV004 {
-        sys_data: sys_data.clone(),
+        sys_data: sys_data_holder.clone(),
     };
 
     for line in lines {
@@ -93,20 +93,16 @@ pub async fn import_v004(
     let max_log_id = raft_log_importer.max_log_id;
     raft_log_importer.flush().await?;
 
-    let s = {
-        let r = sys_data.lock().unwrap();
+    let sys_data = {
+        let r = sys_data_holder.lock().unwrap();
         r.clone()
     };
-
-    tx.send(WriteEntry::Finish(s)).await?;
-    let temp_snapshot_data = join_handle.await??;
-
-    let last_applied = {
-        let r = sys_data.lock().unwrap();
-        *r.last_applied_ref()
-    };
+    let last_applied = *sys_data.last_applied_ref();
     let snapshot_id = MetaSnapshotId::new_with_epoch(last_applied);
-    let db = temp_snapshot_data.move_to_final_path(snapshot_id.to_string())?;
+
+    tx.send(WriteEntry::Finish((snapshot_id.clone(), sys_data)))
+        .await?;
+    let db = join_handle.await??;
 
     eprintln!(
         "{data_version}: Imported {} records, snapshot: {}; snapshot_path: {}; snapshot_stat: {}",

src/meta/control/src/import_v004.rs

@@ -28,11 +28,14 @@ use rotbl::v001::RotblMeta;
 use rotbl::v001::SeqMarked;
 
 use crate::leveled_store::leveled_map::LeveledMap;
+use crate::sm_v003::open_snapshot::OpenSnapshot;
+#[cfg(doc)]
+use crate::sm_v003::SnapshotStoreV004;
+use crate::snapshot_config::SnapshotConfig;
+use crate::state_machine::MetaSnapshotId;
 
 /// Builds a snapshot from series of key-value in `(String, SeqMarked)`
 pub(crate) struct DBBuilder {
-    storage_path: String,
-    rel_path: String,
     rotbl_builder: rotbl::v001::Builder<FsStorage>,
 }
 
@@ -44,20 +47,11 @@ impl DBBuilder {
     ) -> Result<Self, io::Error> {
         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: {:?}", storage_path.as_ref()),
-            )
-        })?;
-
         let rel_path = rel_path.to_string();
 
         let inner = rotbl::v001::Builder::new(storage, rotbl_config, &rel_path)?;
 
         let b = Self {
-            storage_path: base_path.to_string(),
-            rel_path,
             rotbl_builder: inner,
         };
 
@@ -82,11 +76,39 @@ impl DBBuilder {
         Ok(())
     }
 
-    /// 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> {
+    /// Commit the building to the provided [`SnapshotStoreV004`].
+    pub fn commit_to_snapshot_store(
+        self,
+        snapshot_config: &SnapshotConfig,
+        snapshot_id: MetaSnapshotId,
+        sys_data: SysData,
+    ) -> Result<DB, io::Error> {
+        let config = self.rotbl_builder.config().clone();
+        let storage_path = self.rotbl_builder.storage().base_dir_str().to_string();
+
+        let (current_rel_path, _r) = self.commit(sys_data)?;
+
+        let current_path = format!("{}/{}", storage_path, current_rel_path);
+
+        let (_, rel_path) =
+            snapshot_config.move_to_final_path(&current_path, snapshot_id.to_string())?;
+
+        let db = DB::open_snapshot(&storage_path, &rel_path, snapshot_id.to_string(), config)
+            .map_err(|e| {
+                io::Error::new(
+                    e.kind(),
+                    format!(
+                        "{}; when:(open snapshot at: {}/{})",
+                        e, storage_path, rel_path
+                    ),
+                )
+            })?;
+
+        Ok(db)
+    }
+
+    /// Commit the building.
+    pub(crate) fn commit(self, sys_data: SysData) -> Result<(String, Rotbl), io::Error> {
         let meta = serde_json::to_string(&sys_data).map_err(|e| {
             io::Error::new(
                 io::ErrorKind::InvalidData,
@@ -97,8 +119,12 @@ impl DBBuilder {
         // the first arg `seq` is not used.
         let rotbl_meta = RotblMeta::new(0, meta);
 
+        // The relative path of the built rotbl.
+        let rel_path = self.rotbl_builder.rel_path().to_string();
+
         let r = self.rotbl_builder.commit(rotbl_meta)?;
-        Ok((self.storage_path, self.rel_path, r))
+
+        Ok((rel_path, r))
     }
 
     /// Build a [`DB`] from a leveled map.
@@ -118,10 +144,17 @@ impl DBBuilder {
 
         self.append_kv_stream(strm).await?;
 
+        let storage_path = self.storage_path().to_string();
+
         let snapshot_id = make_snapshot_id(&sys_data);
-        let (storage_path, rel_path, r) = self.flush(sys_data)?;
+        let (rel_path, r) = self.commit(sys_data)?;
         let db = DB::new(storage_path, rel_path, snapshot_id, Arc::new(r))?;
 
         Ok(db)
     }
+
+    /// Returns the base dir of the internal FS-storage.
+    fn storage_path(&self) -> &str {
+        self.rotbl_builder.storage().base_dir_str()
+    }
 }

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

@@ -22,22 +22,20 @@ 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(
         storage_path: impl ToString,
         rel_path: impl ToString,
         snapshot_id: SnapshotId,
-        raft_config: &RaftConfig,
+        config: rotbl::v001::Config,
     ) -> 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(storage, config, &rel_path.to_string())?;
 
         info!("Opened snapshot at {storage_path}/{rel_path}");

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

@@ -24,7 +24,6 @@ pub mod open_snapshot;
 pub mod received;
 pub mod receiver_v003;
 pub mod snapshot_loader;
-pub mod temp_snapshot_data;
 pub mod write_entry;
 pub mod writer_stat;
 

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

@@ -14,16 +14,14 @@
 
 use openraft::SnapshotId;
 
-use crate::config::RaftConfig;
-
 /// A trait for opening a snapshot.
 pub trait OpenSnapshot {
     /// Open a snapshot at `<storage_path>/<rel_path>`
     fn open_snapshot(
         storage_path: impl ToString,
         rel_path: impl ToString,
         snapshot_id: SnapshotId,
-        raft_config: &RaftConfig,
+        config: rotbl::v001::Config,
     ) -> Result<Self, std::io::Error>
     where
         Self: Sized;

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

@@ -77,7 +77,7 @@ where SD: OpenSnapshot
             storage_path.clone(),
             rel_path,
             snapshot_id.clone(),
-            self.snapshot_config.raft_config(),
+            self.snapshot_config.raft_config().to_rotbl_config(),
         )
         .map_err(|e| {
             error!("failed to open snapshot file({}): {}", storage_path, e);

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

@@ -18,7 +18,6 @@ use std::ops::Deref;
 use std::ops::DerefMut;
 
 use databend_common_meta_types::snapshot_db::DB;
-use databend_common_meta_types::sys_data::SysData;
 
 use crate::config::RaftConfig;
 use crate::ondisk::DataVersion;
@@ -114,40 +113,4 @@ impl SnapshotStoreV004 {
         self.snapshot_config.ensure_snapshot_dir()?;
         WriterV003::new(&self.snapshot_config)
     }
-
-    /// Create a new temp file to receive snapshot data in binary format
-    pub fn new_temp_file(&self) -> Result<(String, fs::File), io::Error> {
-        self.snapshot_config.ensure_snapshot_dir()?;
-
-        let temp_path = self.snapshot_config.snapshot_temp_path();
-
-        let f = fs::OpenOptions::new()
-            .create_new(true)
-            .write(true)
-            .read(true)
-            .open(&temp_path)
-            .map_err(|e| {
-                io::Error::new(
-                    e.kind(),
-                    format!(
-                        "{}: while(SnapshotStoreV003::new_temp_file(); path: {})",
-                        e, temp_path
-                    ),
-                )
-            })?;
-
-        Ok((temp_path, f))
-    }
-
-    /// This method is only used to pass openraft test.
-    pub fn new_temp(&self) -> Result<DB, io::Error> {
-        self.snapshot_config.ensure_snapshot_dir()?;
-
-        let w = self.new_writer()?;
-        let temp_data = w.flush(SysData::default())?;
-
-        let temp_id = self.snapshot_config.temp_snapshot_id();
-        let db = temp_data.move_to_final_path(temp_id.clone())?;
-        Ok(db)
-    }
 }