Merge remote-tracking branch 'origin/main' into refactor/schema-trait

tonbo-io · Dec 16, 2024 · 14d6e5f · 14d6e5f
2 parents ca90c0e + 1a48277
commit 14d6e5f
Show file tree

Hide file tree

Showing 15 changed files with 45 additions and 42 deletions.
diff --git a/README.md b/README.md
@@ -54,7 +54,7 @@ pub struct User {
 #[tokio::main]
 async fn main() {
     // pluggable async runtime and I/O
-    let db = DB::new("./db_path/users".into(), TokioExecutor::default())
+    let db = DB::new("./db_path/users".into(), TokioExecutor::current())
         .await
         .unwrap();
 

diff --git a/benches/common.rs b/benches/common.rs
@@ -272,7 +272,11 @@ impl BenchDatabase for TonboS3BenchDataBase {
             .unwrap()
             .disable_wal();
 
-        TonboS3BenchDataBase::new(tonbo::DB::new(option, TokioExecutor::new()).await.unwrap())
+        TonboS3BenchDataBase::new(
+            tonbo::DB::new(option, TokioExecutor::current())
+                .await
+                .unwrap(),
+        )
     }
 }
 
@@ -320,7 +324,9 @@ impl BenchDatabase for TonboBenchDataBase {
             DbOption::from(fusio::path::Path::from_filesystem_path(path.as_ref()).unwrap())
                 .disable_wal();
 
-        let db = tonbo::DB::new(option, TokioExecutor::new()).await.unwrap();
+        let db = tonbo::DB::new(option, TokioExecutor::current())
+            .await
+            .unwrap();
         TonboBenchDataBase::new(db)
     }
 }

diff --git a/benches/criterion/writes.rs b/benches/criterion/writes.rs
@@ -61,7 +61,7 @@ fn single_write(c: &mut Criterion) {
         let option = DbOption::from(fusio::path::Path::from_filesystem_path("/tmp/tonbo").unwrap())
             .disable_wal();
         let db = runtime
-            .block_on(async { DB::new(option, TokioExecutor::default()).await })
+            .block_on(async { DB::new(option, TokioExecutor::current()).await })
             .unwrap();
 
         group.bench_with_input(BenchmarkId::new("Tonbo", batch), &batch, |b, batch| {

diff --git a/bindings/python/src/db.rs b/bindings/python/src/db.rs
@@ -63,7 +63,7 @@ impl TonboDB {
             .block_on(async {
                 DB::with_schema(
                     option,
-                    TokioExecutor::new(),
+                    TokioExecutor::current(),
                     schema,
                 )
                 .await

diff --git a/examples/datafusion.rs b/examples/datafusion.rs
@@ -226,7 +226,7 @@ async fn main() -> Result<()> {
         &MusicSchema,
     ));
 
-    let db = DB::new(options, TokioExecutor::default(), MusicSchema)
+    let db = DB::new(options, TokioExecutor::current(), MusicSchema)
         .await
         .unwrap();
     for (id, name, like) in [

diff --git a/examples/declare.rs b/examples/declare.rs
@@ -27,7 +27,7 @@ async fn main() {
         &UserSchema,
     ));
     // pluggable async runtime and I/O
-    let db = DB::new(options, TokioExecutor::default(), UserSchema)
+    let db = DB::new(options, TokioExecutor::current(), UserSchema)
         .await
         .unwrap();
 

diff --git a/examples/dynamic.rs b/examples/dynamic.rs
@@ -22,7 +22,7 @@ async fn main() {
         Path::from_filesystem_path("./db_path/users").unwrap(),
         &schema,
     ));
-    let db = DB::with_schema(options, TokioExecutor::new(), schema)
+    let db = DB::with_schema(options, TokioExecutor::current(), schema)
         .await
         .unwrap();
 

diff --git a/src/compaction/mod.rs b/src/compaction/mod.rs
@@ -1286,7 +1286,7 @@ pub(crate) mod tests {
         option.major_default_oldest_table_num = 1;
         option.trigger_type = TriggerType::Length(5);
 
-        let db: DB<Test, TokioExecutor> = DB::new(option, TokioExecutor::new(), TestSchema)
+        let db: DB<Test, TokioExecutor> = DB::new(option, TokioExecutor::current(), TestSchema)
             .await
             .unwrap();
 

diff --git a/src/executor.rs b/src/executor.rs
@@ -22,14 +22,8 @@ pub mod tokio {
         handle: Handle,
     }
 
-    impl Default for TokioExecutor {
-        fn default() -> Self {
-            Self::new()
-        }
-    }
-
     impl TokioExecutor {
-        pub fn new() -> Self {
+        pub fn current() -> Self {
             Self {
                 handle: Handle::current(),
             }

diff --git a/src/lib.rs b/src/lib.rs
@@ -59,7 +59,7 @@
 //!         &UserSchema,
 //!     ));
 //!     // pluggable async runtime and I/O
-//!     let db = DB::new(options, TokioExecutor::default(), UserSchema)
+//!     let db = DB::new(options, TokioExecutor::current(), UserSchema)
 //!         .await
 //!         .unwrap();
 //!     // insert with owned value
@@ -1606,7 +1606,7 @@ pub(crate) mod tests {
         option.major_default_oldest_table_num = 1;
         option.trigger_type = TriggerType::Length(/* max_mutable_len */ 5);
 
-        let db: DB<Test, TokioExecutor> = DB::new(option, TokioExecutor::new(), TestSchema)
+        let db: DB<Test, TokioExecutor> = DB::new(option, TokioExecutor::current(), TestSchema)
             .await
             .unwrap();
 
@@ -1647,7 +1647,7 @@ pub(crate) mod tests {
         option.major_default_oldest_table_num = 1;
         option.trigger_type = TriggerType::Length(/* max_mutable_len */ 50);
 
-        let db: DB<Test, TokioExecutor> = DB::new(option, TokioExecutor::new(), TestSchema)
+        let db: DB<Test, TokioExecutor> = DB::new(option, TokioExecutor::current(), TestSchema)
             .await
             .unwrap();
 
@@ -1701,10 +1701,13 @@ pub(crate) mod tests {
         schema.flush_wal().await.unwrap();
         drop(schema);
 
-        let db: DB<Test, TokioExecutor> =
-            DB::new(option.as_ref().to_owned(), TokioExecutor::new(), TestSchema)
-                .await
-                .unwrap();
+        let db: DB<Test, TokioExecutor> = DB::new(
+            option.as_ref().to_owned(),
+            TokioExecutor::current(),
+            TestSchema,
+        )
+        .await
+        .unwrap();
 
         let mut sort_items = BTreeMap::new();
         for item in test_items() {
@@ -1781,7 +1784,7 @@ pub(crate) mod tests {
         );
         let dyn_schema = test_dyn_item_schema();
         let db: DB<DynRecord, TokioExecutor> =
-            DB::with_schema(option, TokioExecutor::new(), dyn_schema)
+            DB::with_schema(option, TokioExecutor::current(), dyn_schema)
                 .await
                 .unwrap();
 
@@ -1823,7 +1826,7 @@ pub(crate) mod tests {
         option.major_threshold_with_sst_size = 3;
         option.major_default_oldest_table_num = 1;
         option.trigger_type = TriggerType::Length(5);
-        let db: DB<Test, TokioExecutor> = DB::new(option, TokioExecutor::new(), TestSchema)
+        let db: DB<Test, TokioExecutor> = DB::new(option, TokioExecutor::current(), TestSchema)
             .await
             .unwrap();
 
@@ -1868,7 +1871,7 @@ pub(crate) mod tests {
         option.trigger_type = TriggerType::Length(5);
 
         let db: DB<DynRecord, TokioExecutor> =
-            DB::with_schema(option, TokioExecutor::new(), dyn_schema)
+            DB::with_schema(option, TokioExecutor::current(), dyn_schema)
                 .await
                 .unwrap();
 
@@ -2098,15 +2101,15 @@ pub(crate) mod tests {
         option3.trigger_type = TriggerType::Length(5);
 
         let db1: DB<DynRecord, TokioExecutor> =
-            DB::with_schema(option, TokioExecutor::new(), test_dyn_item_schema())
+            DB::with_schema(option, TokioExecutor::current(), test_dyn_item_schema())
                 .await
                 .unwrap();
         let db2: DB<DynRecord, TokioExecutor> =
-            DB::with_schema(option2, TokioExecutor::new(), test_dyn_item_schema())
+            DB::with_schema(option2, TokioExecutor::current(), test_dyn_item_schema())
                 .await
                 .unwrap();
         let db3: DB<DynRecord, TokioExecutor> =
-            DB::with_schema(option3, TokioExecutor::new(), test_dyn_item_schema())
+            DB::with_schema(option3, TokioExecutor::current(), test_dyn_item_schema())
                 .await
                 .unwrap();
 

diff --git a/src/ondisk/sstable.rs b/src/ondisk/sstable.rs
@@ -197,7 +197,7 @@ pub(crate) mod tests {
                 Path::from_filesystem_path(temp_dir.path()).unwrap(),
                 &TestSchema,
             )),
-            TokioExecutor::new(),
+            TokioExecutor::current(),
         )
         .await;
         let table_path = temp_dir.path().join("projection_query_test.parquet");
@@ -275,7 +275,7 @@ pub(crate) mod tests {
                 Path::from_filesystem_path(temp_dir.path()).unwrap(),
                 &TestSchema,
             )),
-            TokioExecutor::new(),
+            TokioExecutor::current(),
         )
         .await;
         let table_path = temp_dir.path().join("projection_scan_test.parquet");

diff --git a/src/snapshot.rs b/src/snapshot.rs
@@ -166,7 +166,7 @@ mod tests {
         let db = build_db(
             option,
             compaction_rx,
-            TokioExecutor::new(),
+            TokioExecutor::current(),
             schema,
             Arc::new(TestSchema),
             version,

diff --git a/src/transaction.rs b/src/transaction.rs
@@ -275,7 +275,7 @@ mod tests {
                 Path::from_filesystem_path(temp_dir.path()).unwrap(),
                 &StringSchema,
             )),
-            TokioExecutor::new(),
+            TokioExecutor::current(),
             StringSchema,
         )
         .await
@@ -333,7 +333,7 @@ mod tests {
         let db = build_db(
             option,
             compaction_rx,
-            TokioExecutor::new(),
+            TokioExecutor::current(),
             schema,
             Arc::new(TestSchema),
             version,
@@ -407,7 +407,7 @@ mod tests {
             &StringSchema,
         ));
 
-        let db = DB::<String, TokioExecutor>::new(option, TokioExecutor::new(), StringSchema)
+        let db = DB::<String, TokioExecutor>::new(option, TokioExecutor::current(), StringSchema)
             .await
             .unwrap();
 
@@ -443,7 +443,7 @@ mod tests {
             &TestSchema,
         ));
 
-        let db = DB::<Test, TokioExecutor>::new(option, TokioExecutor::new(), TestSchema)
+        let db = DB::<Test, TokioExecutor>::new(option, TokioExecutor::current(), TestSchema)
             .await
             .unwrap();
 
@@ -502,7 +502,7 @@ mod tests {
         let db = build_db(
             option,
             compaction_rx,
-            TokioExecutor::new(),
+            TokioExecutor::current(),
             schema,
             Arc::new(TestSchema),
             version,
@@ -599,7 +599,7 @@ mod tests {
         let db = build_db(
             option,
             compaction_rx,
-            TokioExecutor::new(),
+            TokioExecutor::current(),
             schema,
             Arc::new(TestSchema),
             version,
@@ -777,7 +777,7 @@ mod tests {
         let db = build_db(
             option,
             compaction_rx,
-            TokioExecutor::new(),
+            TokioExecutor::current(),
             schema,
             Arc::new(TestSchema),
             version,
@@ -829,7 +829,7 @@ mod tests {
             "age".to_string(),
             0,
         );
-        let db = DB::with_schema(option, TokioExecutor::default(), test_dyn_item_schema())
+        let db = DB::with_schema(option, TokioExecutor::current(), test_dyn_item_schema())
             .await
             .unwrap();
 

diff --git a/src/version/cleaner.rs b/src/version/cleaner.rs
@@ -159,7 +159,7 @@ pub(crate) mod tests {
 
         let (mut cleaner, tx) = Cleaner::<Test>::new(option.clone(), manager.clone());
 
-        let executor = TokioExecutor::new();
+        let executor = TokioExecutor::current();
 
         executor.spawn(async move {
             if let Err(err) = cleaner.listen().await {

diff --git a/tests/data_integrity.rs b/tests/data_integrity.rs
@@ -75,7 +75,7 @@ mod tests {
             &CustomerSchema,
         ));
 
-        let db: DB<Customer, TokioExecutor> = DB::new(option, TokioExecutor::new(), CustomerSchema)
+        let db: DB<Customer, TokioExecutor> = DB::new(option, TokioExecutor::current(), CustomerSchema)
             .await
             .unwrap();