diff --git a/Makefile b/Makefile index ed1d71fa..6b763ded 100644 --- a/Makefile +++ b/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: diff --git a/scripts/run_tpcc_stable.py b/scripts/run_tpcc_stable.py index d5b24431..8972f378 100755 --- a/scripts/run_tpcc_stable.py +++ b/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 diff --git a/src/db.rs b/src/db.rs index 74cf4775..f569ece0 100644 --- a/src/db.rs +++ b/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 { diff --git a/src/storage/lmdb.rs b/src/storage/lmdb.rs index 6cfa872c..afee651a 100644 --- a/src/storage/lmdb.rs +++ b/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 { const STATEMENT_BITS: u32 = 20; self.statements += 1; @@ -315,6 +320,22 @@ impl Transaction for LmdbTransaction<'_> { } } + fn point_cursor(&self) -> Result, DatabaseError> { + Ok(self.tx.open_ro_cursor(self.db)?) + } + + fn get_at<'a>( + &'a self, + cursor: &Self::PointCursor<'a>, + key: &[u8], + ) -> Result>, 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())?; diff --git a/src/storage/memory.rs b/src/storage/memory.rs index ede8f0e0..f5add38a 100644 --- a/src/storage/memory.rs +++ b/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, DatabaseError> { + Ok(()) + } + fn next_statement_stamp(&mut self) -> Result { let stamp = self.statements.get() + 1; self.statements.set(stamp); diff --git a/src/storage/mod.rs b/src/storage/mod.rs index 5b756bc0..65575840 100644 --- a/src/storage/mod.rs +++ b/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 { 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>, DatabaseError>; + fn point_cursor(&self) -> Result, DatabaseError>; + + fn get_at<'a>( + &'a self, + _cursor: &Self::PointCursor<'a>, + key: &[u8], + ) -> Result>, DatabaseError> { + self.get_borrowed(key) + } + fn exists(&self, key: &[u8]) -> Result { 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>, cover_mapping: Option, with_pk: bool, reverse: bool, @@ -1578,7 +1594,11 @@ impl IndexImpl for PrimaryKeyIndexImpl { value, None, |key, _| { - let Some(bytes) = params.tx.get_borrowed(key)? else { + let bytes = match ¶ms.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 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, diff --git a/src/storage/rocksdb.rs b/src/storage/rocksdb.rs index 7be2dba1..e9f99a66 100644 --- a/src/storage/rocksdb.rs +++ b/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, 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, 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, DatabaseError> { + Ok(()) + } + fn next_statement_stamp(&mut self) -> Result { 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, diff --git a/tpcc/src/backend/dual.rs b/tpcc/src/backend/dual.rs index 782dcc1b..959357c3 100644 --- a/tpcc/src/backend/dual.rs +++ b/tpcc/src/backend/dual.rs @@ -42,7 +42,7 @@ impl PreparedStatement for DualPreparedStatement<'_> { impl DualBackend { pub fn new(path: &str, rocksdb_stats: bool) -> Result { Ok(Self { - kitesql: KiteSqlRocksDbBackend::new(path, rocksdb_stats)?, + kitesql: KiteSqlRocksDbBackend::new(path, rocksdb_stats, false)?, sqlite: SqliteBackend::new_memory()?, }) } diff --git a/tpcc/src/backend/kitesql_lmdb.rs b/tpcc/src/backend/kitesql_lmdb.rs index 47cea981..712ff077 100644 --- a/tpcc/src/backend/kitesql_lmdb.rs +++ b/tpcc/src/backend/kitesql_lmdb.rs @@ -27,11 +27,9 @@ pub struct KiteSqlLmdbBackend { } impl KiteSqlLmdbBackend { - pub fn new(path: &str) -> Result { + pub fn new(path: &str, sync: bool) -> Result { Ok(Self { - database: DataBaseBuilder::path(path) - .lmdb_no_sync(true) - .build_lmdb()?, + database: DataBaseBuilder::path(path).sync(sync).build_lmdb()?, }) } diff --git a/tpcc/src/backend/kitesql_rocksdb.rs b/tpcc/src/backend/kitesql_rocksdb.rs index 9e9434a6..ba542444 100644 --- a/tpcc/src/backend/kitesql_rocksdb.rs +++ b/tpcc/src/backend/kitesql_rocksdb.rs @@ -34,20 +34,22 @@ pub struct KiteSqlRocksBackend { } impl KiteSqlRocksDbBackend { - pub fn new(path: &str, rocksdb_stats: bool) -> Result { + pub fn new(path: &str, rocksdb_stats: bool, sync: bool) -> Result { 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 { + pub fn new(path: &str, rocksdb_stats: bool, sync: bool) -> Result { Ok(Self { database: DataBaseBuilder::path(path) .storage_statistics(rocksdb_stats) + .sync(sync) .build_optimistic()?, }) } diff --git a/tpcc/src/backend/sqlite.rs b/tpcc/src/backend/sqlite.rs index 5e4dd25a..c1d704e2 100644 --- a/tpcc/src/backend/sqlite.rs +++ b/tpcc/src/backend/sqlite.rs @@ -17,7 +17,6 @@ use super::{ }; use crate::TpccError; use chrono::{NaiveDateTime, TimeZone, Utc}; -use clap::ValueEnum; use kite_sql::types::tuple::Tuple; use kite_sql::types::value::{DataValue, Utf8Type}; use kite_sql::types::{CharLengthUnits, LogicalType}; @@ -28,22 +27,16 @@ pub struct SqliteBackend { connection: Connection, } -#[derive(Copy, Clone, Debug, Eq, PartialEq, ValueEnum)] -pub enum SqliteProfile { - Balanced, - Practical, -} - impl SqliteBackend { - pub fn new(path: &str, profile: SqliteProfile) -> Result { + pub fn new(path: &str, sync: bool) -> Result { let connection = Connection::open(path)?; - configure_sqlite(&connection, profile)?; + configure_sqlite(&connection, sync)?; Ok(Self { connection }) } pub fn new_memory() -> Result { let connection = Connection::open(":memory:")?; - configure_sqlite(&connection, SqliteProfile::Balanced)?; + configure_sqlite(&connection, false)?; Ok(Self { connection }) } @@ -274,27 +267,21 @@ fn format_time64(value: i64, precision: u64) -> Result { .to_string()) } -fn configure_sqlite(connection: &Connection, profile: SqliteProfile) -> Result<(), TpccError> { - let pragmas: &[&str] = match profile { - SqliteProfile::Balanced => &[ - "PRAGMA journal_mode = WAL;", - "PRAGMA synchronous = NORMAL;", - "PRAGMA temp_store = FILE;", - "PRAGMA foreign_keys = OFF;", - "PRAGMA busy_timeout = 5000;", - ], - SqliteProfile::Practical => &[ - "PRAGMA journal_mode = WAL;", - "PRAGMA synchronous = NORMAL;", - "PRAGMA temp_store = MEMORY;", - "PRAGMA cache_size = -32768;", - "PRAGMA mmap_size = 67108864;", - "PRAGMA foreign_keys = OFF;", - "PRAGMA busy_timeout = 5000;", - ], +fn configure_sqlite(connection: &Connection, sync: bool) -> Result<(), TpccError> { + let synchronous = if sync { + "PRAGMA synchronous = FULL;" + } else { + "PRAGMA synchronous = OFF;" }; - - for pragma in pragmas { + for pragma in [ + "PRAGMA journal_mode = WAL;", + synchronous, + "PRAGMA temp_store = MEMORY;", + "PRAGMA cache_size = -32768;", + "PRAGMA mmap_size = 67108864;", + "PRAGMA foreign_keys = OFF;", + "PRAGMA busy_timeout = 5000;", + ] { connection.execute(pragma)?; } Ok(()) @@ -310,6 +297,9 @@ fn normalize_sqlite_sql(sql: &str) -> Option { } return Some(format!("ANALYZE {table};")); } + if lower.starts_with("create table") { + return Some(format!("{} WITHOUT ROWID;", trimmed.trim_end_matches(';'))); + } Some(trimmed.to_string()) } diff --git a/tpcc/src/main.rs b/tpcc/src/main.rs index a8385514..9d803a74 100644 --- a/tpcc/src/main.rs +++ b/tpcc/src/main.rs @@ -15,7 +15,7 @@ use crate::backend::dual::DualBackend; use crate::backend::kitesql_lmdb::KiteSqlLmdbBackend; use crate::backend::kitesql_rocksdb::{KiteSqlOptimisticRocksDbBackend, KiteSqlRocksDbBackend}; -use crate::backend::sqlite::{SqliteBackend, SqliteProfile}; +use crate::backend::sqlite::SqliteBackend; use crate::backend::{BackendControl, BackendTransaction, StatementSpec}; use crate::delivery::DeliveryTest; use crate::load::Load; @@ -110,8 +110,8 @@ struct Args { num_ware: usize, #[clap(long, default_value = "false")] rocksdb_stats: bool, - #[clap(long, value_enum, default_value = "balanced")] - sqlite_profile: SqliteProfile, + #[clap(long, default_value = "false")] + sync: bool, #[cfg(feature = "pprof")] #[clap(long)] pprof_output: Option, @@ -137,22 +137,24 @@ fn main() -> Result<(), TpccError> { match args.backend { BackendKind::KitesqlRocksdb => { reset_db_path(Path::new(&args.path))?; - let mut backend = KiteSqlRocksDbBackend::new(&args.path, args.rocksdb_stats)?; + let mut backend = + KiteSqlRocksDbBackend::new(&args.path, args.rocksdb_stats, args.sync)?; run_tpcc(&mut backend, &args, &mut rng) } BackendKind::KitesqlOptimisticRocksdb => { reset_db_path(Path::new(&args.path))?; - let mut backend = KiteSqlOptimisticRocksDbBackend::new(&args.path, args.rocksdb_stats)?; + let mut backend = + KiteSqlOptimisticRocksDbBackend::new(&args.path, args.rocksdb_stats, args.sync)?; run_tpcc(&mut backend, &args, &mut rng) } BackendKind::KitesqlLmdb => { reset_db_path(Path::new(&args.path))?; - let mut backend = KiteSqlLmdbBackend::new(&args.path)?; + let mut backend = KiteSqlLmdbBackend::new(&args.path, args.sync)?; run_tpcc(&mut backend, &args, &mut rng) } BackendKind::Sqlite => { reset_db_path(Path::new(&args.path))?; - let mut backend = SqliteBackend::new(&args.path, args.sqlite_profile)?; + let mut backend = SqliteBackend::new(&args.path, args.sync)?; run_tpcc(&mut backend, &args, &mut rng) } BackendKind::Dual => {