merge

Author: b41sh

.github/actions/test_logs/action.yml

@@ -33,6 +33,9 @@ runs:
         bash ./tests/logging/test-logs.sh
 
     - name: Run History Tables Tests
+      env:
+        LOG_HISTORY_STORAGE_S3_ACCESS_KEY_ID: 'minioadmin'
+        LOG_HISTORY_STORAGE_S3_SECRET_ACCESS_KEY: 'minioadmin'
       shell: bash
       run: |
         bash ./tests/logging/test-history-tables.sh

Cargo.lock

@@ -466,7 +466,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

@@ -116,7 +116,7 @@ pub async fn read_metadata_async(
     let buffer_len = buffer.len();
 
     let map_err =
-        |e: ParquetError| ErrorCode::BadBytes(format!("Invalid Parquet File {path}: {e}",));
+        |e: ParquetError| ErrorCode::BadBytes(format!("Invalid Parquet file '{path}': {e}",));
     let footer_tail = ParquetMetaDataReader::decode_footer_tail(
         &buffer[(buffer_len - FOOTER_SIZE as usize)..]
             .try_into()
@@ -160,7 +160,7 @@ pub fn read_metadata_sync(
     check_footer_size(file_size, path)?;
 
     let map_err =
-        |e: ParquetError| ErrorCode::BadBytes(format!("Invalid Parquet File {path}: {e}",));
+        |e: ParquetError| ErrorCode::BadBytes(format!("Invalid Parquet file '{path}': {e}",));
     // read and cache up to DEFAULT_FOOTER_READ_SIZE bytes from the end and process the footer
     let default_end_len = DEFAULT_FOOTER_READ_SIZE.min(file_size);
     let buffer = blocking
@@ -198,7 +198,7 @@ pub fn read_metadata_sync(
 fn check_footer_size(file_size: u64, path: &str) -> Result<()> {
     if file_size < FOOTER_SIZE {
         Err(ErrorCode::BadBytes(format!(
-            "Invalid Parquet file {path}. Size is smaller than footer."
+            "Not a parquet file ({path}): only {file_size} bytes."
         )))
     } else {
         Ok(())
@@ -209,7 +209,7 @@ fn check_footer_size(file_size: u64, path: &str) -> Result<()> {
 fn check_meta_size(file_size: u64, metadata_len: u64, path: &str) -> Result<()> {
     if metadata_len + FOOTER_SIZE > file_size {
         Err(ErrorCode::BadBytes(format!(
-            "Invalid Parquet file {path}. Reported metadata length of {} + {} byte footer, but file is only {} bytes",
+            "Invalid Parquet file '{path}': Reported metadata length of {} + {} byte footer, but file is only {} bytes",
             metadata_len, FOOTER_SIZE, file_size
         )))
     } else {

src/common/storage/src/parquet.rs

@@ -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);