refactor(tpcc): split backends and improve benchmark tooling

Author: KKould

.gitignore

@@ -25,4 +25,4 @@ kite_sql_tpcc
 copy.csv
 
 tests/data/row_20000.csv
-tests/data/distinct_rows.csv
+tests/data/distinct_rows.csv

Makefile

@@ -3,8 +3,11 @@ CARGO ?= cargo
 WASM_PACK ?= wasm-pack
 SQLLOGIC_PATH ?= tests/slt/**/*.slt
 PYO3_PYTHON ?= /usr/bin/python3.12
+TPCC_MEASURE_TIME ?= 15
+TPCC_NUM_WARE ?= 1
+TPCC_PPROF_OUTPUT ?= /tmp/tpcc_lmdb.svg
 
-.PHONY: test test-python test-wasm test-slt test-all wasm-build check tpcc tpcc-dual cargo-check build wasm-examples native-examples fmt clippy
+.PHONY: test test-python test-wasm test-slt test-all wasm-build check tpcc tpcc-kitesql-rocksdb tpcc-kitesql-lmdb tpcc-lmdb-flamegraph tpcc-sqlite-practical tpcc-dual cargo-check build wasm-examples native-examples fmt clippy
 
 ## Run default Rust tests in the current environment (non-WASM).
 test:
@@ -48,9 +51,23 @@ clippy:
 ## Run formatting (check mode) and clippy linting together.
 check: fmt clippy
 
-## Execute the TPCC workload example as a standalone command.
-tpcc:
-	$(CARGO) run -p tpcc --release
+tpcc: tpcc-kitesql-lmdb
+
+## Execute the TPCC workload on KiteSQL with RocksDB storage.
+tpcc-kitesql-rocksdb:
+	$(CARGO) run -p tpcc --release -- --backend kitesql-rocksdb
+
+## Execute the TPCC workload on KiteSQL with LMDB storage.
+tpcc-kitesql-lmdb:
+	$(CARGO) run -p tpcc --release -- --backend kitesql-lmdb
+
+## Execute TPCC on LMDB and emit a pprof flamegraph SVG.
+tpcc-lmdb-flamegraph:
+	CARGO_PROFILE_RELEASE_DEBUG=true $(CARGO) run -p tpcc --release --features pprof -- --backend kitesql-lmdb --measure-time $(TPCC_MEASURE_TIME) --num-ware $(TPCC_NUM_WARE) --pprof-output $(TPCC_PPROF_OUTPUT)
+
+## Execute the TPCC workload on SQLite with the practical profile.
+tpcc-sqlite-practical:
+	$(CARGO) run -p tpcc --release -- --backend sqlite --sqlite-profile practical --path kite_sql_tpcc.sqlite
 
 ## Execute TPCC while mirroring every statement to an in-memory SQLite instance for validation.
 tpcc-dual:

README.md

@@ -179,17 +179,14 @@ Run `make tpcc-dual` to mirror every TPCC statement to an in-memory SQLite datab
 - KIOXIA-EXCERIA PLUS G3 SSD
 - Tips: TPC-C currently only supports single thread
 
-All cases have been fully optimized.
-```shell
-<90th Percentile RT (MaxRT)>
-   New-Order : 0.002  (0.005)
-     Payment : 0.001  (0.013)
-Order-Status : 0.002  (0.006)
-    Delivery : 0.010  (0.023)
- Stock-Level : 0.002  (0.017)
-<TpmC>
-27226 Tpmc
-```
+Recent 720-second local comparison on the machine above:
+
+| Backend | TpmC | New-Order p90 | Payment p90 | Order-Status p90 | Delivery p90 | Stock-Level p90 |
+| --- | ---: | ---: | ---: | ---: | ---: | ---: |
+| SQLite practical | 35516 | 0.001s | 0.001s | 0.001s | 0.001s | 0.001s |
+| KiteSQL LMDB | 29171 | 0.001s | 0.001s | 0.001s | 0.015s | 0.002s |
+
+The detailed raw outputs for both runs are recorded in [tpcc/README.md](tpcc/README.md).
 #### 👉[check more](tpcc/README.md)
 
 ## Roadmap

src/execution/dql/aggregate/hash_agg.rs

@@ -88,22 +88,6 @@ impl HashAggExecutor {
 
             while arena.next_tuple(self.input)? {
                 let tuple = arena.result_tuple();
-                let mut values = Vec::with_capacity(self.agg_calls.len());
-
-                for expr in &self.agg_calls {
-                    if let ScalarExpression::AggCall { args, .. } = expr {
-                        if args.len() > 1 {
-                            return Err(DatabaseError::UnsupportedStmt(
-                                "currently aggregate functions only support a single Column as a parameter"
-                                    .to_string(),
-                            ));
-                        }
-                        values.push(args[0].eval(Some((tuple, &self.input_schema)))?);
-                    } else {
-                        unreachable!()
-                    }
-                }
-
                 let group_keys = self
                     .groupby_exprs
                     .iter()
@@ -114,8 +98,19 @@ impl HashAggExecutor {
                     Entry::Occupied(entry) => entry.into_mut(),
                     Entry::Vacant(entry) => entry.insert(create_accumulators(&self.agg_calls)?),
                 };
-                for (acc, value) in entry.iter_mut().zip_eq(values.iter()) {
-                    acc.update_value(value)?;
+
+                for (acc, expr) in entry.iter_mut().zip_eq(self.agg_calls.iter()) {
+                    let ScalarExpression::AggCall { args, .. } = expr else {
+                        unreachable!()
+                    };
+                    if args.len() > 1 {
+                        return Err(DatabaseError::UnsupportedStmt(
+                            "currently aggregate functions only support a single Column as a parameter"
+                                .to_string(),
+                        ));
+                    }
+                    let value = args[0].eval(Some((tuple, &self.input_schema)))?;
+                    acc.update_value(&value)?;
                 }
             }
 
@@ -128,12 +123,15 @@ impl HashAggExecutor {
         };
 
         let output = arena.result_tuple_mut();
+
         output.pk = None;
-        output.values = accs
-            .iter()
-            .map(|acc| acc.evaluate())
-            .chain(group_keys.into_iter().map(Ok))
-            .try_collect()?;
+        output.values.clear();
+        output.values.reserve(accs.len() + group_keys.len());
+
+        for acc in accs.iter() {
+            output.values.push(acc.evaluate()?);
+        }
+        output.values.extend(group_keys);
         arena.resume();
         Ok(())
     }

src/execution/dql/aggregate/mod.rs

@@ -63,6 +63,7 @@ fn create_accumulator(expr: &ScalarExpression) -> Result<Box<dyn Accumulator>, D
     }
 }
 
+#[inline]
 pub(crate) fn create_accumulators(
     exprs: &[ScalarExpression],
 ) -> Result<Vec<Box<dyn Accumulator>>, DatabaseError> {

src/execution/dql/aggregate/simple_agg.rs

@@ -20,9 +20,6 @@ use crate::planner::operator::aggregate::AggregateOperator;
 use crate::planner::LogicalPlan;
 use crate::storage::Transaction;
 use crate::types::tuple::SchemaRef;
-use crate::types::value::DataValue;
-use itertools::Itertools;
-
 pub struct SimpleAggExecutor {
     agg_calls: Vec<ScalarExpression>,
     input_schema: SchemaRef,
@@ -78,25 +75,29 @@ impl SimpleAggExecutor {
 
         while arena.next_tuple(input)? {
             let tuple = arena.result_tuple();
-            let values: Vec<DataValue> = self
-                .agg_calls
-                .iter()
-                .map(|expr| match expr {
-                    ScalarExpression::AggCall { args, .. } => {
-                        args[0].eval(Some((tuple, &self.input_schema)))
-                    }
-                    _ => unreachable!(),
-                })
-                .try_collect()?;
+            for (acc, expr) in accs.iter_mut().zip(self.agg_calls.iter()) {
+                let ScalarExpression::AggCall { args, .. } = expr else {
+                    unreachable!()
+                };
+                if args.len() > 1 {
+                    return Err(DatabaseError::UnsupportedStmt(
+                        "currently aggregate functions only support a single Column as a parameter"
+                            .to_string(),
+                    ));
+                }
 
-            for (acc, value) in accs.iter_mut().zip(values.iter()) {
-                acc.update_value(value)?;
+                let value = args[0].eval(Some((tuple, &self.input_schema)))?;
+                acc.update_value(&value)?;
             }
         }
 
         let output = arena.result_tuple_mut();
         output.pk = None;
-        output.values = accs.into_iter().map(|acc| acc.evaluate()).try_collect()?;
+        output.values.clear();
+        output.values.reserve(accs.len());
+        for acc in accs {
+            output.values.push(acc.evaluate()?);
+        }
         arena.resume();
         Ok(())
     }

tpcc/Cargo.toml

@@ -3,6 +3,9 @@ name = "tpcc"
 version = "0.1.1"
 edition = "2021"
 
+[features]
+pprof = ["dep:pprof"]
+
 [dependencies]
 clap            = { version = "4", features = ["derive"] }
 chrono          = { version = "0.4" }
@@ -14,3 +17,6 @@ rust_decimal    = { version = "1" }
 thiserror       = { version = "1" }
 sqlite          = { version = "0.34" }
 sqlparser       = { version = "0.61" }
+
+[target.'cfg(unix)'.dependencies]
+pprof = { version = "0.15", features = ["flamegraph"], optional = true }