25672443658

Committed 11 May 2026 01:14PM UTC coverage: 68.857% (-0.2%) from 69.041%

Build # 25672443658

Build Type

push

github

Committed by

web-flow

Commit Message

bench(W13): Phase 11.11b — concurrent-writers workload (SQLR-22 / SQLR-16) (#133)

Adds the headline Phase-11 differentiator workload to the SQLR-16
benchmark harness. Splits the original 11.11 scope: this slice ships
the bench workload only; the Go SDK cross-pool sibling shape is now
Phase 11.11c (a separate slice because it touches the Go binding
architecture, not the bench harness).

Workload shape (`benchmarks/src/workloads/concurrent_writers.rs`):

  CREATE TABLE counters (id INTEGER PRIMARY KEY, n INTEGER NOT NULL);
  -- preload 1_000 rows with n=0
  -- 4 worker threads × 50 BEGIN/UPDATE/COMMIT cycles each
  --   BEGIN <CONCURRENT|IMMEDIATE>;
  --   UPDATE counters SET n = n + 1 WHERE id = ?;  -- random in 1..=1000
  --   COMMIT;                                     -- retry on Busy/Locked

Per-engine specifics flow through four new `Driver` trait methods,
each with a sensible default so existing workloads (W1..W12) and
drivers are untouched:

- `connect_sibling(primary, path)` — SQLRite overrides to call
  `Connection::connect()` so workers share the primary's
  `Arc<Mutex<Database>>` (SQLRite's `Connection::open` takes
  `flock(LOCK_EX)`, so workers can't re-open). Default opens a
  fresh connection at `path` — works for SQLite (separate
  connections coexist under WAL).
- `enable_concurrent_mode(conn)` — SQLRite runs
  `PRAGMA journal_mode = mvcc;`. Default no-op (SQLite's WAL +
  busy_timeout setup happens at `open`).
- `concurrent_begin_sql()` — SQLRite returns `"BEGIN CONCURRENT"`,
  SQLite returns `"BEGIN IMMEDIATE"`. Default `"BEGIN"`.
- `is_retryable_busy(err)` — SQLRite downcasts to
  `SQLRiteError::is_retryable()`; SQLite walks the anyhow chain for
  rusqlite's `DatabaseBusy` / `DatabaseLocked`. Default false.

SQLite driver gains a `busy_timeout = 5s` pragma at open so its
`BEGIN IMMEDIATE` blocks rather than fails on contention. SQLRite
driver's `.map_err(|e| anyhow::anyhow!(...))` pattern in `execute`
/ `execute_with_params` was dro... (continued)

Coverage Stats

0 of 46 new or added lines in 3 files covered. (0.0%)

3 existing lines in 2 files now uncovered.

11148 of 16190 relevant lines covered (68.86%)

1.23 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/benchmarks/src/drivers/sqlrite.rs

//! SQLRite driver.
//!
//! Binds against the engine's public [`sqlrite::Connection`] surface —
//! the same API the language SDKs use.
//!
//! ## SQLR-23 — bound + cached path
//!
//! SQLRite gained a prepared-statement plan cache + parameter binding
//! in SQLR-23. This driver uses both:
//!
//! - `query_one` / `query_all` route through [`sqlrite::Connection::prepare_cached`]
//!   so a hot SELECT pays the sqlparser walk exactly once across the
//!   whole bench loop (cache cap defaults to 16, plenty for any single
//!   workload).
//! - `execute_with_params` does the same for INSERT-loop hot paths.
//! - `Value::Vector` binds directly through `Statement::query_with_params`
//!   without round-tripping through a 4 KB bracket-array SQL literal —
//!   this is the W10/W12 unlock. The HNSW probe optimizer recognizes
//!   the bound vector via the same in-band shape an inline `[…]` would
//!   produce, so the optimizer hook still kicks in on bound queries.
//!
//! That's how a perf-conscious SQLRite user would write hot-path code
//! today.

use std::path::Path;

use anyhow::{Context, Result};

use crate::{Driver, Value};

#[derive(Clone, Copy)]
pub struct SQLRiteDriver;

impl Driver for SQLRiteDriver {
    type Conn = sqlrite::Connection;

    fn name(&self) -> &'static str {
        "sqlrite"
    }

    fn open(&self, path: &Path) -> Result<Self::Conn> {
        sqlrite::Connection::open(path)
            .map_err(|e| anyhow::anyhow!("sqlrite open({}): {e}", path.display()))
    }

    fn execute(&self, conn: &mut Self::Conn, sql: &str) -> Result<()> {
        // Preserve the typed `SQLRiteError` as the anyhow source so
        // [`is_retryable_busy`] can downcast — the W13 retry loop
        // needs to distinguish `Busy` / `BusySnapshot` from other
        // failures. Adding context via `.with_context` keeps the
        // human-readable wrapper while threading the typed source
        // underneath.
        conn.execute(sql)
            .map_err(anyhow::Error::new)
            .with_context(|| format!("sqlrite execute: {sql}"))?;
        Ok(())
    }

    fn execute_with_params(
        &self,
        conn: &mut Self::Conn,
        sql: &str,
        params: &[Value],
    ) -> Result<()> {
        let bound = to_engine_values(params);
        let mut stmt = conn
            .prepare_cached(sql)
            .map_err(anyhow::Error::new)
            .with_context(|| format!("sqlrite prepare_cached: {sql}"))?;
        stmt.execute_with_params(&bound)
            .map_err(anyhow::Error::new)
            .with_context(|| format!("sqlrite execute_with_params: {sql}"))?;
        Ok(())
    }

    fn query_one(&self, conn: &mut Self::Conn, sql: &str, params: &[Value]) -> Result<Vec<Value>> {
        let bound = to_engine_values(params);
        let stmt = conn
            .prepare_cached(sql)
            .map_err(|e| anyhow::anyhow!("sqlrite prepare_cached: {e}\n  sql: {sql}"))?;
        let mut rows = stmt
            .query_with_params(&bound)
            .map_err(|e| anyhow::anyhow!("sqlrite query_with_params: {e}\n  sql: {sql}"))?;
        let row = rows
            .next()
            .map_err(|e| anyhow::anyhow!("sqlrite row read: {e}"))?
            .context("sqlrite query_one: zero rows returned")?;
        let cols = row.columns().len();
        let mut out = Vec::with_capacity(cols);
        for i in 0..cols {
            let v: sqlrite::Value = row.get(i)?;
            out.push(from_engine_value(v));
        }
        if rows
            .next()
            .map_err(|e| anyhow::anyhow!("sqlrite row read: {e}"))?
            .is_some()
        {
            anyhow::bail!("sqlrite query_one: >1 rows returned");
        }
        Ok(out)
    }

    fn query_all(
        &self,
        conn: &mut Self::Conn,
        sql: &str,
        params: &[Value],
    ) -> Result<Vec<Vec<Value>>> {
        let bound = to_engine_values(params);
        let stmt = conn
            .prepare_cached(sql)
            .map_err(|e| anyhow::anyhow!("sqlrite prepare_cached: {e}\n  sql: {sql}"))?;
        let mut rows = stmt
            .query_with_params(&bound)
            .map_err(|e| anyhow::anyhow!("sqlrite query_with_params: {e}\n  sql: {sql}"))?;
        let mut out = Vec::new();
        while let Some(row) = rows
            .next()
            .map_err(|e| anyhow::anyhow!("sqlrite row read: {e}"))?
        {
            let cols = row.columns().len();
            let mut buf = Vec::with_capacity(cols);
            for i in 0..cols {
                let v: sqlrite::Value = row.get(i)?;
                buf.push(from_engine_value(v));
            }
            out.push(buf);
        }
        Ok(out)
    }

    /// Mint a sibling Connection that shares the primary's
    /// `Arc<Mutex<Database>>`. A fresh `Connection::open(path)`
    /// would fail here because the primary already holds an
    /// exclusive `flock(LOCK_EX)` on the WAL sidecar.
    fn connect_sibling(&self, primary: &Self::Conn, _path: &Path) -> Result<Self::Conn> {
        Ok(primary.connect())
    }

    fn enable_concurrent_mode(&self, conn: &mut Self::Conn) -> Result<()> {
        // `BEGIN CONCURRENT` requires `journal_mode = mvcc;`
        // otherwise the engine surfaces a typed error. The PRAGMA
        // is per-database (not per-connection), so toggling once
        // on the primary suffices for every sibling.
        conn.execute("PRAGMA journal_mode = mvcc")
            .map_err(anyhow::Error::new)
            .context("PRAGMA journal_mode = mvcc")?;
        Ok(())
    }

    fn concurrent_begin_sql(&self) -> &'static str {
        "BEGIN CONCURRENT"
    }

    /// SQLRite signals both `Busy` (write-write conflict at commit)
    /// and `BusySnapshot` (snapshot GC'd under a long-lived reader)
    /// via `SQLRiteError::is_retryable()`. The bench harness wraps
    /// engine errors in `anyhow::Error`, so we peel back to the
    /// typed source and consult the predicate.
    fn is_retryable_busy(&self, err: &anyhow::Error) -> bool {
        err.downcast_ref::<sqlrite::SQLRiteError>()
            .map(|e| e.is_retryable())
            .unwrap_or(false)
            || err
                .chain()
                .filter_map(|e| e.downcast_ref::<sqlrite::SQLRiteError>())
                .any(|e| e.is_retryable())
    }
}

/// Map the bench harness's `Value` to SQLRite's engine `Value`. Both
/// enums carry the same logical shapes; this is just a name-mapping.
fn to_engine_values(params: &[Value]) -> Vec<sqlrite::Value> {
    params.iter().map(to_engine_value).collect()
}

fn to_engine_value(v: &Value) -> sqlrite::Value {
    match v {
        Value::Null => sqlrite::Value::Null,
        Value::Integer(i) => sqlrite::Value::Integer(*i),
        Value::Real(f) => sqlrite::Value::Real(*f),
        Value::Text(s) => sqlrite::Value::Text(s.clone()),
        Value::Vector(v) => sqlrite::Value::Vector(v.clone()),
    }
}

fn from_engine_value(v: sqlrite::Value) -> Value {
    match v {
        sqlrite::Value::Null => Value::Null,
        sqlrite::Value::Integer(i) => Value::Integer(i),
        sqlrite::Value::Real(f) => Value::Real(f),
        sqlrite::Value::Text(s) => Value::Text(s),
        sqlrite::Value::Vector(v) => Value::Vector(v),
        // Bool / JSON aren't yet a bench `Value` variant — workloads
        // don't surface them. If a future workload reads one back,
        // grow this match alongside the harness `Value` enum.
        other => Value::Text(format!("{other:?}")),
    }
}

1	//! SQLRite driver.
2	//!
3	//! Binds against the engine's public [`sqlrite::Connection`] surface —
4	//! the same API the language SDKs use.
5	//!
6	//! ## SQLR-23 — bound + cached path
7	//!
8	//! SQLRite gained a prepared-statement plan cache + parameter binding
9	//! in SQLR-23. This driver uses both:
10	//!
11	//! - `query_one` / `query_all` route through [`sqlrite::Connection::prepare_cached`]
12	//! so a hot SELECT pays the sqlparser walk exactly once across the
13	//! whole bench loop (cache cap defaults to 16, plenty for any single
14	//! workload).
15	//! - `execute_with_params` does the same for INSERT-loop hot paths.
16	//! - `Value::Vector` binds directly through `Statement::query_with_params`
17	//! without round-tripping through a 4 KB bracket-array SQL literal —
18	//! this is the W10/W12 unlock. The HNSW probe optimizer recognizes
19	//! the bound vector via the same in-band shape an inline `[…]` would
20	//! produce, so the optimizer hook still kicks in on bound queries.
21	//!
22	//! That's how a perf-conscious SQLRite user would write hot-path code
23	//! today.
24
25	use std::path::Path;
26
27	use anyhow::{Context, Result};
28
29	use crate::{Driver, Value};
30
31	#[derive(Clone, Copy)]
32	pub struct SQLRiteDriver;
33
34	impl Driver for SQLRiteDriver {
35	type Conn = sqlrite::Connection;
36
37	fn name(&self) -> &'static str {	×
38	"sqlrite"	×
39	}
40
41	fn open(&self, path: &Path) -> Result<Self::Conn> {	×
42	sqlrite::Connection::open(path)	×
43	.map_err(\|e\| anyhow::anyhow!("sqlrite open({}): {e}", path.display()))	×
44	}
45
46	fn execute(&self, conn: &mut Self::Conn, sql: &str) -> Result<()> {	×
47	// Preserve the typed `SQLRiteError` as the anyhow source so
48	// [`is_retryable_busy`] can downcast — the W13 retry loop
49	// needs to distinguish `Busy` / `BusySnapshot` from other
50	// failures. Adding context via `.with_context` keeps the
51	// human-readable wrapper while threading the typed source
52	// underneath.
53	conn.execute(sql)	×
NEW 54	.map_err(anyhow::Error::new)	×
NEW 55	.with_context(\|\| format!("sqlrite execute: {sql}"))?;	×
UNCOV 56	Ok(())	×
57	}
58
59	fn execute_with_params(	×
60	&self,
61	conn: &mut Self::Conn,
62	sql: &str,
63	params: &[Value],
64	) -> Result<()> {
65	let bound = to_engine_values(params);	×
66	let mut stmt = conn	×
67	.prepare_cached(sql)	×
NEW 68	.map_err(anyhow::Error::new)	×
NEW 69	.with_context(\|\| format!("sqlrite prepare_cached: {sql}"))?;	×
70	stmt.execute_with_params(&bound)	×
NEW 71	.map_err(anyhow::Error::new)	×
NEW 72	.with_context(\|\| format!("sqlrite execute_with_params: {sql}"))?;	×
UNCOV 73	Ok(())	×
74	}
75
76	fn query_one(&self, conn: &mut Self::Conn, sql: &str, params: &[Value]) -> Result<Vec<Value>> {	×
77	let bound = to_engine_values(params);	×
78	let stmt = conn	×
79	.prepare_cached(sql)	×
80	.map_err(\|e\| anyhow::anyhow!("sqlrite prepare_cached: {e}\n sql: {sql}"))?;	×
81	let mut rows = stmt	×
82	.query_with_params(&bound)	×
83	.map_err(\|e\| anyhow::anyhow!("sqlrite query_with_params: {e}\n sql: {sql}"))?;	×
84	let row = rows	×
85	.next()	×
86	.map_err(\|e\| anyhow::anyhow!("sqlrite row read: {e}"))?	×
87	.context("sqlrite query_one: zero rows returned")?;	×
88	let cols = row.columns().len();	×
89	let mut out = Vec::with_capacity(cols);	×
90	for i in 0..cols {	×
91	let v: sqlrite::Value = row.get(i)?;	×
92	out.push(from_engine_value(v));	×
93	}
94	if rows	×
95	.next()	×
96	.map_err(\|e\| anyhow::anyhow!("sqlrite row read: {e}"))?	×
97	.is_some()
98	{
99	anyhow::bail!("sqlrite query_one: >1 rows returned");	×
100	}
101	Ok(out)	×
102	}
103
104	fn query_all(	×
105	&self,
106	conn: &mut Self::Conn,
107	sql: &str,
108	params: &[Value],
109	) -> Result<Vec<Vec<Value>>> {
110	let bound = to_engine_values(params);	×
111	let stmt = conn	×
112	.prepare_cached(sql)	×
113	.map_err(\|e\| anyhow::anyhow!("sqlrite prepare_cached: {e}\n sql: {sql}"))?;	×
114	let mut rows = stmt	×
115	.query_with_params(&bound)	×
116	.map_err(\|e\| anyhow::anyhow!("sqlrite query_with_params: {e}\n sql: {sql}"))?;	×
117	let mut out = Vec::new();	×
118	while let Some(row) = rows	×
119	.next()	×
120	.map_err(\|e\| anyhow::anyhow!("sqlrite row read: {e}"))?	×
121	{
122	let cols = row.columns().len();	×
123	let mut buf = Vec::with_capacity(cols);	×
124	for i in 0..cols {	×
125	let v: sqlrite::Value = row.get(i)?;	×
126	buf.push(from_engine_value(v));	×
127	}
128	out.push(buf);	×
129	}
130	Ok(out)	×
131	}
132
133	/// Mint a sibling Connection that shares the primary's
134	/// `Arc<Mutex<Database>>`. A fresh `Connection::open(path)`
135	/// would fail here because the primary already holds an
136	/// exclusive `flock(LOCK_EX)` on the WAL sidecar.
NEW 137	fn connect_sibling(&self, primary: &Self::Conn, _path: &Path) -> Result<Self::Conn> {	×
NEW 138	Ok(primary.connect())	×
139	}
140
NEW 141	fn enable_concurrent_mode(&self, conn: &mut Self::Conn) -> Result<()> {	×
142	// `BEGIN CONCURRENT` requires `journal_mode = mvcc;`
143	// otherwise the engine surfaces a typed error. The PRAGMA
144	// is per-database (not per-connection), so toggling once
145	// on the primary suffices for every sibling.
NEW 146	conn.execute("PRAGMA journal_mode = mvcc")	×
NEW 147	.map_err(anyhow::Error::new)	×
NEW 148	.context("PRAGMA journal_mode = mvcc")?;	×
NEW 149	Ok(())	×
150	}
151
NEW 152	fn concurrent_begin_sql(&self) -> &'static str {	×
NEW 153	"BEGIN CONCURRENT"	×
154	}
155
156	/// SQLRite signals both `Busy` (write-write conflict at commit)
157	/// and `BusySnapshot` (snapshot GC'd under a long-lived reader)
158	/// via `SQLRiteError::is_retryable()`. The bench harness wraps
159	/// engine errors in `anyhow::Error`, so we peel back to the
160	/// typed source and consult the predicate.
NEW 161	fn is_retryable_busy(&self, err: &anyhow::Error) -> bool {	×
NEW 162	err.downcast_ref::<sqlrite::SQLRiteError>()	×
NEW 163	.map(\|e\| e.is_retryable())	×
NEW 164	.unwrap_or(false)	×
NEW 165	\|\| err	×
NEW 166	.chain()	×
NEW 167	.filter_map(\|e\| e.downcast_ref::<sqlrite::SQLRiteError>())	×
NEW 168	.any(\|e\| e.is_retryable())	×
169	}
170	}
171
172	/// Map the bench harness's `Value` to SQLRite's engine `Value`. Both
173	/// enums carry the same logical shapes; this is just a name-mapping.
174	fn to_engine_values(params: &[Value]) -> Vec<sqlrite::Value> {
175	params.iter().map(to_engine_value).collect()
176	}
177
178	fn to_engine_value(v: &Value) -> sqlrite::Value {
179	match v {
180	Value::Null => sqlrite::Value::Null,
181	Value::Integer(i) => sqlrite::Value::Integer(*i),
182	Value::Real(f) => sqlrite::Value::Real(*f),
183	Value::Text(s) => sqlrite::Value::Text(s.clone()),
184	Value::Vector(v) => sqlrite::Value::Vector(v.clone()),
185	}
186	}
187
188	fn from_engine_value(v: sqlrite::Value) -> Value {
189	match v {
190	sqlrite::Value::Null => Value::Null,
191	sqlrite::Value::Integer(i) => Value::Integer(i),
192	sqlrite::Value::Real(f) => Value::Real(f),
193	sqlrite::Value::Text(s) => Value::Text(s),
194	sqlrite::Value::Vector(v) => Value::Vector(v),
195	// Bool / JSON aren't yet a bench `Value` variant — workloads
196	// don't surface them. If a future workload reads one back,
197	// grow this match alongside the harness `Value` enum.
198	other => Value::Text(format!("{other:?}")),
199	}
200	}

joaoh82 / rust_sqlite / 25672443658

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous