persistant_relational/
speed.rs

1// SPDX-License-Identifier: MIT OR Apache-2.0
2//! Stage 3 of backend benchmarking for the relational kind: **speed**, as data, with the environment
3//! recorded. The rules and the data types are the kind-agnostic ones in [`persistant_core::bench`]
4//! (only a backend that passes the contract leaf a workload depends on is timed; the rest are listed
5//! as excluded, never ranked; a figure carries its environment and the command that re-runs it).
6//! This module owns the relational workloads and how they are timed: wall-clock, one statement at a
7//! time, from the caller's side of the trait. A network database includes its round trip.
8
9use crate::conformance::{Leaf, Sql, Ty, drop_tables};
10use crate::contract;
11use crate::{RelError, RelationalBackend, RelationalTx, Value};
12use persistant_core::choose::Eligibility;
13use persistant_core::{Config, Reason};
14use std::time::Instant;
15
16pub use persistant_core::bench::{
17    Comparison, Environment, Left, Outcome, Ranked, SpeedParseError as ParseError, SpeedRow,
18    SpeedTable, Stats, compare,
19};
20
21/// One thing worth timing, with the contract leaf it depends on.
22#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
23#[non_exhaustive]
24pub enum Workload {
25    /// `INSERT` of one row to a fresh key, outside a transaction.
26    Insert,
27    /// `SELECT` of one row by primary key, over a table of 100 rows.
28    SelectByKey,
29    /// `UPDATE` of one row by primary key.
30    Update,
31    /// `begin`, one `INSERT`, `commit`.
32    Transaction,
33    /// Inside an open transaction: `SAVEPOINT`, one `INSERT`, `ROLLBACK TO SAVEPOINT` (three statements).
34    Savepoint,
35    /// `INSERT .. ON CONFLICT (id) DO UPDATE .. WHERE`, over 100 rows, always taking the update branch.
36    UpsertWhere,
37}
38
39impl Workload {
40    /// Every workload.
41    pub const ALL: [Workload; 6] = [
42        Workload::Insert,
43        Workload::SelectByKey,
44        Workload::Update,
45        Workload::Transaction,
46        Workload::Savepoint,
47        Workload::UpsertWhere,
48    ];
49
50    /// The id used in data files.
51    pub fn id(self) -> &'static str {
52        match self {
53            Workload::Insert => "speed.rel.insert",
54            Workload::SelectByKey => "speed.rel.select_by_key",
55            Workload::Update => "speed.rel.update",
56            Workload::Transaction => "speed.rel.transaction",
57            Workload::Savepoint => "speed.rel.savepoint",
58            Workload::UpsertWhere => "speed.rel.upsert_where",
59        }
60    }
61
62    /// The contract leaf the workload needs: no pass, no number.
63    pub fn required(self) -> Leaf {
64        match self {
65            Workload::Insert | Workload::SelectByKey | Workload::Update => Leaf::Values,
66            Workload::Transaction => Leaf::Transaction,
67            Workload::Savepoint => Leaf::Savepoint,
68            Workload::UpsertWhere => Leaf::UpsertWhere,
69        }
70    }
71}
72
73/// Time `workloads` on `target`, `iterations` operations each. A workload whose leaf the target does
74/// not pass gets an excluded row and no timing.
75pub async fn run<R: RelationalBackend>(
76    target: &R,
77    config: &Config,
78    env: &Environment,
79    workloads: &[Workload],
80    iterations: u32,
81) -> Vec<SpeedRow> {
82    let mut out = Vec::with_capacity(workloads.len());
83    for w in workloads.iter().copied() {
84        let leaf = w.required();
85        let outcome = match contract::check(target, &[leaf]).await {
86            Eligibility::Eligible => match time(target, w, iterations).await {
87                Ok(Some(s)) => Outcome::Measured(s),
88                Ok(None) => Outcome::Excluded {
89                    leaf: leaf.id().to_string(),
90                    reason: Reason::Untested,
91                    note: "no operations were timed".to_string(),
92                },
93                Err(e) => Outcome::Excluded {
94                    leaf: leaf.id().to_string(),
95                    reason: Reason::Missing,
96                    note: format!("the workload failed on a backend that passed the leaf: {e}"),
97                },
98            },
99            Eligibility::Excluded(ex) => {
100                let e = &ex[0];
101                Outcome::Excluded {
102                    leaf: e.leaf.clone(),
103                    reason: e.reason,
104                    note: e.note.clone(),
105                }
106            }
107        };
108        out.push(SpeedRow {
109            workload: w.id().to_string(),
110            backend: target.backend().to_string(),
111            version: target.version().to_string(),
112            config: config.clone(),
113            env: env.clone(),
114            outcome,
115        });
116    }
117    out
118}
119
120const TABLE: &str = "persistant_speed_rel";
121const SEED_ROWS: i64 = 100;
122
123async fn time<R: RelationalBackend>(
124    b: &R,
125    w: Workload,
126    iterations: u32,
127) -> Result<Option<Stats>, RelError> {
128    let s = Sql(b.dialect());
129    drop_tables(b, &[TABLE]).await?;
130    b.execute(
131        &format!(
132            "CREATE TABLE {TABLE} ({}, n BIGINT NOT NULL, t {} NOT NULL){}",
133            s.pk("id"),
134            s.ty(Ty::Text),
135            s.tail()
136        ),
137        &[],
138    )
139    .await?;
140    let result = timed_loop(b, &s, w, iterations).await;
141    // Best effort: leave a shared target as found, whatever happened.
142    let _ = drop_tables(b, &[TABLE]).await;
143    result
144}
145
146async fn timed_loop<R: RelationalBackend>(
147    b: &R,
148    s: &Sql,
149    w: Workload,
150    iterations: u32,
151) -> Result<Option<Stats>, RelError> {
152    let ins = format!(
153        "INSERT INTO {TABLE} (id, n, t) VALUES ({}, {}, {})",
154        s.p(1),
155        s.p(2),
156        s.p(3)
157    );
158    let sel = format!("SELECT n, t FROM {TABLE} WHERE id = {}", s.p(1));
159    let upd = format!("UPDATE {TABLE} SET n = n + 1 WHERE id = {}", s.p(1));
160    let row = |id: i64| [Value::Int(id), Value::Int(id), Value::Text("x".repeat(64))];
161    // `n = id` for a seeded row, so an upsert of `n = 1000 + i` is always higher and always updates.
162    let ups = format!(
163        "INSERT INTO {TABLE} (id, n, t) VALUES ({}, {}, {}) ON CONFLICT (id) DO UPDATE SET n = EXCLUDED.n WHERE {TABLE}.n < EXCLUDED.n",
164        s.p(1),
165        s.p(2),
166        s.p(3)
167    );
168    if matches!(
169        w,
170        Workload::SelectByKey | Workload::Update | Workload::UpsertWhere
171    ) {
172        for id in 0..SEED_ROWS {
173            b.execute(&ins, &row(id)).await?;
174        }
175    }
176    let mut us: Vec<u64> = Vec::with_capacity(iterations as usize);
177    macro_rules! timed {
178        ($e:expr) => {{
179            let t = Instant::now();
180            let r = $e;
181            us.push(u64::try_from(t.elapsed().as_nanos()).unwrap_or(u64::MAX));
182            r
183        }};
184    }
185    // One transaction for every iteration of the savepoint workload, rolled back at the end.
186    let mut sp_tx = if w == Workload::Savepoint {
187        Some(b.begin(crate::Isolation::ReadCommitted).await?)
188    } else {
189        None
190    };
191    for i in 0..i64::from(iterations) {
192        match w {
193            Workload::Insert => {
194                timed!(b.execute(&ins, &row(i)).await?);
195            }
196            Workload::SelectByKey => {
197                let rows = timed!(b.query(&sel, &[Value::Int(i % SEED_ROWS)]).await?);
198                if rows.len() != 1 {
199                    return Err(RelError::Backend(
200                        "select by key did not return its row".to_string(),
201                    ));
202                }
203            }
204            Workload::Update => {
205                let n = timed!(b.execute(&upd, &[Value::Int(i % SEED_ROWS)]).await?);
206                if n != 1 {
207                    return Err(RelError::Backend(
208                        "update by key did not touch one row".to_string(),
209                    ));
210                }
211            }
212            Workload::Transaction => {
213                let t = Instant::now();
214                let mut tx = b.begin(crate::Isolation::ReadCommitted).await?;
215                tx.execute(&ins, &row(i)).await?;
216                tx.commit().await?;
217                us.push(u64::try_from(t.elapsed().as_nanos()).unwrap_or(u64::MAX));
218            }
219            Workload::Savepoint => {
220                let tx = sp_tx.as_mut().expect("opened above for this workload");
221                let t = Instant::now();
222                tx.execute("SAVEPOINT persistant_speed_sp", &[]).await?;
223                tx.execute(&ins, &row(i)).await?;
224                tx.execute("ROLLBACK TO SAVEPOINT persistant_speed_sp", &[])
225                    .await?;
226                us.push(u64::try_from(t.elapsed().as_nanos()).unwrap_or(u64::MAX));
227            }
228            Workload::UpsertWhere => {
229                let id = i % SEED_ROWS;
230                let r = [
231                    Value::Int(id),
232                    Value::Int(1000 + i),
233                    Value::Text("x".repeat(64)),
234                ];
235                timed!(b.execute(&ups, &r).await?);
236                let got = b
237                    .query(
238                        &format!("SELECT n FROM {TABLE} WHERE id = {}", s.p(1)),
239                        &[Value::Int(id)],
240                    )
241                    .await?;
242                if got.first().and_then(|r| r.int(0)) != Some(1000 + i) {
243                    return Err(RelError::Backend(
244                        "the upsert did not take the update branch".to_string(),
245                    ));
246                }
247            }
248        }
249    }
250    if let Some(tx) = sp_tx {
251        tx.rollback().await?;
252    }
253    Ok(Stats::from_samples(us))
254}
persistant_relational/speed.rs

persistant_relational/
speed.rs
