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::{Dialect, 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 tail = if matches!(s.0, Dialect::MySql | Dialect::MariaDb) {
163        format!("ON DUPLICATE KEY UPDATE n = IF({TABLE}.n < VALUES(n), VALUES(n), {TABLE}.n)")
164    } else {
165        format!("ON CONFLICT (id) DO UPDATE SET n = EXCLUDED.n WHERE {TABLE}.n < EXCLUDED.n")
166    };
167    let ups = format!(
168        "INSERT INTO {TABLE} (id, n, t) VALUES ({}, {}, {}) {tail}",
169        s.p(1),
170        s.p(2),
171        s.p(3)
172    );
173    if matches!(
174        w,
175        Workload::SelectByKey | Workload::Update | Workload::UpsertWhere
176    ) {
177        for id in 0..SEED_ROWS {
178            b.execute(&ins, &row(id)).await?;
179        }
180    }
181    let mut us: Vec<u64> = Vec::with_capacity(iterations as usize);
182    macro_rules! timed {
183        ($e:expr) => {{
184            let t = Instant::now();
185            let r = $e;
186            us.push(u64::try_from(t.elapsed().as_nanos()).unwrap_or(u64::MAX));
187            r
188        }};
189    }
190    // One transaction for every iteration of the savepoint workload, rolled back at the end.
191    let mut sp_tx = if w == Workload::Savepoint {
192        Some(b.begin(crate::Isolation::ReadCommitted).await?)
193    } else {
194        None
195    };
196    for i in 0..i64::from(iterations) {
197        match w {
198            Workload::Insert => {
199                timed!(b.execute(&ins, &row(i)).await?);
200            }
201            Workload::SelectByKey => {
202                let rows = timed!(b.query(&sel, &[Value::Int(i % SEED_ROWS)]).await?);
203                if rows.len() != 1 {
204                    return Err(RelError::Backend(
205                        "select by key did not return its row".to_string(),
206                    ));
207                }
208            }
209            Workload::Update => {
210                let n = timed!(b.execute(&upd, &[Value::Int(i % SEED_ROWS)]).await?);
211                if n != 1 {
212                    return Err(RelError::Backend(
213                        "update by key did not touch one row".to_string(),
214                    ));
215                }
216            }
217            Workload::Transaction => {
218                let t = Instant::now();
219                let mut tx = b.begin(crate::Isolation::ReadCommitted).await?;
220                tx.execute(&ins, &row(i)).await?;
221                tx.commit().await?;
222                us.push(u64::try_from(t.elapsed().as_nanos()).unwrap_or(u64::MAX));
223            }
224            Workload::Savepoint => {
225                let tx = sp_tx.as_mut().expect("opened above for this workload");
226                let t = Instant::now();
227                tx.execute("SAVEPOINT persistant_speed_sp", &[]).await?;
228                tx.execute(&ins, &row(i)).await?;
229                tx.execute("ROLLBACK TO SAVEPOINT persistant_speed_sp", &[])
230                    .await?;
231                us.push(u64::try_from(t.elapsed().as_nanos()).unwrap_or(u64::MAX));
232            }
233            Workload::UpsertWhere => {
234                let id = i % SEED_ROWS;
235                let r = [
236                    Value::Int(id),
237                    Value::Int(1000 + i),
238                    Value::Text("x".repeat(64)),
239                ];
240                timed!(b.execute(&ups, &r).await?);
241                let got = b
242                    .query(
243                        &format!("SELECT n FROM {TABLE} WHERE id = {}", s.p(1)),
244                        &[Value::Int(id)],
245                    )
246                    .await?;
247                if got.first().and_then(|r| r.int(0)) != Some(1000 + i) {
248                    return Err(RelError::Backend(
249                        "the upsert did not take the update branch".to_string(),
250                    ));
251                }
252            }
253        }
254    }
255    if let Some(tx) = sp_tx {
256        tx.rollback().await?;
257    }
258    Ok(Stats::from_samples(us))
259}
persistant_relational/speed.rs

persistant_relational/
speed.rs
