feat: enable levels to cache sstables to local disk if uploaded to S3

benches/common.rs

@@ -257,30 +257,35 @@ impl BenchDatabase for TonboS3BenchDataBase {
                 0,
                 fusio::path::Path::from_url_path("/l0").unwrap(),
                 fs_options.clone(),
+                false,
             )
             .unwrap()
             .level_path(
                 1,
                 fusio::path::Path::from_url_path("/l1").unwrap(),
                 fs_options.clone(),
+                false,
             )
             .unwrap()
             .level_path(
                 2,
                 fusio::path::Path::from_url_path("/l2").unwrap(),
                 fs_options.clone(),
+                false,
             )
             .unwrap()
             .level_path(
                 3,
                 fusio::path::Path::from_url_path("/l3").unwrap(),
                 fs_options.clone(),
+                false,
             )
             .unwrap()
             .level_path(
                 4,
                 fusio::path::Path::from_url_path("/l4").unwrap(),
                 fs_options.clone(),
+                false,
             )
             .unwrap()
             .disable_wal();

bindings/js/src/db.rs

@@ -349,11 +349,11 @@ mod tests {
 
         let option = DbOption::new("write_s3".to_string())
             .expect("cannot open DB")
-            .level_path(0, "js/l0".to_string(), fs_option.clone())
+            .level_path(0, "js/l0".to_string(), fs_option.clone(), true)
             .unwrap()
-            .level_path(1, "js/l1".to_string(), fs_option.clone())
+            .level_path(1, "js/l1".to_string(), fs_option.clone(), false)
             .unwrap()
-            .level_path(2, "js/l2".to_string(), fs_option.clone())
+            .level_path(2, "js/l2".to_string(), fs_option.clone(), true)
             .unwrap();
 
         let schema = schema();

bindings/js/src/options.rs

@@ -26,7 +26,7 @@ pub struct DbOption {
     /// build the `DB` storage directory based on the passed path
     path: String,
     base_fs: FsOptions,
-    level_paths: Vec<Option<(String, FsOptions)>>,
+    level_paths: Vec<Option<(String, FsOptions, bool)>>,
 }
 
 #[wasm_bindgen]
@@ -57,8 +57,9 @@ impl DbOption {
         level: usize,
         path: String,
         fs_options: FsOptions,
+        cached: bool,
     ) -> Result<Self, JsValue> {
-        self.level_paths[level] = Some((path.to_string(), fs_options));
+        self.level_paths[level] = Some((path.to_string(), fs_options, cached));
         Ok(self)
     }
 }
@@ -77,10 +78,10 @@ impl DbOption {
             .base_fs(self.base_fs.into_fs_options());
 
         for (level, path) in self.level_paths.into_iter().enumerate() {
-            if let Some((path, fs_options)) = path {
+            if let Some((path, fs_options, cached)) = path {
                 let path = fs_options.path(path).unwrap();
                 opt = opt
-                    .level_path(level, path, fs_options.into_fs_options())
+                    .level_path(level, path, fs_options.into_fs_options(), cached)
                     .unwrap();
             }
         }

bindings/python/example/fusion_storage.py

@@ -21,8 +21,8 @@ async def main():
         os.makedirs("db_path/user/l1")
 
     option = DbOption(from_filesystem_path("db_path/user"))
-    option.level_path(0, from_filesystem_path("db_path/user/l0"), FsOptions.Local())
-    option.level_path(1, from_filesystem_path("db_path/user/l1"), FsOptions.Local())
+    option.level_path(0, from_filesystem_path("db_path/user/l0"), FsOptions.Local(), False)
+    option.level_path(1, from_filesystem_path("db_path/user/l1"), FsOptions.Local(), False)
 
     option.immutable_chunk_num = 1
     option.major_threshold_with_sst_size = 3

bindings/python/src/options.rs

@@ -38,7 +38,7 @@ pub struct DbOption {
     path: String,
     #[pyo3(get, set)]
     base_fs: FsOptions,
-    level_paths: Vec<Option<(String, FsOptions)>>,
+    level_paths: Vec<Option<(String, FsOptions, bool)>>,
 }
 
 #[pymethods]
@@ -61,11 +61,11 @@ impl DbOption {
         }
     }
 
-    fn level_path(&mut self, level: usize, path: String, fs_options: FsOptions) -> PyResult<()> {
+    fn level_path(&mut self, level: usize, path: String, fs_options: FsOptions, cached: bool) -> PyResult<()> {
         if level >= MAX_LEVEL {
             ExceedsMaxLevelError::new_err("Exceeds max level");
         }
-        self.level_paths[level] = Some((path, fs_options));
+        self.level_paths[level] = Some((path, fs_options, cached));
         Ok(())
     }
 }
@@ -82,12 +82,13 @@ impl DbOption {
             .version_log_snapshot_threshold(self.version_log_snapshot_threshold)
             .base_fs(tonbo::option::FsOptions::from(self.base_fs));
         for (level, path) in self.level_paths.into_iter().enumerate() {
-            if let Some((path, fs_options)) = path {
+            if let Some((path, fs_options, cached)) = path {
                 opt = opt
                     .level_path(
                         level,
                         Path::from(path),
                         tonbo::option::FsOptions::from(fs_options),
+                        cached,
                     )
                     .unwrap();
             }

bindings/python/tests/test_s3.py

@@ -27,9 +27,9 @@ async def test_s3_read_write():
     fs_option = FsOptions.S3("wasm-data", credential,"ap-southeast-2",None, None, None)
 
     option = DbOption(temp_dir.name)
-    option.level_path(0, from_url_path("l0"), fs_option)
-    option.level_path(1, from_url_path("l1"), fs_option)
-    option.level_path(2, from_url_path("l2"), fs_option)
+    option.level_path(0, from_url_path("l0"), fs_option, False)
+    option.level_path(1, from_url_path("l1"), fs_option, False)
+    option.level_path(2, from_url_path("l2"), fs_option, False)
 
     option.immutable_chunk_num = 1
     option.major_threshold_with_sst_size = 3

bindings/python/tests/test_table_level.py

@@ -36,8 +36,8 @@ async def test_table_level_local():
     temp_dir1 = tempfile.TemporaryDirectory()
 
     option = DbOption(from_filesystem_path(temp_dir.name))
-    option.level_path(0, from_filesystem_path(temp_dir0.name), FsOptions.Local())
-    option.level_path(1, from_filesystem_path(temp_dir1.name), FsOptions.Local())
+    option.level_path(0, from_filesystem_path(temp_dir0.name), FsOptions.Local(), False)
+    option.level_path(1, from_filesystem_path(temp_dir1.name), FsOptions.Local(), False)
 
     option.immutable_chunk_num = 1
     option.major_threshold_with_sst_size = 3

guide/src/start.md

@@ -219,8 +219,8 @@ let s3_option = FsOptions::S3 {
 let options = DbOption::new(
     Path::from_filesystem_path("./db_path/users").unwrap(),
     &UserSchema,
-).level_path(2, "l2", s3_option.clone())
-).level_path(3, "l3", s3_option);
+).level_path(2, "l2", s3_option.clone(), false)
+).level_path(3, "l3", s3_option, true);
 ```
 
 In this example, data for level 2 and level 3 will be stored in S3, while all other levels remain on the local disk. If there is data in level 2 and level 3, you can verify and access it in S3:

guide/src/usage/advance.md

@@ -256,7 +256,7 @@ async fn main() {
     ];
     let schema = DynSchema::new(descs, 0);
     let options = DbOption::new(Path::from_filesystem_path("s3_path").unwrap(), &schema)
-        .level_path(2, "l2", fs_option);
+        .level_path(2, "l2", fs_option, true);
 
 
     let db = DB::<DynRecord, TokioExecutor>::new(options, TokioExecutor::current(), schema)

guide/src/usage/tonbo.md

@@ -202,7 +202,7 @@ async fn main() {
     };
 
     let options = DbOption::new(Path::from_filesystem_path("s3_path").unwrap(), &UserSchema)
-        .level_path(2, "l2", fs_option);
+        .level_path(2, "l2", fs_option, false);
 
     let db = DB::<User, TokioExecutor>::new(options, TokioExecutor::current(), UserSchema)
         .await