refactor: Added RwLock abstraction to the Executor trait

Author: ethe

CLAUDE.md

@@ -13,8 +13,9 @@ Tonbo is an embedded, persistent key-value database written in Rust that uses Lo
 ```bash
 # Code quality checks
 cargo check                    # Check for compilation errors
-cargo clippy                   # Run linter for code improvements
+cargo clippy --workspace -- -D warnings  # Run linter (fail on warnings)
 cargo +nightly fmt             # Format code (must use nightly)
+cargo +nightly fmt -- --check  # Check formatting without changing files
 
 # Building
 cargo build                    # Standard debug build
@@ -54,11 +55,12 @@ cargo build --features datafusion
 cd bindings/python
 maturin develop              # Build and install locally for development
 pytest tests/                 # Run Python tests
+pytest tests/bench/ --benchmark-only  # Run Python benchmarks
 
 # JavaScript/WASM bindings (requires wasm-pack)
 cd bindings/js
 wasm-pack build              # Build WASM module
-wasm-pack test --headless    # Run WASM tests
+wasm-pack test --chrome --headless  # Run WASM tests in Chrome
 ```
 
 ### Code Coverage
@@ -71,14 +73,18 @@ cargo llvm-cov --workspace --lcov --output-path lcov.info
 
 The codebase is organized into several key modules:
 
-- **src/compaction/**: LSM tree compaction strategies and implementation
+- **src/compaction/**: LSM tree compaction strategies (leveled compaction implementation)
 - **src/fs/**: File system abstractions supporting multiple backends (local, S3, OPFS)
 - **src/inmem/**: In-memory data structures including mutable and immutable memtables
 - **src/ondisk/**: On-disk storage using SSTable format with Arrow/Parquet
 - **src/record/**: Record types, schemas, and the `#[derive(Record)]` macro system
-- **src/stream/**: Streaming operations for efficient data processing
+  - **dynamic/**: Runtime-defined record schemas with Value enum for flexible typing
+  - **key/**: Type implementations for primary keys (strings, numbers, timestamps, lists)
+- **src/stream/**: Streaming operations for efficient data processing and merging
 - **src/version/**: Version management and multi-version concurrency control
 - **src/wal/**: Write-ahead logging for durability
+- **src/transaction.rs**: Transaction support with optimistic concurrency control
+- **src/manifest.rs**: Database manifest management
 
 The project uses procedural macros (in `tonbo_macros/`) to provide a type-safe API where users define their key-value schema using Rust structs.
 
@@ -89,38 +95,35 @@ The project uses procedural macros (in `tonbo_macros/`) to provide a type-safe A
 2. **Feature Flags**: Important features include:
    - `tokio`: Async filesystem operations (default)
    - `wasm`: WebAssembly support with OPFS backend
-   - `datafusion`: SQL query support
-   - `aws`: S3 storage backend
-   - `bench`: Enables comparison benchmarks
+   - `datafusion`: SQL query support via Apache DataFusion
+   - `aws`: S3 storage backend support
+   - `bench`: Enables comparison benchmarks against RocksDB, Sled, and Redb
+   - `bytes`: Bytes type support (default)
 
-3. **Type System**: Records must derive from `tonbo::Record` trait. The macro generates necessary serialization and Arrow schema implementations.
+3. **Type System**: Records must derive from `tonbo::Record` trait. The macro generates necessary serialization and Arrow schema implementations. Dynamic records are also supported for runtime-defined schemas.
 
-4. **Testing**: Tests should cover both sync and async APIs when applicable. Integration tests go in `tests/` directory.
+4. **Testing**: 
+   - Unit tests use `#[tokio::test]` for async code
+   - Integration tests in `tests/` directory
+   - Macro correctness tests using trybuild in `tests/success/` and `tests/fail/`
+   - WASM-specific tests in `tests/wasm.rs`
 
-5. **Error Handling**: Uses custom error types defined in `src/error.rs`. Always propagate errors appropriately using `?`.
+5. **Error Handling**: Uses custom error types. Always propagate errors appropriately using `?`.
 
 ## Working with the Codebase
 
 1. **Adding New Features**: Check feature flags in `Cargo.toml` and ensure proper conditional compilation with `#[cfg(feature = "...")]`.
 
-2. **Modifying Storage Layer**: Changes to on-disk format should maintain backward compatibility or increment version numbers appropriately.
+2. **Modifying Storage Layer**: Changes to on-disk format should maintain backward compatibility or increment version numbers appropriately. The magic number in `src/magic.rs` helps identify file format versions.
 
-3. **Performance**: Run benchmarks before and after changes that might impact performance. Compare against baseline using `cargo bench`.
+3. **Performance**: Run benchmarks before and after changes that might impact performance. Use `cargo bench --features bench` to compare against other embedded databases.
 
-4. **Documentation**: Update both inline documentation and the guide (in `guide/` directory) for user-facing changes.
-
-5. **Cross-Platform**: Ensure changes work across all supported platforms (Linux, macOS, Windows) and targets (native, WASM).
-
-## Current Development Focus
-
-Based on recent commits, the project is actively developing:
-- Date/time type support (date32/date64/time32/time64)
-- Dynamic record implementation
-- Enhanced type system flexibility
+4. **Cross-Platform**: Ensure changes work across all supported platforms (Linux, macOS, Windows) and targets (native, WASM).
 
 ## Important Notes
 
 - Rust toolchain version is pinned to 1.85 in `rust-toolchain.toml`
-- The project uses specific git revisions for fusio-* dependencies
+- The project uses fusio crates (v0.4.0) for pluggable I/O and storage backends
 - Formatting is strict - always run `cargo +nightly fmt` before committing
-- The project follows semantic versioning (currently at 0.3.2)
+- The project follows semantic versioning (currently at 0.3.2)
+- Default features include: aws, bytes, tokio, tokio-http, async-trait

Cargo.toml

@@ -40,6 +40,7 @@ tokio = [
     "parquet/default",
     "tokio/fs",
     "tokio/rt-multi-thread",
+    "dep:tokio",
 ]
 tokio-http = ["fusio/tokio-http", "fusio-log/tokio-http"]
 wasm = ["aws", "bytes", "opfs", "wasm-http", "dep:async-trait"]
@@ -93,6 +94,7 @@ fusio-log = { version = "0.4.0", default-features = false, features = [
     "bytes",
 ] }
 fusio-parquet = { version = "0.4.0" }
+futures = "0.3"
 futures-core = "0.3"
 futures-util = "0.3"
 itertools = { version = "0.14.0" }
@@ -108,8 +110,10 @@ parquet = { version = "55", default-features = false, features = [
 ] }
 parquet-lru = { version = "0.3.0", path = "parquet-lru" }
 pin-project-lite = "0.2"
-thiserror = "2.0.3"
-tokio = { version = "1", features = ["io-util"], default-features = false }
+thiserror = "2.0.12"
+tokio = { version = "1.47.1", features = [
+    "io-util",
+], default-features = false, optional = true }
 tonbo_macros = { version = "0.3.1", path = "tonbo_macros" }
 tracing = "0.1"
 ulid = { version = "1", features = ["serde"] }

README.md

@@ -63,7 +63,7 @@ async fn main() {
         &UserSchema,
     );
 
-    let db = DB::new(options, TokioExecutor::current(), UserSchema)
+    let db = DB::new(options, TokioExecutor::default(), UserSchema)
         .await
         .unwrap();
 

benches/common.rs

@@ -286,7 +286,7 @@ impl BenchDatabase for TonboS3BenchDataBase {
             .disable_wal();
 
         TonboS3BenchDataBase::new(
-            tonbo::DB::new(option, TokioExecutor::current(), CustomerSchema)
+            tonbo::DB::new(option, TokioExecutor::default(), CustomerSchema)
                 .await
                 .unwrap(),
         )
@@ -339,7 +339,7 @@ impl BenchDatabase for TonboBenchDataBase {
         )
         .disable_wal();
 
-        let db = tonbo::DB::new(option, TokioExecutor::current(), CustomerSchema)
+        let db = tonbo::DB::new(option, TokioExecutor::default(), CustomerSchema)
             .await
             .unwrap();
         TonboBenchDataBase::new(db)

benches/criterion/writes.rs

@@ -64,7 +64,7 @@ fn single_write(c: &mut Criterion) {
         )
         .disable_wal();
         let db = runtime
-            .block_on(async { DB::new(option, TokioExecutor::current(), KVSchema).await })
+            .block_on(async { DB::new(option, TokioExecutor::default(), KVSchema).await })
             .unwrap();
 
         group.bench_with_input(BenchmarkId::new("Tonbo", batch), &batch, |b, batch| {

bindings/js/src/transaction.rs

@@ -3,6 +3,7 @@ use std::{mem::transmute, sync::Arc};
 use futures::StreamExt;
 use js_sys::Object;
 use tonbo::{
+    executor::opfs::OpfsExecutor,
     record::{DynRecord, DynamicField},
     transaction, Projection,
 };
@@ -15,22 +16,22 @@ use crate::{
 
 #[wasm_bindgen]
 pub struct Transaction {
-    txn: Option<transaction::Transaction<'static, DynRecord>>,
+    txn: Option<transaction::Transaction<'static, DynRecord, OpfsExecutor>>,
     desc: Arc<Vec<DynamicField>>,
     primary_key_index: usize,
 }
 
 impl Transaction {
     pub(crate) fn new<'txn>(
-        txn: transaction::Transaction<'txn, DynRecord>,
+        txn: transaction::Transaction<'txn, DynRecord, OpfsExecutor>,
         desc: Arc<Vec<DynamicField>>,
         primary_key_index: usize,
     ) -> Self {
         Transaction {
             txn: Some(unsafe {
                 transmute::<
-                    transaction::Transaction<'txn, DynRecord>,
-                    transaction::Transaction<'static, DynRecord>,
+                    transaction::Transaction<'txn, DynRecord, OpfsExecutor>,
+                    transaction::Transaction<'static, DynRecord, OpfsExecutor>,
                 >(txn)
             }),
             desc,
@@ -123,8 +124,8 @@ impl Transaction {
         let txn = self.txn.as_ref().unwrap();
         let txn = unsafe {
             transmute::<
-                &transaction::Transaction<'_, DynRecord>,
-                &'static transaction::Transaction<'_, DynRecord>,
+                &transaction::Transaction<'_, DynRecord, OpfsExecutor>,
+                &'static transaction::Transaction<'_, DynRecord, OpfsExecutor>,
             >(txn)
         };
         let mut scan = txn

bindings/js/src/utils.rs

@@ -71,7 +71,7 @@ pub(crate) fn to_record(schema: &[DynamicField], cols: &[Value]) -> JsValue {
             Value::Float64(v) => (*v).into(),
             Value::String(v) => v.to_owned().into(),
             Value::Binary(v) => v.to_vec().into(),
-            Value::FixedSizeBinary(v, w) => v.to_vec().into(),
+            Value::FixedSizeBinary(v, _w) => v.to_vec().into(),
             Value::Date32(_)
             | Value::Date64(_)
             | Value::List(_, _)

bindings/python/src/db.rs

@@ -58,7 +58,7 @@ impl TonboDB {
         let schema = DynSchema::new(desc, primary_key_index.unwrap());
         let option = option.into_option(&schema);
         let db = get_runtime()
-            .block_on(async { DB::new(option, TokioExecutor::current(), schema).await })
+            .block_on(async { DB::new(option, TokioExecutor::default(), schema).await })
             .unwrap();
         Ok(Self {
             db: Arc::new(db),

bindings/python/src/transaction.rs

@@ -7,6 +7,7 @@ use pyo3::{
 };
 use pyo3_async_runtimes::tokio::future_into_py;
 use tonbo::{
+    executor::tokio::TokioExecutor,
     record::{DynRecord, Value},
     transaction, Projection,
 };
@@ -21,14 +22,14 @@ use crate::{
 
 #[pyclass]
 pub struct Transaction {
-    txn: Option<transaction::Transaction<'static, DynRecord>>,
+    txn: Option<transaction::Transaction<'static, DynRecord, TokioExecutor>>,
     desc: Arc<Vec<Column>>,
     primary_key_index: usize,
 }
 
 impl Transaction {
     pub(crate) fn new<'txn>(
-        txn: transaction::Transaction<'txn, DynRecord>,
+        txn: transaction::Transaction<'txn, DynRecord, TokioExecutor>,
         desc: Arc<Vec<Column>>,
     ) -> Self {
         let primary_key_index = desc
@@ -40,8 +41,8 @@ impl Transaction {
         Transaction {
             txn: Some(unsafe {
                 transmute::<
-                    transaction::Transaction<'txn, DynRecord>,
-                    transaction::Transaction<'static, DynRecord>,
+                    transaction::Transaction<'txn, DynRecord, TokioExecutor>,
+                    transaction::Transaction<'static, DynRecord, TokioExecutor>,
                 >(txn)
             }),
             desc,
@@ -87,8 +88,8 @@ impl Transaction {
         let txn = self.txn.as_ref().unwrap();
         let txn = unsafe {
             transmute::<
-                &transaction::Transaction<'_, DynRecord>,
-                &'static transaction::Transaction<'_, DynRecord>,
+                &transaction::Transaction<'_, DynRecord, TokioExecutor>,
+                &'static transaction::Transaction<'_, DynRecord, TokioExecutor>,
             >(txn)
         };
 
@@ -176,8 +177,8 @@ impl Transaction {
         let txn = self.txn.as_ref().unwrap();
         let txn = unsafe {
             transmute::<
-                &transaction::Transaction<'_, DynRecord>,
-                &'static transaction::Transaction<'_, DynRecord>,
+                &transaction::Transaction<'_, DynRecord, TokioExecutor>,
+                &'static transaction::Transaction<'_, DynRecord, TokioExecutor>,
             >(txn)
         };
         let col_desc = self.desc.get(self.primary_key_index).unwrap();

examples/datafusion.rs

@@ -238,7 +238,7 @@ async fn main() -> Result<()> {
         &MusicSchema,
     );
 
-    let db = DB::new(options, TokioExecutor::current(), MusicSchema)
+    let db = DB::new(options, TokioExecutor::default(), MusicSchema)
         .await
         .unwrap();
     for (id, name, like) in [