chore: codefmt

tonbo-io · Jul 30, 2024 · 9dcd0ee · 9dcd0ee
1 parent 023eb80
commit 9dcd0ee
Show file tree

Hide file tree

Showing 10 changed files with 55 additions and 48 deletions.
diff --git a/src/compaction/mod.rs b/src/compaction/mod.rs
@@ -15,7 +15,10 @@ use ulid::Ulid;
 
 use crate::{
     fs::{FileId, FileProvider},
-    inmem::immutable::{ArrowArrays, Builder, Immutable},
+    inmem::{
+        immutable::{ArrowArrays, Builder, Immutable},
+        mutable::Mutable,
+    },
     ondisk::sstable::SsTable,
     record::{KeyRef, Record},
     scope::Scope,
@@ -27,7 +30,6 @@ use crate::{
     },
     DbOption, Schema,
 };
-use crate::inmem::mutable::Mutable;
 
 #[derive(Debug)]
 pub(crate) enum CompactTask {
@@ -69,7 +71,7 @@ where
         let mut guard = self.schema.write().await;
 
         if guard.mutable.is_empty() {
-            return Ok(())
+            return Ok(());
         }
         let mutable = mem::replace(&mut guard.mutable, Mutable::new(&self.option).await?);
         let (file_id, immutable) = mutable.to_immutable().await?;

diff --git a/src/inmem/immutable.rs b/src/inmem/immutable.rs
@@ -8,9 +8,7 @@ use arrow::array::RecordBatch;
 use crossbeam_skiplist::SkipMap;
 use parquet::arrow::ProjectionMask;
 
-use super::mutable::Mutable;
 use crate::{
-    fs::FileProvider,
     record::{internal::InternalRecordRef, Key, Record, RecordRef},
     stream::record_batch::RecordBatchEntry,
     timestamp::{Timestamp, Timestamped, TimestampedRef, EPOCH},

diff --git a/src/inmem/mutable.rs b/src/inmem/mutable.rs
@@ -1,4 +1,4 @@
-use std::{collections::BTreeMap, intrinsics::transmute, ops::Bound};
+use std::{intrinsics::transmute, ops::Bound};
 
 use async_lock::Mutex;
 use crossbeam_skiplist::{
@@ -10,17 +10,13 @@ use ulid::Ulid;
 
 use crate::{
     fs::{FileId, FileProvider},
-    inmem::immutable::{ArrowArrays, Builder, Immutable},
+    inmem::immutable::Immutable,
     record::{Key, KeyRef, Record},
     timestamp::{
         timestamped::{Timestamped, TimestampedRef},
         Timestamp, EPOCH,
     },
-    wal::{
-        log::{Log, LogType},
-        record_entry::RecordEntry,
-        WalFile,
-    },
+    wal::{log::LogType, WalFile},
     DbOption, WriteError,
 };
 

diff --git a/src/lib.rs b/src/lib.rs
@@ -1,5 +1,3 @@
-#![allow(dead_code)]
-mod arrows;
 mod compaction;
 pub mod executor;
 pub mod fs;
@@ -17,7 +15,7 @@ mod wal;
 
 use std::{collections::VecDeque, io, marker::PhantomData, mem, ops::Bound, sync::Arc};
 
-use async_lock::{Mutex, RwLock, RwLockReadGuard};
+use async_lock::{RwLock, RwLockReadGuard};
 use flume::{bounded, Sender};
 use fs::FileProvider;
 use futures_core::Stream;
@@ -65,10 +63,10 @@ where
         let option = Arc::new(option);
         E::create_dir_all(&option.path).await?;
 
-        let (task_tx, mut task_rx) = bounded(1);
+        let (task_tx, task_rx) = bounded(1);
         let schema = Arc::new(RwLock::new(Schema::new(option.clone(), task_tx).await?));
 
-        let (mut cleaner, clean_sender) = Cleaner::new(option.clone());
+        let (mut cleaner, clean_sender) = Cleaner::<E>::new(option.clone());
 
         let version_set = VersionSet::new(clean_sender, option.clone()).await?;
         let mut compactor =
@@ -350,23 +348,25 @@ pub(crate) mod tests {
         array::{Array, AsArray, RecordBatch},
         datatypes::{DataType, Field, Schema, UInt32Type},
     };
-    use async_lock::{Mutex, RwLock};
+    use async_lock::RwLock;
     use flume::{bounded, Receiver};
     use futures_util::io;
     use once_cell::sync::Lazy;
     use parquet::arrow::ProjectionMask;
     use tempfile::TempDir;
     use tracing::error;
 
-    use crate::{compaction::{CompactTask, Compactor}, executor::{tokio::TokioExecutor, Executor}, fs::FileId, inmem::{
-        immutable::{ArrowArrays, Builder},
-        mutable::Mutable,
-    }, record::{internal::InternalRecordRef, Key, RecordRef}, serdes::{
-        option::{DecodeError, EncodeError},
-        Decode, Encode,
-    }, timestamp::Timestamped, version::{cleaner::Cleaner, set::tests::build_version_set, Version}, wal::log::LogType, DbOption, Immutable, Record, WriteError, DB, Projection};
-    use crate::inmem::immutable::tests::TestImmutableArrays;
-    use crate::record::{RecordDecodeError, RecordEncodeError};
+    use crate::{
+        compaction::{CompactTask, Compactor},
+        executor::{tokio::TokioExecutor, Executor},
+        fs::FileId,
+        inmem::{immutable::tests::TestImmutableArrays, mutable::Mutable},
+        record::{internal::InternalRecordRef, RecordDecodeError, RecordEncodeError, RecordRef},
+        serdes::{Decode, Encode},
+        version::{cleaner::Cleaner, set::tests::build_version_set, Version},
+        wal::log::LogType,
+        DbOption, Immutable, Projection, Record, WriteError, DB,
+    };
 
     #[derive(Debug, PartialEq, Eq)]
     pub struct Test {
@@ -554,11 +554,11 @@ pub(crate) mod tests {
         }
     }
 
-    pub(crate) async fn get_test_record_batch<E: Executor>(
+    pub(crate) async fn get_test_record_batch<E: Executor + Send + Sync + 'static>(
         option: DbOption,
         executor: E,
     ) -> RecordBatch {
-        let db: DB<Test, E> = DB::new(option, executor).await.unwrap();
+        let db: DB<Test, E> = DB::new(option.clone(), executor).await.unwrap();
 
         db.write(
             Test {
@@ -590,7 +590,9 @@ pub(crate) mod tests {
             .clone()
     }
 
-    pub(crate) async fn build_schema(option: DbOption) -> io::Result<(crate::Schema<Test, TokioExecutor>, Receiver<CompactTask>)> {
+    pub(crate) async fn build_schema(
+        option: Arc<DbOption>,
+    ) -> io::Result<(crate::Schema<Test, TokioExecutor>, Receiver<CompactTask>)> {
         let mutable = Mutable::new(&option).await?;
 
         mutable
@@ -702,7 +704,7 @@ pub(crate) mod tests {
 
         let schema = Arc::new(RwLock::new(schema));
 
-        let (mut cleaner, clean_sender) = Cleaner::new(option.clone());
+        let (mut cleaner, clean_sender) = Cleaner::<E>::new(option.clone());
         let version_set = build_version_set(version, clean_sender, option.clone()).await?;
         let mut compactor =
             Compactor::<R, E>::new(schema.clone(), option.clone(), version_set.clone());
@@ -948,7 +950,7 @@ pub(crate) mod tests {
         option.level_sst_magnification = 10;
         option.max_sst_file_size = 2 * 1024 * 1024;
 
-        let db: DB<Test, TokioExecutor> = DB::new(Arc::new(option), TokioExecutor::new()).await.unwrap();
+        let db: DB<Test, TokioExecutor> = DB::new(option, TokioExecutor::new()).await.unwrap();
 
         for item in test_items() {
             db.write(item, 0.into()).await.unwrap();
@@ -958,9 +960,9 @@ pub(crate) mod tests {
         let key = 20.to_string();
         let option1 = tx.get(&key, Projection::All).await.unwrap().unwrap();
 
-        assert_eq!(option1.get().map(|test_ref| test_ref.vstring), Some("20"));
-        assert_eq!(option1.get().map(|test_ref| test_ref.vu32), Some(Some(0)));
-        assert_eq!(option1.get().map(|test_ref| test_ref.vbool), Some(Some(true)));
+        assert_eq!(option1.get().vstring, "20");
+        assert_eq!(option1.get().vu32, Some(0));
+        assert_eq!(option1.get().vbool, Some(true));
 
         dbg!(db.version_set.current().await);
     }

diff --git a/src/ondisk/sstable.rs b/src/ondisk/sstable.rs
@@ -130,7 +130,7 @@ where
 
 #[cfg(test)]
 pub(crate) mod tests {
-    use std::{borrow::Borrow, ops::Bound, path::PathBuf, sync::Arc};
+    use std::{borrow::Borrow, ops::Bound, path::PathBuf};
 
     use futures_util::StreamExt;
     use parquet::arrow::{arrow_to_parquet_schema, ProjectionMask};

diff --git a/src/option.rs b/src/option.rs
@@ -8,7 +8,7 @@ use crate::{
     version::Version,
 };
 
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 pub struct DbOption {
     pub(crate) path: PathBuf,
     pub(crate) max_mem_table_size: usize,

diff --git a/src/timestamp/mod.rs b/src/timestamp/mod.rs
@@ -1,6 +1,6 @@
 pub mod timestamped;
 
-use std::{future::Future, io};
+use std::io;
 
 use arrow::{
     array::{PrimitiveArray, Scalar},

diff --git a/src/transaction.rs b/src/transaction.rs
@@ -13,6 +13,7 @@ use parquet::errors::ParquetError;
 use thiserror::Error;
 
 use crate::{
+    compaction::CompactTask,
     fs::FileProvider,
     record::{Key, KeyRef},
     stream,
@@ -21,7 +22,6 @@ use crate::{
     wal::log::LogType,
     LockMap, Projection, Record, Scan, Schema, WriteError,
 };
-use crate::compaction::CompactTask;
 
 pub(crate) struct TransactionScan<'scan, R: Record> {
     inner: Range<'scan, R::Key, Option<R>>,

diff --git a/src/version/cleaner.rs b/src/version/cleaner.rs
@@ -1,21 +1,29 @@
-use std::{collections::BTreeMap, io, sync::Arc};
+use std::{collections::BTreeMap, io, marker::PhantomData, sync::Arc};
 
 use flume::{Receiver, Sender};
-use tokio::fs;
-use crate::{fs::FileId, timestamp::Timestamp, DbOption};
+
+use crate::{
+    fs::{FileId, FileProvider},
+    timestamp::Timestamp,
+    DbOption,
+};
 
 pub enum CleanTag {
     Add { ts: Timestamp, gens: Vec<FileId> },
     Clean { ts: Timestamp },
 }
 
-pub(crate) struct Cleaner {
+pub(crate) struct Cleaner<FP: FileProvider> {
     tag_recv: Receiver<CleanTag>,
     gens_map: BTreeMap<Timestamp, (Vec<FileId>, bool)>,
     option: Arc<DbOption>,
+    _p: PhantomData<FP>,
 }
 
-impl Cleaner {
+impl<FP> Cleaner<FP>
+where
+    FP: FileProvider,
+{
     pub(crate) fn new(option: Arc<DbOption>) -> (Self, Sender<CleanTag>) {
         let (tag_send, tag_recv) = flume::bounded(option.clean_channel_buffer);
 
@@ -24,6 +32,7 @@ impl Cleaner {
                 tag_recv,
                 gens_map: Default::default(),
                 option,
+                _p: Default::default(),
             },
             tag_send,
         )
@@ -45,7 +54,7 @@ impl Cleaner {
                             break;
                         }
                         for gen in gens {
-                            fs::remove_file(self.option.table_path(&gen)).await?;
+                            FP::remove(self.option.table_path(&gen)).await?;
                         }
                     }
                 }

diff --git a/src/wal/mod.rs b/src/wal/mod.rs
@@ -124,8 +124,8 @@ mod tests {
     use futures_util::StreamExt;
     use tokio_util::compat::TokioAsyncReadCompatExt;
 
-    use super::{log::LogType, FileId, Log, WalFile};
-    use crate::{timestamp::Timestamped, wal::record_entry::RecordEntry};
+    use super::{log::LogType, FileId, WalFile};
+    use crate::timestamp::Timestamped;
 
     #[tokio::test]
     async fn write_and_recover() {
@@ -146,7 +146,7 @@ mod tests {
 
             {
                 let mut stream = pin!(wal.recover());
-                let (log_ty, key, value) = stream.next().await.unwrap().unwrap();
+                let (_, key, value) = stream.next().await.unwrap().unwrap();
                 assert_eq!(key.ts, 0.into());
                 assert_eq!(value, Some("hello".to_string()));
             }