Add tpcc sync switch and reuse LMDB point-lookup cursor

Makefile

@@ -9,7 +9,6 @@ TPCC_NUM_WARE ?= 1
 TPCC_PPROF_OUTPUT ?= /tmp/tpcc_lmdb.svg
 TPCC_HEAPTRACK_MEASURE_TIME ?= 300
 TPCC_HEAPTRACK_OUTPUT ?= /tmp/tpcc_lmdb_heaptrack
-TPCC_SQLITE_PROFILE ?= balanced
 CODECOV_OUTPUT ?= lcov.info
 COVERAGE_PROFILE_DIR ?= target/grcov/profraw
 COVERAGE_HTML_DIR ?= target/grcov/html
@@ -36,7 +35,7 @@ FUZZ_ARGS_sql_gen = -max_len=4096
 FUZZ_PASS_EXIT_sql_gen = 0
 COVERAGE_REPORT_ARGS ?= --llvm --ignore-not-existing --keep-only 'src/**' --ignore 'src/**/tests/**' --ignore 'tests/**' --ignore 'tpcc/**' --excl-start 'GRCOV_EXCL_START' --excl-stop 'GRCOV_EXCL_STOP'
 
-.PHONY: test test-python test-wasm test-slt test-all codecov codecov-html wasm-build check tpcc tpcc-kitesql-rocksdb tpcc-kitesql-lmdb tpcc-lmdb-flamegraph tpcc-lmdb-heaptrack tpcc-sqlite tpcc-sqlite-practical tpcc-sqlite-balanced tpcc-dual cargo-check build wasm-examples native-examples fmt clippy fuzz
+.PHONY: test test-python test-wasm test-slt test-all codecov codecov-html wasm-build check tpcc tpcc-kitesql-rocksdb tpcc-kitesql-lmdb tpcc-lmdb-flamegraph tpcc-lmdb-heaptrack tpcc-sqlite tpcc-dual cargo-check build wasm-examples native-examples fmt clippy fuzz
 
 ## Run default Rust tests in the current environment (non-WASM).
 test:
@@ -139,17 +138,9 @@ tpcc-lmdb-heaptrack:
 	@ls -1 $(TPCC_HEAPTRACK_OUTPUT)*
 	@echo "open gui: heaptrack_gui $$(ls -1 $(TPCC_HEAPTRACK_OUTPUT)* | tail -n 1)"
 
-## Execute the TPCC workload on SQLite with the practical profile.
+## Execute the TPCC workload on SQLite.
 tpcc-sqlite:
-	$(CARGO) run -p tpcc --release -- --backend sqlite --sqlite-profile $(TPCC_SQLITE_PROFILE) --path kite_sql_tpcc.sqlite
-
-## Execute the TPCC workload on SQLite with the practical profile.
-tpcc-sqlite-practical:
-	$(MAKE) tpcc-sqlite TPCC_SQLITE_PROFILE=practical
-
-## Execute the TPCC workload on SQLite with the balanced profile.
-tpcc-sqlite-balanced:
-	$(MAKE) tpcc-sqlite TPCC_SQLITE_PROFILE=balanced
+	$(CARGO) run -p tpcc --release -- --backend sqlite --path kite_sql_tpcc.sqlite
 
 ## Execute TPCC while mirroring every statement to an in-memory SQLite instance for validation.
 tpcc-dual:

scripts/run_tpcc_stable.py

@@ -26,7 +26,6 @@ class Variant:
     name: str
     backend: str
     db_name: str
-    sqlite_profile: str | None = None
 
 
 @dataclass(frozen=True)
@@ -44,16 +43,15 @@ def format(self) -> str:
 VARIANTS = [
     Variant("kitesql-lmdb", "kitesql-lmdb", "kitesql-lmdb"),
     Variant("kitesql-rocksdb", "kitesql-rocksdb", "kitesql-rocksdb"),
-    Variant("sqlite-balanced", "sqlite", "sqlite-balanced.sqlite", "balanced"),
-    Variant("sqlite-practical", "sqlite", "sqlite-practical.sqlite", "practical"),
+    Variant("sqlite", "sqlite", "sqlite.sqlite"),
 ]
 
 
 def parse_args() -> argparse.Namespace:
     parser = argparse.ArgumentParser(
         description=(
-            "Run kitesql-lmdb, kitesql-rocksdb, sqlite-balanced, and "
-            "sqlite-practical TPCC benchmarks with cooldown gates between runs."
+            "Run kitesql-lmdb, kitesql-rocksdb, and sqlite TPCC benchmarks "
+            "with cooldown gates between runs."
         )
     )
     parser.add_argument("--binary", type=Path, default=DEFAULT_BINARY)
@@ -354,8 +352,6 @@ def build_command(args: argparse.Namespace, variant: Variant, db_path: Path) ->
     ]
     if args.measure_time is not None:
         cmd.extend(["--measure-time", str(args.measure_time)])
-    if variant.sqlite_profile is not None:
-        cmd.extend(["--sqlite-profile", variant.sqlite_profile])
     return cmd
 
 

src/db.rs

@@ -180,6 +180,24 @@ impl DataBaseBuilder {
         self
     }
 
+    #[cfg(all(
+        not(target_arch = "wasm32"),
+        any(feature = "rocksdb", feature = "lmdb")
+    ))]
+    pub fn sync(mut self, enable: bool) -> Self {
+        #[cfg(feature = "rocksdb")]
+        {
+            self.storage_config.sync = enable;
+        }
+        #[cfg(feature = "lmdb")]
+        {
+            self.lmdb_config
+                .flags
+                .set(lmdb::EnvironmentFlags::NO_SYNC, !enable);
+        }
+        self
+    }
+
     /// Sets the LMDB map size in bytes.
     #[cfg(all(not(target_arch = "wasm32"), feature = "lmdb"))]
     pub fn lmdb_map_size(mut self, map_size: usize) -> Self {

src/storage/lmdb.rs

@@ -291,6 +291,11 @@ impl Transaction for LmdbTransaction<'_> {
         = LmdbIter<'a>
     where
         Self: 'a;
+
+    type PointCursor<'a>
+        = RoCursor<'a>
+    where
+        Self: 'a;
     fn next_statement_stamp(&mut self) -> Result<u64, DatabaseError> {
         const STATEMENT_BITS: u32 = 20;
         self.statements += 1;
@@ -315,6 +320,22 @@ impl Transaction for LmdbTransaction<'_> {
         }
     }
 
+    fn point_cursor(&self) -> Result<Self::PointCursor<'_>, DatabaseError> {
+        Ok(self.tx.open_ro_cursor(self.db)?)
+    }
+
+    fn get_at<'a>(
+        &'a self,
+        cursor: &Self::PointCursor<'a>,
+        key: &[u8],
+    ) -> Result<Option<Self::BorrowedBytes<'a>>, DatabaseError> {
+        match cursor.get(Some(key), None, lmdb_sys::MDB_SET) {
+            Ok((_, value)) => Ok(Some(value)),
+            Err(lmdb::Error::NotFound) => Ok(None),
+            Err(err) => Err(err.into()),
+        }
+    }
+
     fn set(&mut self, key: &[u8], value: &[u8]) -> Result<(), DatabaseError> {
         self.tx
             .put(self.db, &key, &value, lmdb::WriteFlags::empty())?;

src/storage/memory.rs

@@ -162,6 +162,16 @@ impl Transaction for MemoryTransaction {
         = MemoryIter<'a>
     where
         Self: 'a;
+
+    type PointCursor<'a>
+        = ()
+    where
+        Self: 'a;
+
+    fn point_cursor(&self) -> Result<Self::PointCursor<'_>, DatabaseError> {
+        Ok(())
+    }
+
     fn next_statement_stamp(&mut self) -> Result<u64, DatabaseError> {
         let stamp = self.statements.get() + 1;
         self.statements.set(stamp);

src/storage/mod.rs

@@ -166,6 +166,10 @@ pub trait Transaction: Sized {
     where
         Self: 'a;
 
+    type PointCursor<'a>
+    where
+        Self: 'a;
+
     fn next_statement_stamp(&mut self) -> Result<u64, DatabaseError> {
         Ok(0)
     }
@@ -276,6 +280,7 @@ pub trait Transaction: Sized {
                 deserializers,
                 total_len,
                 tx: self,
+                cursor: None,
                 cover_mapping,
                 with_pk,
                 reverse: ranges.reverse,
@@ -1162,6 +1167,16 @@ pub trait Transaction: Sized {
         key: &[u8],
     ) -> Result<Option<Self::BorrowedBytes<'a>>, DatabaseError>;
 
+    fn point_cursor(&self) -> Result<Self::PointCursor<'_>, DatabaseError>;
+
+    fn get_at<'a>(
+        &'a self,
+        _cursor: &Self::PointCursor<'a>,
+        key: &[u8],
+    ) -> Result<Option<Self::BorrowedBytes<'a>>, DatabaseError> {
+        self.get_borrowed(key)
+    }
+
     fn exists(&self, key: &[u8]) -> Result<bool, DatabaseError> {
         Ok(self.get_borrowed(key)?.is_some())
     }
@@ -1386,6 +1401,7 @@ struct IndexImplParams<'a, T: Transaction> {
     deserializers: Cow<'a, [TupleValueSerializableImpl]>,
     total_len: usize,
     tx: &'a T,
+    cursor: Option<T::PointCursor<'a>>,
     cover_mapping: Option<TupleMapping>,
     with_pk: bool,
     reverse: bool,
@@ -1578,7 +1594,11 @@ impl<T: Transaction> IndexImpl<T> for PrimaryKeyIndexImpl {
             value,
             None,
             |key, _| {
-                let Some(bytes) = params.tx.get_borrowed(key)? else {
+                let bytes = match &params.cursor {
+                    Some(cursor) => params.tx.get_at(cursor, key)?,
+                    None => params.tx.get_borrowed(key)?,
+                };
+                let Some(bytes) = bytes else {
                     return Ok(false);
                 };
                 TableCodec::decode_tuple_into(
@@ -2107,6 +2127,11 @@ impl<T: Transaction> Iter for IndexIter<'_, T> {
                             self.state = IndexIterState::Range(iter);
                         }
                         Range::Eq(val) => {
+                            if self.params.cursor.is_none()
+                                && matches!(self.inner, IndexImplEnum::PrimaryKey(_))
+                            {
+                                self.params.cursor = Some(self.params.tx.point_cursor()?);
+                            }
                             match self.inner.eq_to_res(
                                 table_codec,
                                 tuple,

src/storage/rocksdb.rs

@@ -21,8 +21,9 @@ use librocksdb_sys as ffi;
 use rocksdb::{
     checkpoint::Checkpoint,
     statistics::{StatsLevel, Ticker},
-    DBPinnableSlice, DBRawIteratorWithThreadMode, OptimisticTransactionDB, Options, ReadOptions,
-    SliceTransform, SnapshotWithThreadMode, TransactionDB,
+    DBPinnableSlice, DBRawIteratorWithThreadMode, OptimisticTransactionDB,
+    OptimisticTransactionOptions, Options, ReadOptions, SliceTransform, SnapshotWithThreadMode,
+    TransactionDB, TransactionOptions, WriteOptions,
 };
 use std::collections::Bound;
 #[cfg(feature = "unsafe_txdb_checkpoint")]
@@ -59,6 +60,13 @@ pub struct RocksDbMetrics {
 #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
 pub struct StorageConfig {
     pub enable_statistics: bool,
+    pub sync: bool,
+}
+
+fn sync_write_options() -> WriteOptions {
+    let mut options = WriteOptions::default();
+    options.set_sync(true);
+    options
 }
 
 impl RocksDbMetrics {
@@ -440,9 +448,17 @@ impl Storage for OptimisticRocksStorage {
         isolation: TransactionIsolationLevel,
     ) -> Result<Self::TransactionType<'_>, DatabaseError> {
         self.validate_transaction_isolation(isolation)?;
+        let tx = if self.config.sync {
+            self.inner.transaction_opt(
+                &sync_write_options(),
+                &OptimisticTransactionOptions::default(),
+            )
+        } else {
+            self.inner.transaction()
+        };
         Ok(OptimisticRocksTransaction {
             db: self.inner.as_ref(),
-            tx: self.inner.transaction(),
+            tx,
             isolation,
             current_snapshot: matches!(isolation, TransactionIsolationLevel::RepeatableRead)
                 .then(|| self.inner.snapshot()),
@@ -488,9 +504,15 @@ impl Storage for RocksStorage {
         isolation: TransactionIsolationLevel,
     ) -> Result<Self::TransactionType<'_>, DatabaseError> {
         self.validate_transaction_isolation(isolation)?;
+        let tx = if self.config.sync {
+            self.inner
+                .transaction_opt(&sync_write_options(), &TransactionOptions::default())
+        } else {
+            self.inner.transaction()
+        };
         Ok(RocksTransaction {
             db: self.inner.as_ref(),
-            tx: self.inner.transaction(),
+            tx,
             isolation,
             current_snapshot: matches!(isolation, TransactionIsolationLevel::RepeatableRead)
                 .then(|| self.inner.snapshot()),
@@ -629,6 +651,16 @@ macro_rules! impl_transaction {
                 = $iter<'storage, 'iter>
             where
                 Self: 'iter;
+
+            type PointCursor<'a>
+                = ()
+            where
+                Self: 'a;
+
+            fn point_cursor(&self) -> Result<Self::PointCursor<'_>, DatabaseError> {
+                Ok(())
+            }
+
             fn next_statement_stamp(&mut self) -> Result<u64, DatabaseError> {
                 statement_stamp(self.db, &mut self.statements)
             }
@@ -1114,6 +1146,7 @@ mod test {
                 deserializers: std::borrow::Cow::Owned(deserializers),
                 total_len: table.columns_len(),
                 tx: &transaction,
+                cursor: None,
                 cover_mapping: None,
                 with_pk: true,
                 reverse: false,

tpcc/src/backend/dual.rs

@@ -42,7 +42,7 @@ impl PreparedStatement for DualPreparedStatement<'_> {
 impl DualBackend {
     pub fn new(path: &str, rocksdb_stats: bool) -> Result<Self, TpccError> {
         Ok(Self {
-            kitesql: KiteSqlRocksDbBackend::new(path, rocksdb_stats)?,
+            kitesql: KiteSqlRocksDbBackend::new(path, rocksdb_stats, false)?,
             sqlite: SqliteBackend::new_memory()?,
         })
     }

tpcc/src/backend/kitesql_lmdb.rs

@@ -27,11 +27,9 @@ pub struct KiteSqlLmdbBackend {
 }
 
 impl KiteSqlLmdbBackend {
-    pub fn new(path: &str) -> Result<Self, TpccError> {
+    pub fn new(path: &str, sync: bool) -> Result<Self, TpccError> {
         Ok(Self {
-            database: DataBaseBuilder::path(path)
-                .lmdb_no_sync(true)
-                .build_lmdb()?,
+            database: DataBaseBuilder::path(path).sync(sync).build_lmdb()?,
         })
     }
 

tpcc/src/backend/kitesql_rocksdb.rs

@@ -34,20 +34,22 @@ pub struct KiteSqlRocksBackend<S: Storage> {
 }
 
 impl KiteSqlRocksDbBackend {
-    pub fn new(path: &str, rocksdb_stats: bool) -> Result<Self, TpccError> {
+    pub fn new(path: &str, rocksdb_stats: bool, sync: bool) -> Result<Self, TpccError> {
         Ok(Self {
             database: DataBaseBuilder::path(path)
                 .storage_statistics(rocksdb_stats)
+                .sync(sync)
                 .build_rocksdb()?,
         })
     }
 }
 
 impl KiteSqlOptimisticRocksDbBackend {
-    pub fn new(path: &str, rocksdb_stats: bool) -> Result<Self, TpccError> {
+    pub fn new(path: &str, rocksdb_stats: bool, sync: bool) -> Result<Self, TpccError> {
         Ok(Self {
             database: DataBaseBuilder::path(path)
                 .storage_statistics(rocksdb_stats)
+                .sync(sync)
                 .build_optimistic()?,
         })
     }