feat: add CacheReader for SStable

.github/workflows/ci.yml

@@ -93,7 +93,6 @@ jobs:
       contents: write
       pull-requests: write
       repository-projects: write
-    if: github.event_name == 'pull_request'
     steps:
       - uses: actions/checkout@v4
       - name: Install latest nightly

Cargo.toml

@@ -1,4 +1,4 @@
-workspace = { members = ["tonbo_macros"] }
+workspace = { members = [ "tonbo_ext_reader","tonbo_macros"] }
 
 [package]
 description = "An embedded persistent KV database in Rust."
@@ -11,13 +11,14 @@ resolver = "2"
 version = "0.2.0"
 
 [package.metadata]
-msrv = "1.79.0"
+msrv = "1.82.0"
 
 [features]
-bench = ["redb", "rocksdb", "sled"]
+bench = ["redb", "rocksdb", "sled", "foyer"]
 bytes = ["dep:bytes"]
 datafusion = ["dep:async-trait", "dep:datafusion"]
 default = ["bytes", "tokio"]
+foyer = ["tonbo_ext_reader/foyer"]
 load_tbl = []
 redb = ["dep:redb"]
 rocksdb = ["dep:rocksdb"]
@@ -58,19 +59,19 @@ crc32fast = "1"
 crossbeam-skiplist = "0.1"
 datafusion = { version = "42", optional = true }
 flume = { version = "0.11", features = ["async"] }
-fusio = { package = "fusio", version = "0.3.1", features = [
+fusio = { package = "fusio", version = "0.3.3", features = [
     "aws",
     "dyn",
     "fs",
     "object_store",
     "tokio",
     "tokio-http",
 ] }
-fusio-dispatch = { package = "fusio-dispatch", version = "0.2.1", features = [
+fusio-dispatch = { package = "fusio-dispatch", version = "0.2.2", features = [
     "aws",
     "tokio",
 ] }
-fusio-parquet = { package = "fusio-parquet", version = "0.2.1" }
+fusio-parquet = { package = "fusio-parquet", version = "0.2.2" }
 futures-core = "0.3"
 futures-io = "0.3"
 futures-util = "0.3"
@@ -83,6 +84,7 @@ thiserror = "1"
 tokio = { version = "1", features = ["io-util"], default-features = false }
 tokio-util = { version = "0.7" }
 tonbo_macros = { version = "0.2.0", path = "tonbo_macros" }
+tonbo_ext_reader = { version = "0.1.0", path = "tonbo_ext_reader" }
 tracing = "0.1"
 ulid = "1"
 

benches/common.rs

@@ -7,6 +7,7 @@ use std::{
 };
 
 use async_stream::stream;
+use fusio_dispatch::FsOptions;
 use futures_core::Stream;
 use futures_util::StreamExt;
 use parquet::data_type::AsBytes;
@@ -16,6 +17,7 @@ use tokio::fs::create_dir_all;
 use tonbo::{
     executor::tokio::TokioExecutor, stream, transaction::TransactionEntry, DbOption, Projection,
 };
+use tonbo_ext_reader::{foyer_reader::FoyerReader, lru_reader::LruReader, CacheReader};
 use tonbo_macros::Record;
 
 const RNG_SEED: u64 = 3;
@@ -184,24 +186,164 @@ pub trait BenchReader {
     ) -> impl Stream<Item = ProjectionResult> + 'a;
 }
 
+pub struct TonboS3BenchDataBase {
+    db: tonbo::DB<Customer, TokioExecutor, FoyerReader>,
+}
+
+impl TonboS3BenchDataBase {
+    #[allow(dead_code)]
+    pub fn new(db: tonbo::DB<Customer, TokioExecutor, FoyerReader>) -> Self {
+        TonboS3BenchDataBase { db }
+    }
+}
+
+impl BenchDatabase for TonboS3BenchDataBase {
+    type W<'db>
+        = TonboBenchWriteTransaction<'db, FoyerReader>
+    where
+        Self: 'db;
+    type R<'db>
+        = TonboBenchReadTransaction<'db, FoyerReader>
+    where
+        Self: 'db;
+
+    fn db_type_name() -> &'static str {
+        "tonbo on s3"
+    }
+
+    async fn write_transaction(&self) -> Self::W<'_> {
+        TonboBenchWriteTransaction {
+            txn: self.db.transaction().await,
+        }
+    }
+
+    async fn read_transaction(&self) -> Self::R<'_> {
+        TonboBenchReadTransaction {
+            txn: self.db.transaction().await,
+        }
+    }
+
+    async fn build(path: impl AsRef<Path>) -> Self {
+        create_dir_all(path.as_ref()).await.unwrap();
+
+        let fs_options = FsOptions::S3 {
+            bucket: "data".to_string(),
+            credential: Some(fusio::remotes::aws::credential::AwsCredential {
+                key_id: "user".to_string(),
+                secret_key: "password".to_string(),
+                token: None,
+            }),
+            endpoint: Some("http://localhost:9000".to_string()),
+            sign_payload: None,
+            checksum: None,
+            region: None,
+        };
+
+        let path = fusio::path::Path::from_filesystem_path(path.as_ref()).unwrap();
+        let option = DbOption::from(path.clone())
+            .level_path(
+                0,
+                fusio::path::Path::from_url_path("/l0").unwrap(),
+                fs_options.clone(),
+            )
+            .unwrap()
+            .level_path(
+                1,
+                fusio::path::Path::from_url_path("/l1").unwrap(),
+                fs_options.clone(),
+            )
+            .unwrap()
+            .level_path(
+                2,
+                fusio::path::Path::from_url_path("/l2").unwrap(),
+                fs_options.clone(),
+            )
+            .unwrap()
+            .level_path(
+                3,
+                fusio::path::Path::from_url_path("/l3").unwrap(),
+                fs_options.clone(),
+            )
+            .unwrap()
+            .level_path(
+                4,
+                fusio::path::Path::from_url_path("/l4").unwrap(),
+                fs_options.clone(),
+            )
+            .unwrap()
+            .disable_wal();
+
+        TonboS3BenchDataBase::new(tonbo::DB::new(option, TokioExecutor::new()).await.unwrap())
+    }
+}
+
+pub struct TonboFoyerBenchDataBase {
+    db: tonbo::DB<Customer, TokioExecutor, FoyerReader>,
+}
+
+impl TonboFoyerBenchDataBase {
+    #[allow(dead_code)]
+    pub fn new(db: tonbo::DB<Customer, TokioExecutor, FoyerReader>) -> Self {
+        TonboFoyerBenchDataBase { db }
+    }
+}
+
+impl BenchDatabase for TonboFoyerBenchDataBase {
+    type W<'db>
+        = TonboBenchWriteTransaction<'db, FoyerReader>
+    where
+        Self: 'db;
+    type R<'db>
+        = TonboBenchReadTransaction<'db, FoyerReader>
+    where
+        Self: 'db;
+
+    fn db_type_name() -> &'static str {
+        "tonbo_foyer"
+    }
+
+    async fn write_transaction(&self) -> Self::W<'_> {
+        TonboBenchWriteTransaction {
+            txn: self.db.transaction().await,
+        }
+    }
+
+    async fn read_transaction(&self) -> Self::R<'_> {
+        TonboBenchReadTransaction {
+            txn: self.db.transaction().await,
+        }
+    }
+
+    async fn build(path: impl AsRef<Path>) -> Self {
+        create_dir_all(path.as_ref()).await.unwrap();
+
+        let option =
+            DbOption::from(fusio::path::Path::from_filesystem_path(path.as_ref()).unwrap())
+                .disable_wal();
+
+        let db = tonbo::DB::new(option, TokioExecutor::new()).await.unwrap();
+        TonboFoyerBenchDataBase::new(db)
+    }
+}
+
 pub struct TonboBenchDataBase {
-    db: tonbo::DB<Customer, TokioExecutor>,
+    db: tonbo::DB<Customer, TokioExecutor, LruReader>,
 }
 
 impl TonboBenchDataBase {
     #[allow(dead_code)]
-    pub fn new(db: tonbo::DB<Customer, TokioExecutor>) -> Self {
+    pub fn new(db: tonbo::DB<Customer, TokioExecutor, LruReader>) -> Self {
         TonboBenchDataBase { db }
     }
 }
 
 impl BenchDatabase for TonboBenchDataBase {
     type W<'db>
-        = TonboBenchWriteTransaction<'db>
+        = TonboBenchWriteTransaction<'db, LruReader>
     where
         Self: 'db;
     type R<'db>
-        = TonboBenchReadTransaction<'db>
+        = TonboBenchReadTransaction<'db, LruReader>
     where
         Self: 'db;
 
@@ -233,13 +375,13 @@ impl BenchDatabase for TonboBenchDataBase {
     }
 }
 
-pub struct TonboBenchReadTransaction<'a> {
-    txn: tonbo::transaction::Transaction<'a, Customer>,
+pub struct TonboBenchReadTransaction<'a, R: CacheReader> {
+    txn: tonbo::transaction::Transaction<'a, Customer, R>,
 }
 
-impl<'db> BenchReadTransaction for TonboBenchReadTransaction<'db> {
+impl<'db, R: CacheReader + 'static> BenchReadTransaction for TonboBenchReadTransaction<'db, R> {
     type T<'txn>
-        = TonboBenchReader<'db, 'txn>
+        = TonboBenchReader<'db, 'txn, R>
     where
         Self: 'txn;
 
@@ -248,11 +390,11 @@ impl<'db> BenchReadTransaction for TonboBenchReadTransaction<'db> {
     }
 }
 
-pub struct TonboBenchReader<'db, 'txn> {
-    txn: &'txn tonbo::transaction::Transaction<'db, Customer>,
+pub struct TonboBenchReader<'db, 'txn, R: CacheReader> {
+    txn: &'txn tonbo::transaction::Transaction<'db, Customer, R>,
 }
 
-impl BenchReader for TonboBenchReader<'_, '_> {
+impl<R: CacheReader + 'static> BenchReader for TonboBenchReader<'_, '_, R> {
     async fn get<'a>(&'a self, key: &'a ItemKey) -> Option<BenchResult> {
         self.txn
             .get(key, Projection::All)
@@ -288,13 +430,13 @@ impl BenchReader for TonboBenchReader<'_, '_> {
     }
 }
 
-pub struct TonboBenchWriteTransaction<'a> {
-    txn: tonbo::transaction::Transaction<'a, Customer>,
+pub struct TonboBenchWriteTransaction<'a, R: CacheReader> {
+    txn: tonbo::transaction::Transaction<'a, Customer, R>,
 }
 
-impl<'db> BenchWriteTransaction for TonboBenchWriteTransaction<'db> {
+impl<'db, R: CacheReader + 'static> BenchWriteTransaction for TonboBenchWriteTransaction<'db, R> {
     type W<'txn>
-        = TonboBenchInserter<'db, 'txn>
+        = TonboBenchInserter<'db, 'txn, R>
     where
         Self: 'txn;
 
@@ -308,11 +450,11 @@ impl<'db> BenchWriteTransaction for TonboBenchWriteTransaction<'db> {
     }
 }
 
-pub struct TonboBenchInserter<'db, 'txn> {
-    txn: &'txn mut tonbo::transaction::Transaction<'db, Customer>,
+pub struct TonboBenchInserter<'db, 'txn, R: CacheReader> {
+    txn: &'txn mut tonbo::transaction::Transaction<'db, Customer, R>,
 }
 
-impl BenchInserter for TonboBenchInserter<'_, '_> {
+impl<R: CacheReader + 'static> BenchInserter for TonboBenchInserter<'_, '_, R> {
     fn insert(&mut self, record: Customer) -> Result<(), ()> {
         self.txn.insert(record);
         Ok(())

benches/criterion/writes.rs

@@ -3,6 +3,7 @@ use std::{iter::repeat_with, sync::Arc};
 use criterion::{criterion_group, criterion_main, BenchmarkId, Criterion};
 use mimalloc::MiMalloc;
 use tonbo::{executor::tokio::TokioExecutor, DbOption, Record, DB};
+use tonbo_ext_reader::foyer_reader::FoyerReader;
 
 #[global_allocator]
 static GLOBAL: MiMalloc = MiMalloc;
@@ -15,7 +16,7 @@ pub struct KV {
 }
 
 #[inline(never)]
-async fn tonbo_write(db: &DB<KV, TokioExecutor>, batch_size: usize) {
+async fn tonbo_write(db: &DB<KV, TokioExecutor, FoyerReader>, batch_size: usize) {
     let mut kvs = Vec::with_capacity(128);
     for _ in 0..batch_size {
         let key = repeat_with(fastrand::alphanumeric).take(256).collect();

benches/read_bench.rs

@@ -12,7 +12,8 @@ use tokio::{fs, io::AsyncWriteExt};
 
 use crate::common::{
     read_tbl, BenchDatabase, BenchReadTransaction, BenchReader, RedbBenchDatabase,
-    RocksdbBenchDatabase, SledBenchDatabase, TonboBenchDataBase, ITERATIONS, NUM_SCAN, READ_TIMES,
+    RocksdbBenchDatabase, SledBenchDatabase, TonboBenchDataBase, TonboFoyerBenchDataBase,
+    TonboS3BenchDataBase, ITERATIONS, NUM_SCAN, READ_TIMES,
 };
 
 async fn benchmark<T: BenchDatabase + Send + Sync>(
@@ -152,26 +153,37 @@ async fn main() {
 
         load::<TonboBenchDataBase>(&tbl_path, data_dir.join("tonbo")).await;
         load::<RocksdbBenchDatabase>(&tbl_path, data_dir.join("rocksdb")).await;
+        load::<TonboFoyerBenchDataBase>(&tbl_path, data_dir.join("tonbo_foyer")).await;
+        load::<TonboS3BenchDataBase>(&tbl_path, data_dir.join("tonbo_s3")).await;
     }
 
     let tonbo_latency_results = { benchmark::<TonboBenchDataBase>(data_dir.join("tonbo")).await };
+    let tonbo_foyer_latency_results =
+        { benchmark::<TonboFoyerBenchDataBase>(data_dir.join("tonbo_foyer")).await };
     let rocksdb_results = { benchmark::<RocksdbBenchDatabase>(data_dir.join("rocksdb")).await };
+    let tonbo_s3_latency_results =
+        { benchmark::<TonboS3BenchDataBase>(data_dir.join("tonbo_s3")).await };
 
     let mut rows: Vec<Vec<String>> = Vec::new();
 
     for (benchmark, _duration) in &tonbo_latency_results {
         rows.push(vec![benchmark.to_string()]);
     }
 
-    for results in [tonbo_latency_results, rocksdb_results] {
+    for results in [
+        tonbo_latency_results,
+        tonbo_foyer_latency_results,
+        rocksdb_results,
+        tonbo_s3_latency_results,
+    ] {
         for (i, (_benchmark, duration)) in results.iter().enumerate() {
             rows[i].push(format!("{}ms", duration.as_millis()));
         }
     }
 
     let mut table = comfy_table::Table::new();
     table.set_width(100);
-    table.set_header(["", "tonbo", "rocksdb"]);
+    table.set_header(["", "tonbo", "tonbo_foyer", "rocksdb", "tonbo_s3"]);
     for row in rows {
         table.add_row(row);
     }