1use std::borrow::Cow;
4use std::path::{Path, PathBuf};
5use std::sync::Mutex;
6use std::time::{Duration, Instant};
7
8use rusqlite::types::Value as SqlValue;
9use rusqlite::{Connection, OpenFlags, OptionalExtension};
10
11use crate::error::{Db, Error, Result};
12use crate::slices::{check_entry, cmp_keys, is_hash, is_instant, version_slice, Constraint, Kind, Pattern};
13use crate::sqlite_sql::L;
14use crate::{Binding, Namespace, SliceMember, SliceValue, Store, Version};
15use datumesh_core::json::{self, Value};
16use datumesh_core::lexical::{check_dns_name, check_function_name, check_hex, check_instant};
17use datumesh_core::{
18 keys_from_response, sha256_hex, verify, Arg, Args, Cell, ErrorCode, FunctionDescriptor, Keys, Manifest, Receipt,
19 Resolver, Scalar, Slice, Type,
20};
21
22#[derive(Clone, Debug, PartialEq, Eq)]
26pub struct SqliteOptions {
27 max_size: usize,
28 busy_timeout: Duration,
29 durable: bool,
30}
31
32impl Default for SqliteOptions {
33 fn default() -> SqliteOptions {
34 SqliteOptions::new()
35 }
36}
37
38impl SqliteOptions {
39 pub fn new() -> SqliteOptions {
43 SqliteOptions { max_size: 8, busy_timeout: Duration::from_secs(30), durable: true }
44 }
45
46 pub fn max_size(mut self, n: usize) -> SqliteOptions {
50 self.max_size = n.max(1);
51 self
52 }
53
54 pub fn busy_timeout(mut self, d: Duration) -> SqliteOptions {
59 self.busy_timeout = d;
60 self
61 }
62
63 pub fn durable(mut self, yes: bool) -> SqliteOptions {
69 self.durable = yes;
70 self
71 }
72}
73
74pub struct SqliteStore {
92 path: PathBuf,
93 options: SqliteOptions,
94 idle: Mutex<Vec<Connection>>,
95 writer: Mutex<()>,
99 existing: bool,
101}
102
103impl std::fmt::Debug for SqliteStore {
104 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
105 f.debug_struct("SqliteStore").field("path", &self.path).finish_non_exhaustive()
106 }
107}
108
109struct Pooled<'a> {
111 store: &'a SqliteStore,
112 conn: Option<Connection>,
113}
114
115impl std::ops::Deref for Pooled<'_> {
116 type Target = Connection;
117 fn deref(&self) -> &Connection {
118 self.conn.as_ref().expect("a pooled connection")
119 }
120}
121
122impl Drop for Pooled<'_> {
123 fn drop(&mut self) {
124 if let Some(c) = self.conn.take() {
125 if c.is_autocommit() {
127 if let Ok(mut idle) = self.store.idle.lock() {
128 if idle.len() < self.store.options.max_size {
129 idle.push(c);
130 }
131 }
132 }
133 }
134 }
135}
136
137impl SqliteStore {
138 pub fn open(path: impl AsRef<Path>) -> Result<SqliteStore> {
143 SqliteStore::open_with(path, SqliteOptions::new())
144 }
145
146 pub fn open_with(path: impl AsRef<Path>, options: SqliteOptions) -> Result<SqliteStore> {
152 let store = SqliteStore {
153 path: path.as_ref().to_path_buf(),
154 options,
155 idle: Mutex::new(Vec::new()),
156 writer: Mutex::new(()),
157 existing: false,
158 };
159 let c = store.open_connection()?;
160 if let Ok(mut idle) = store.idle.lock() {
161 idle.push(c);
162 }
163 Ok(store)
164 }
165
166 pub fn open_existing(path: impl AsRef<Path>) -> Result<SqliteStore> {
173 let path = path.as_ref();
174 if !path.is_file() {
175 return Err(Error::Database(crate::error::DatabaseError::no_such_file(path)));
176 }
177 let store = SqliteStore {
178 path: path.to_path_buf(),
179 options: SqliteOptions::new(),
180 idle: Mutex::new(Vec::new()),
181 writer: Mutex::new(()),
182 existing: true,
183 };
184 let c = store.open_connection()?;
185 if let Ok(mut idle) = store.idle.lock() {
186 idle.push(c);
187 }
188 Ok(store)
189 }
190
191 pub fn connect(url: &str) -> Result<SqliteStore> {
195 SqliteStore::open(sqlite_path(url))
196 }
197
198 pub fn path(&self) -> &Path {
200 &self.path
201 }
202
203 fn open_connection(&self) -> Result<Connection> {
204 let mut flags = OpenFlags::SQLITE_OPEN_READ_WRITE | OpenFlags::SQLITE_OPEN_NO_MUTEX;
205 if !self.existing {
206 flags |= OpenFlags::SQLITE_OPEN_CREATE;
207 }
208 let c = Connection::open_with_flags(&self.path, flags).db()?;
209 c.busy_timeout(self.options.busy_timeout).db()?;
210 c.set_prepared_statement_cache_capacity(L::ALL.len() + 16);
211 if self.existing {
212 c.execute_batch("PRAGMA foreign_keys = ON").db()?;
214 return Ok(c);
215 }
216 let start = Instant::now();
220 loop {
221 match c.query_row("PRAGMA journal_mode = WAL", [], |r| r.get::<_, String>(0)) {
222 Ok(mode) if mode.eq_ignore_ascii_case("wal") => break,
223 Ok(mode) => return Err(Error::Config(format!("SQLite: journal mode WAL refused (it is {mode})"))),
224 Err(e) if is_busy(&e) && start.elapsed() < self.options.busy_timeout => {
225 std::thread::sleep(Duration::from_millis(5));
226 }
227 Err(e) => return Err(e).db(),
228 }
229 }
230 let sync = if self.options.durable { "FULL" } else { "NORMAL" };
231 c.execute_batch(&format!("PRAGMA foreign_keys = ON; PRAGMA synchronous = {sync};")).db()?;
232 Ok(c)
233 }
234
235 fn conn(&self) -> Result<Pooled<'_>> {
237 let idle = self.idle.lock().ok().and_then(|mut v| v.pop());
238 let conn = match idle {
239 Some(c) => c,
240 None => self.open_connection()?,
241 };
242 Ok(Pooled { store: self, conn: Some(conn) })
243 }
244
245 pub fn check_statements(&self) -> Result<usize> {
249 let c = self.open_connection()?;
250 for q in L::ALL {
251 c.prepare(q.sql()).db()?;
252 }
253 Ok(L::ALL.len())
254 }
255
256 pub(crate) fn raw(&self) -> Result<impl std::ops::Deref<Target = Connection> + '_> {
258 self.conn()
259 }
260
261 fn resolve<T>(&self, op: &'static str, f: impl FnOnce(&Connection) -> Result<Option<T>>) -> Option<T> {
263 self.run(op, f).ok().flatten()
264 }
265
266 fn run<T>(&self, op: &'static str, f: impl FnOnce(&Connection) -> Result<T>) -> Result<T> {
268 let result = self.conn().and_then(|c| f(&c));
269 if let Err(e @ Error::Database(_)) = &result {
270 tracing::error!(target: "datumesh_store", op, error = %e, "database failure");
271 }
272 result
273 }
274
275 fn write<T>(&self, op: &'static str, f: impl FnOnce(&Connection) -> Result<T>) -> Result<T> {
278 self.run(op, |c| {
279 let v = self.locked_write(c, f)?;
280 self.bound_wal(c);
281 Ok(v)
282 })
283 }
284
285 fn locked_write<T>(&self, c: &Connection, f: impl FnOnce(&Connection) -> Result<T>) -> Result<T> {
287 let _queue = self.writer.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
288 write(c, f)
289 }
290
291 fn bound_wal(&self, c: &Connection) {
299 let mut wal = self.path.clone().into_os_string();
300 wal.push("-wal");
301 if std::fs::metadata(&wal).is_ok_and(|m| m.len() > WAL_LIMIT) {
302 if let Err(e) = c.query_row("PRAGMA wal_checkpoint(TRUNCATE)", [], |_| Ok(())) {
303 tracing::debug!(target: "datumesh_store", error = %e, "checkpoint deferred");
304 }
305 }
306 }
307
308 fn asof_version(&self, op: &'static str, q: L, cell_id: &str, asof: &str) -> Result<Option<Version>> {
310 check_hex(cell_id, 64, "cell_id")?;
311 check_instant(asof, "asof")?;
312 self.run(op, |c| {
313 c.prepare_cached(q.sql())
314 .db()?
315 .query_row(rusqlite::params![cell_id, asof], |r| version_row(r, cell_id))
316 .optional()
317 .db()
318 })
319 }
320
321 fn asof_with_receipt(
324 &self,
325 op: &'static str,
326 q: L,
327 cell_id: &str,
328 asof: &str,
329 ) -> Result<Option<(Version, Vec<u8>)>> {
330 check_hex(cell_id, 64, "cell_id")?;
331 check_instant(asof, "asof")?;
332 self.run(op, |c| {
333 c.prepare_cached(q.sql())
334 .db()?
335 .query_row(rusqlite::params![cell_id, asof], |r| Ok((version_row(r, cell_id)?, r.get(3)?)))
336 .optional()
337 .db()
338 })
339 }
340}
341
342const WAL_LIMIT: u64 = 64 << 20;
344
345pub(crate) fn write<T>(c: &Connection, f: impl FnOnce(&Connection) -> Result<T>) -> Result<T> {
347 c.execute_batch("BEGIN IMMEDIATE").db()?;
348 let result = f(c).and_then(|v| c.execute_batch("COMMIT").db().map(|()| v));
349 if result.is_err() && !c.is_autocommit() {
350 let _ = c.execute_batch("ROLLBACK");
351 }
352 result
353}
354
355fn is_busy(e: &rusqlite::Error) -> bool {
356 matches!(e, rusqlite::Error::SqliteFailure(f, _) if f.code == rusqlite::ErrorCode::DatabaseBusy)
357}
358
359pub(crate) fn sqlite_path(url: &str) -> &str {
361 match url.strip_prefix("sqlite:") {
362 Some(rest) => rest.strip_prefix("//").unwrap_or(rest),
363 None => url,
364 }
365}
366
367pub(crate) fn result_code_name(code: std::os::raw::c_int) -> String {
369 let name = match code {
370 1 => "SQLITE_ERROR",
371 2 => "SQLITE_INTERNAL",
372 3 => "SQLITE_PERM",
373 4 => "SQLITE_ABORT",
374 5 => "SQLITE_BUSY",
375 6 => "SQLITE_LOCKED",
376 7 => "SQLITE_NOMEM",
377 8 => "SQLITE_READONLY",
378 9 => "SQLITE_INTERRUPT",
379 10 => "SQLITE_IOERR",
380 11 => "SQLITE_CORRUPT",
381 12 => "SQLITE_NOTFOUND",
382 13 => "SQLITE_FULL",
383 14 => "SQLITE_CANTOPEN",
384 15 => "SQLITE_PROTOCOL",
385 16 => "SQLITE_EMPTY",
386 17 => "SQLITE_SCHEMA",
387 18 => "SQLITE_TOOBIG",
388 19 => "SQLITE_CONSTRAINT",
389 20 => "SQLITE_MISMATCH",
390 21 => "SQLITE_MISUSE",
391 22 => "SQLITE_NOLFS",
392 23 => "SQLITE_AUTH",
393 24 => "SQLITE_FORMAT",
394 25 => "SQLITE_RANGE",
395 26 => "SQLITE_NOTADB",
396 261 => "SQLITE_BUSY_RECOVERY",
397 517 => "SQLITE_BUSY_SNAPSHOT",
398 773 => "SQLITE_BUSY_TIMEOUT",
399 262 => "SQLITE_LOCKED_SHAREDCACHE",
400 264 => "SQLITE_READONLY_RECOVERY",
401 520 => "SQLITE_READONLY_CANTLOCK",
402 270 => "SQLITE_CANTOPEN_NOTEMPDIR",
403 526 => "SQLITE_CANTOPEN_ISDIR",
404 782 => "SQLITE_CANTOPEN_FULLPATH",
405 267 => "SQLITE_CORRUPT_VTAB",
406 275 => "SQLITE_CONSTRAINT_CHECK",
407 531 => "SQLITE_CONSTRAINT_COMMITHOOK",
408 787 => "SQLITE_CONSTRAINT_FOREIGNKEY",
409 1043 => "SQLITE_CONSTRAINT_FUNCTION",
410 1299 => "SQLITE_CONSTRAINT_NOTNULL",
411 1555 => "SQLITE_CONSTRAINT_PRIMARYKEY",
412 1811 => "SQLITE_CONSTRAINT_TRIGGER",
413 2067 => "SQLITE_CONSTRAINT_UNIQUE",
414 2323 => "SQLITE_CONSTRAINT_VTAB",
415 2579 => "SQLITE_CONSTRAINT_ROWID",
416 3091 => "SQLITE_CONSTRAINT_DATATYPE",
417 _ => {
418 let primary = code & 0xff;
420 return if primary != code && primary != 0 {
421 format!("{}({code})", result_code_name(primary))
422 } else {
423 format!("SQLITE({code})")
424 };
425 }
426 };
427 name.to_string()
428}
429
430fn version_row(r: &rusqlite::Row<'_>, cell_id: &str) -> rusqlite::Result<Version> {
431 Ok(Version { cell_id: cell_id.to_string(), tx: r.get(0)?, receipt_hash: r.get(1)?, asof: r.get(2)? })
432}
433
434fn get_bytes(c: &Connection, q: L, key: &str) -> Result<Option<Vec<u8>>> {
435 c.prepare_cached(q.sql()).db()?.query_row([key], |r| r.get(0)).optional().db()
436}
437
438fn get_strings(c: &Connection, q: L, key: &str) -> Result<Vec<String>> {
439 let mut st = c.prepare_cached(q.sql()).db()?;
440 let rows = st.query_map([key], |r| r.get(0)).db()?;
441 rows.collect::<rusqlite::Result<Vec<String>>>().db()
442}
443
444fn exists(c: &Connection, q: L, key: &str) -> Result<bool> {
445 c.prepare_cached(q.sql()).db()?.query_row([key], |r| r.get(0)).db()
446}
447
448fn exec(c: &Connection, q: L, params: impl rusqlite::Params) -> Result<usize> {
449 c.prepare_cached(q.sql()).db()?.execute(params).db()
450}
451
452fn bad(what: &str) -> Error {
454 Error::Invalid(datumesh_core::Error::new(ErrorCode::CanonType, "$", format!("a stored {what}")))
455}
456
457fn args_of(text: &str) -> Result<Args> {
459 let Ok(Value::Array(list)) = json::parse(text.as_bytes()) else { return Err(bad("argument list")) };
460 let mut args = Args::new();
461 for a in &list {
462 let item = |i: usize| match a {
463 Value::Array(x) => x.get(i).and_then(Value::as_str).ok_or_else(|| bad("argument")),
464 _ => Err(bad("argument")),
465 };
466 let ty = Type::from_name(item(1)?).ok_or_else(|| bad("type"))?;
467 args.insert(item(0)?.to_string(), Arg::Value(Scalar::from_lexical(ty, item(2)?)?));
468 }
469 Ok(args)
470}
471
472pub(crate) fn sort_key(ty: &str, lexical: &str) -> String {
476 if ty == "number" {
477 decimal_key(lexical)
478 } else {
479 lexical.to_string()
480 }
481}
482
483pub(crate) fn decimal_key(lexical: &str) -> String {
492 let (negative, magnitude) = match lexical.strip_prefix('-') {
493 Some(m) => (true, m),
494 None => (false, lexical),
495 };
496 let (int, frac) = magnitude.split_once('.').unwrap_or((magnitude, ""));
497 let int = int.trim_start_matches('0');
498 let frac = frac.trim_end_matches('0');
499 let (exponent, digits): (i64, String) = if !int.is_empty() {
500 (int.len() as i64, format!("{int}{frac}").trim_end_matches('0').to_string())
501 } else {
502 let zeros = frac.len() - frac.trim_start_matches('0').len();
503 (-(zeros as i64), frac[zeros..].to_string())
504 };
505 if digits.is_empty() {
506 return "1".to_string();
507 }
508 let e = (exponent + 500).clamp(0, 999);
509 if negative {
510 let complement: String = digits.bytes().map(|d| char::from(b'9' - (d - b'0'))).collect();
511 format!("0{:03}{complement}~", 999 - e)
512 } else {
513 format!("2{e:03}{digits}")
514 }
515}
516
517impl Resolver for SqliteStore {
518 fn descriptor(&self, fn_hash: &str) -> Option<Cow<'_, [u8]>> {
519 if !is_hash(fn_hash) {
520 return None;
521 }
522 self.resolve("descriptor", |c| get_bytes(c, L::DescriptorGet, fn_hash)).map(Cow::Owned)
523 }
524
525 fn receipt(&self, hash: &str) -> Option<Cow<'_, [u8]>> {
526 if !is_hash(hash) {
527 return None;
528 }
529 self.resolve("receipt", |c| get_bytes(c, L::ReceiptGet, hash)).map(Cow::Owned)
530 }
531
532 fn manifest(&self, entry: &str) -> Option<Cow<'_, [u8]>> {
535 if check_entry(entry, "$").is_err() {
536 return None;
537 }
538 self.resolve("manifest", |c| get_bytes(c, L::ManifestGet, entry)).map(Cow::Owned)
539 }
540
541 fn evaluate(&self, cell: &Cell, asof: &str) -> Option<Scalar> {
544 if !is_instant(asof) {
545 return None;
546 }
547 self.resolve("evaluate", |c| {
548 let Some(desc) = get_bytes(c, L::DescriptorGet, &cell.fn_hash)? else { return Ok(None) };
549 match FunctionDescriptor::from_json(&desc) {
550 Ok(d) if d.lang == "stored" => {}
551 Ok(_) => return Ok(None),
552 Err(e) => {
553 tracing::error!(target: "datumesh_store", fn_hash = %cell.fn_hash, error = %e, "stored descriptor does not parse");
554 return Ok(None);
555 }
556 }
557 let bytes: Option<Vec<u8>> = c
558 .prepare_cached(L::ValueAsof.sql())
559 .db()?
560 .query_row(rusqlite::params![cell.id(), asof], |r| r.get(0))
561 .optional()
562 .db()?;
563 let Some(bytes) = bytes else { return Ok(None) };
564 match Receipt::from_json(&bytes) {
565 Ok(r) => Ok(Some(r.value().clone())),
566 Err(e) => {
567 tracing::error!(target: "datumesh_store", cell_id = %cell.id(), error = %e, "stored receipt does not parse");
568 Ok(None)
569 }
570 }
571 })
572 }
573}
574
575impl Store for SqliteStore {
576 fn put_descriptor(&self, bytes: &[u8]) -> Result<String> {
577 let d = FunctionDescriptor::from_json(bytes)?;
578 let hash = d.hash();
579 let canonical = d.canonical();
580 self.run("put_descriptor", |c| {
581 if exists(c, L::DescriptorExists, &hash)? {
582 return Ok(());
583 }
584 self.locked_write(c, |c| exec(c, L::DescriptorPut, rusqlite::params![hash, canonical]).map(|_| ()))?;
585 self.bound_wal(c);
586 Ok(())
587 })?;
588 tracing::debug!(target: "datumesh_store", fn_hash = %hash, "put_descriptor");
589 Ok(hash)
590 }
591
592 fn put_receipt(&self, bytes: &[u8]) -> Result<String> {
593 let r = Receipt::from_json(bytes)?;
595 let canonical = json::parse(bytes)?.to_jcs();
596 let hash = r.hash().to_string();
597 let cell = r.cell();
598 let cell_id = cell.id();
599 let host = r.signer_host().to_string();
600 let (mut entries, mut inputs): (Vec<String>, Vec<String>) =
601 r.inputs().iter().cloned().partition(|e| e.starts_with("slice:"));
602 entries.sort();
603 entries.dedup();
604 inputs.sort();
605 inputs.dedup();
606 let mut args = Vec::with_capacity(cell.args.len());
607 for (name, a) in &cell.args {
608 let Some(s) = a.scalar() else {
609 return Err(Error::Invalid(datumesh_core::Error::new(
610 ErrorCode::Schema,
611 "$.cell.args",
612 "a receipt's cell has no open argument",
613 )));
614 };
615 let ty = s.ty().as_str();
616 args.push((name.clone(), ty, s.lexical().to_string(), sort_key(ty, s.lexical())));
617 }
618 self.run("put_receipt", |c| {
619 if let Some(stored) = get_bytes(c, L::ReceiptGet, &hash)? {
621 if stored == canonical {
622 return Ok(());
623 }
624 }
625 let mut keys = Keys::new();
628 if let Some(doc) = get_bytes(c, L::NamespaceGet, &host)? {
629 keys.insert_response(&host, 200, Some(&doc));
630 }
631 let out = verify(bytes, &keys, None, None);
632 if !out.ok {
633 return Err(Error::Invalid(out.error().unwrap_or_else(|| {
634 datumesh_core::Error::new(ErrorCode::Signature, "$", "the receipt did not verify")
635 })));
636 }
637 self.locked_write(c, |c| {
638 if let Some(stored) = get_bytes(c, L::ReceiptGet, &hash)? {
640 return if stored == canonical { Ok(()) } else { Err(Error::Conflict { hash: hash.clone() }) };
641 }
642 let value = r.value();
643 exec(
644 c,
645 L::ReceiptPut,
646 rusqlite::params![
647 hash,
648 canonical,
649 cell_id,
650 r.tx(),
651 cell.fn_hash,
652 r.publisher(),
653 r.asof(),
654 value.ty().as_str(),
655 value.lexical()
656 ],
657 )?;
658 for i in &inputs {
659 exec(c, L::EdgePut, rusqlite::params![hash, i])?;
660 }
661 for e in &entries {
662 exec(c, L::SliceEdgePut, rusqlite::params![hash, e])?;
663 }
664 if !exists(c, L::CellExists, &cell_id)? {
665 exec(
666 c,
667 L::CellPut,
668 rusqlite::params![cell_id, cell.fn_hash, cell.ns, cell.fn_name, args.len() as i64],
669 )?;
670 for (name, ty, value, key) in &args {
671 exec(c, L::CellArgPut, rusqlite::params![cell_id, name, ty, value, key, cell.fn_hash])?;
672 }
673 }
674 Ok(())
675 })?;
676 self.bound_wal(c);
677 Ok(())
678 })?;
679 tracing::debug!(target: "datumesh_store", hash = %hash, cell_id = %cell_id, "put_receipt");
680 Ok(hash)
681 }
682
683 fn put_blob(&self, bytes: &[u8]) -> Result<String> {
684 let sha = sha256_hex(bytes);
685 let length = i64::try_from(bytes.len()).unwrap_or(i64::MAX);
686 self.run("put_blob", |c| {
687 if exists(c, L::BlobExists, &sha)? {
688 return Ok(());
689 }
690 self.locked_write(c, |c| exec(c, L::BlobPut, rusqlite::params![sha, bytes, length]).map(|_| ()))?;
691 self.bound_wal(c);
692 Ok(())
693 })?;
694 tracing::debug!(target: "datumesh_store", sha256 = %sha, length, "put_blob");
695 Ok(sha)
696 }
697
698 fn blob(&self, sha256: &str) -> Result<Option<Cow<'_, [u8]>>> {
699 check_hex(sha256, 64, "sha256")?;
700 Ok(self.run("blob", |c| get_bytes(c, L::BlobGet, sha256))?.map(Cow::Owned))
701 }
702
703 fn put_version(&self, cell_id: &str, tx: &str, receipt_hash: &str) -> Result<()> {
704 check_hex(cell_id, 64, "cell_id")?;
705 check_instant(tx, "tx")?;
706 check_hex(receipt_hash, 64, "receipt_hash")?;
707 self.write("put_version", |c| {
708 let meta: Option<(String, String)> = c
709 .prepare_cached(L::ReceiptMeta.sql())
710 .db()?
711 .query_row([receipt_hash], |r| Ok((r.get(0)?, r.get(1)?)))
712 .optional()
713 .db()?;
714 let Some((r_cell, r_tx)) = meta else {
715 return Err(Error::NotFound { what: "receipt", key: receipt_hash.to_string() });
716 };
717 if r_cell != cell_id {
718 return Err(Error::Mismatch { what: "cell_id", expected: cell_id.to_string(), found: r_cell });
719 }
720 if r_tx != tx {
721 return Err(Error::Mismatch { what: "tx", expected: tx.to_string(), found: r_tx });
722 }
723 if let Some(l) = get_strings(c, L::VersionLatestTx, cell_id)?.pop() {
724 if tx <= l.as_str() {
725 return Err(Error::NotLater {
726 what: "version",
727 key: cell_id.to_string(),
728 given: tx.to_string(),
729 latest: l,
730 });
731 }
732 }
733 exec(c, L::VersionPut, rusqlite::params![cell_id, tx, receipt_hash])?;
734 Ok(())
735 })?;
736 tracing::debug!(target: "datumesh_store", cell_id, tx, receipt_hash, "put_version");
737 Ok(())
738 }
739
740 fn version_asof(&self, cell_id: &str, asof: &str) -> Result<Option<Version>> {
741 self.asof_version("version_asof", L::VersionAsof, cell_id, asof)
742 }
743
744 fn versions(&self, cell_id: &str, from: &str, to: &str) -> Result<Vec<Version>> {
745 check_hex(cell_id, 64, "cell_id")?;
746 check_instant(from, "from")?;
747 check_instant(to, "to")?;
748 self.run("versions", |c| {
749 let mut st = c.prepare_cached(L::VersionRange.sql()).db()?;
750 let rows = st.query_map(rusqlite::params![cell_id, from, to], |r| version_row(r, cell_id)).db()?;
751 rows.collect::<rusqlite::Result<Vec<Version>>>().db()
752 })
753 }
754
755 fn latest(&self, cell_id: &str) -> Result<Option<Version>> {
756 check_hex(cell_id, 64, "cell_id")?;
757 self.run("latest", |c| {
758 c.prepare_cached(L::VersionLatest.sql())
759 .db()?
760 .query_row([cell_id], |r| version_row(r, cell_id))
761 .optional()
762 .db()
763 })
764 }
765
766 fn dependents(&self, receipt_hash: &str) -> Result<Vec<String>> {
767 check_hex(receipt_hash, 64, "receipt_hash")?;
768 self.run("dependents", |c| get_strings(c, L::Dependents, receipt_hash))
769 }
770
771 fn inputs(&self, receipt_hash: &str) -> Result<Vec<String>> {
772 check_hex(receipt_hash, 64, "receipt_hash")?;
773 self.run("inputs", |c| get_strings(c, L::Inputs, receipt_hash))
774 }
775
776 fn put_namespace(&self, host: &str, keydoc_bytes: &[u8], validated_at: &str) -> Result<()> {
777 keys_from_response(host, 200, Some(keydoc_bytes))?;
778 check_instant(validated_at, "validated_at")?;
779 let canonical = json::parse(keydoc_bytes)?.to_jcs();
780 self.write("put_namespace", |c| {
781 let at: Option<String> =
782 c.prepare_cached(L::NamespaceGet.sql()).db()?.query_row([host], |r| r.get(1)).optional().db()?;
783 if let Some(at) = at {
784 if validated_at < at.as_str() {
785 return Err(Error::NotLater {
786 what: "namespace",
787 key: host.to_string(),
788 given: validated_at.to_string(),
789 latest: at,
790 });
791 }
792 }
793 exec(c, L::NamespacePut, rusqlite::params![host, canonical, validated_at])?;
794 Ok(())
795 })?;
796 tracing::debug!(target: "datumesh_store", host, validated_at, "put_namespace");
797 Ok(())
798 }
799
800 fn namespace(&self, host: &str) -> Result<Option<Namespace>> {
801 check_dns_name(host, "host")?;
802 self.run("namespace", |c| {
803 c.prepare_cached(L::NamespaceGet.sql())
804 .db()?
805 .query_row([host], |r| {
806 Ok(Namespace {
807 host: host.to_string(),
808 keydoc: r.get(0)?,
809 validated_at: r.get(1)?,
810 suspended_at: r.get(2)?,
811 })
812 })
813 .optional()
814 .db()
815 })
816 }
817
818 fn bind_name(&self, fn_name: &str, fn_hash: &str, tx: &str) -> Result<()> {
819 check_function_name(fn_name, "fn_name")?;
820 check_hex(fn_hash, 64, "fn_hash")?;
821 check_instant(tx, "tx")?;
822 self.write("bind_name", |c| {
823 if !exists(c, L::DescriptorExists, fn_hash)? {
824 return Err(Error::NotFound { what: "descriptor", key: fn_hash.to_string() });
825 }
826 if let Some(l) = get_strings(c, L::NameLatest, fn_name)?.pop() {
827 if tx <= l.as_str() {
828 return Err(Error::NotLater {
829 what: "name",
830 key: fn_name.to_string(),
831 given: tx.to_string(),
832 latest: l,
833 });
834 }
835 }
836 exec(c, L::NamePut, rusqlite::params![fn_name, tx, fn_hash])?;
837 Ok(())
838 })?;
839 tracing::debug!(target: "datumesh_store", fn_name, fn_hash, tx, "bind_name");
840 Ok(())
841 }
842
843 fn resolve_name(&self, fn_name: &str, asof: &str) -> Result<Option<String>> {
844 check_function_name(fn_name, "fn_name")?;
845 check_instant(asof, "asof")?;
846 self.run("resolve_name", |c| {
847 c.prepare_cached(L::NameAsof.sql()).db()?.query_row([fn_name, asof], |r| r.get(0)).optional().db()
848 })
849 }
850
851 fn resolve_binding(&self, fn_name: &str, asof: &str) -> Result<Option<Binding>> {
852 check_function_name(fn_name, "fn_name")?;
853 check_instant(asof, "asof")?;
854 self.run("resolve_binding", |c| {
855 c.prepare_cached(L::BindingAsof.sql())
856 .db()?
857 .query_row([fn_name, asof], |r| Ok(Binding { fn_hash: r.get(0)?, tx: r.get(1)? }))
858 .optional()
859 .db()
860 })
861 }
862
863 fn put_computed(&self, cell_id: &str, tx: &str, receipt_hash: &str) -> Result<()> {
864 check_hex(cell_id, 64, "cell_id")?;
865 check_instant(tx, "tx")?;
866 check_hex(receipt_hash, 64, "receipt_hash")?;
867 let asof = self.write("put_computed", |c| {
868 let meta: Option<(String, String, String)> = c
869 .prepare_cached(L::ReceiptMeta.sql())
870 .db()?
871 .query_row([receipt_hash], |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)))
872 .optional()
873 .db()?;
874 let Some((r_cell, r_tx, asof)) = meta else {
875 return Err(Error::NotFound { what: "receipt", key: receipt_hash.to_string() });
876 };
877 if r_cell != cell_id {
878 return Err(Error::Mismatch { what: "cell_id", expected: cell_id.to_string(), found: r_cell });
879 }
880 if r_tx != tx {
881 return Err(Error::Mismatch { what: "tx", expected: tx.to_string(), found: r_tx });
882 }
883 let stored: Option<String> = c
884 .prepare_cached(L::ComputedGet.sql())
885 .db()?
886 .query_row([cell_id, tx, asof.as_str()], |r| r.get(0))
887 .optional()
888 .db()?;
889 match stored {
890 Some(h) if h == receipt_hash => {}
891 Some(_) => return Err(Error::Conflict { hash: receipt_hash.to_string() }),
892 None => {
893 exec(c, L::ComputedPut, rusqlite::params![cell_id, tx, asof, receipt_hash])?;
894 }
895 }
896 Ok(asof)
897 })?;
898 tracing::debug!(target: "datumesh_store", cell_id, tx, asof, receipt_hash, "put_computed");
899 Ok(())
900 }
901
902 fn computed_asof(&self, cell_id: &str, asof: &str) -> Result<Option<Version>> {
903 self.asof_version("computed_asof", L::ComputedAsof, cell_id, asof)
904 }
905
906 fn stored_asof(&self, cell_id: &str, asof: &str) -> Result<Option<Version>> {
907 self.asof_version("stored_asof", L::StoredAsof, cell_id, asof)
908 }
909
910 fn cell_dependents(&self, cell_id: &str) -> Result<Vec<String>> {
911 check_hex(cell_id, 64, "cell_id")?;
912 self.run("cell_dependents", |c| get_strings(c, L::CellDependents, cell_id))
913 }
914
915 fn computed_asof_with_receipt(&self, cell_id: &str, asof: &str) -> Result<Option<(Version, Cow<'_, [u8]>)>> {
917 Ok(self
918 .asof_with_receipt("computed_asof_with_receipt", L::ComputedAsofReceipt, cell_id, asof)?
919 .map(|(v, b)| (v, Cow::Owned(b))))
920 }
921
922 fn stored_asof_with_receipt(&self, cell_id: &str, asof: &str) -> Result<Option<(Version, Cow<'_, [u8]>)>> {
924 Ok(self
925 .asof_with_receipt("stored_asof_with_receipt", L::StoredAsofReceipt, cell_id, asof)?
926 .map(|(v, b)| (v, Cow::Owned(b))))
927 }
928
929 fn put_manifest(&self, bytes: &[u8]) -> Result<String> {
930 let m = Manifest::from_json(bytes)?;
931 let entry = m.entry();
932 let canonical = m.canonical();
933 let members = m.members();
934 let n = i64::try_from(members.len()).unwrap_or(i64::MAX);
935 self.run("put_manifest", |c| {
936 if exists(c, L::ManifestExists, &entry)? {
938 return Ok(());
939 }
940 self.locked_write(c, |c| {
941 if exists(c, L::ManifestExists, &entry)? {
942 return Ok(());
943 }
944 exec(
945 c,
946 L::ManifestPut,
947 rusqlite::params![entry, m.slice().id(), m.members_hash(), m.slice().fn_hash(), n, canonical],
948 )?;
949 let mut st = c.prepare_cached(L::ManifestMemberPut.sql()).db()?;
950 for h in members {
951 st.execute([entry.as_str(), h.as_str()]).db()?;
952 }
953 Ok(())
954 })?;
955 self.bound_wal(c);
956 Ok(())
957 })?;
958 tracing::debug!(target: "datumesh_store", entry = %entry, members = n, "put_manifest");
959 Ok(entry)
960 }
961
962 fn member_manifests(&self, receipt_hash: &str) -> Result<Vec<String>> {
963 check_hex(receipt_hash, 64, "receipt_hash")?;
964 self.run("member_manifests", |c| get_strings(c, L::MemberManifests, receipt_hash))
965 }
966
967 fn slice_entries(&self, receipt_hash: &str) -> Result<Vec<String>> {
968 check_hex(receipt_hash, 64, "receipt_hash")?;
969 self.run("slice_entries", |c| get_strings(c, L::SliceEntries, receipt_hash))
970 }
971
972 fn slice_dependents(&self, entry: &str) -> Result<Vec<String>> {
973 check_entry(entry, "entry")?;
974 self.run("slice_dependents", |c| get_strings(c, L::SliceDependents, entry))
975 }
976
977 fn slice_members(&self, slice: &Slice, asof: &str) -> Result<Vec<SliceMember>> {
978 check_instant(asof, "asof")?;
979 let v = slice.to_value();
980 if let Some((cell_id, from, to)) = version_slice(slice, &v)? {
981 return self.versions_between(&cell_id, &from, &to, asof);
982 }
983 let p = Pattern::of(&v)?;
984 self.run("slice_members", |c| {
985 let rows = rows(&p, c, asof, [L::SliceMembersAll, L::SliceMembersDrive], |r| {
986 let cell_id: String = r.get(4)?;
987 Ok((
988 SliceMember {
989 version: Version { cell_id, tx: r.get(0)?, receipt_hash: r.get(1)?, asof: r.get(2)? },
990 receipt: r.get(3)?,
991 },
992 r.get::<_, String>(5)?,
993 ))
994 })?;
995 in_key_order(&p, rows, |m| &m.version.cell_id)
996 })
997 }
998
999 fn versions_between(&self, cell_id: &str, from: &str, to: &str, asof: &str) -> Result<Vec<SliceMember>> {
1000 check_hex(cell_id, 64, "cell_id")?;
1001 check_instant(from, "from")?;
1002 check_instant(to, "to")?;
1003 check_instant(asof, "asof")?;
1004 self.run("versions_between", |c| {
1005 let mut st = c.prepare_cached(L::VersionsBetween.sql()).db()?;
1006 let rows = st
1007 .query_map(rusqlite::params![cell_id, from, to, asof], |r| {
1008 Ok(SliceMember { version: version_row(r, cell_id)?, receipt: r.get(3)? })
1009 })
1010 .db()?;
1011 rows.collect::<rusqlite::Result<Vec<SliceMember>>>().db()
1012 })
1013 }
1014
1015 fn slice_values(&self, slice: &Slice, asof: &str) -> Result<Vec<SliceValue>> {
1016 check_instant(asof, "asof")?;
1017 let v = slice.to_value();
1018 if let Some((cell_id, from, to)) = version_slice(slice, &v)? {
1019 return self.run("slice_values", |c| {
1020 let mut st = c.prepare_cached(L::VersionsBetweenValues.sql()).db()?;
1021 let rows = st.query_map(rusqlite::params![cell_id, from, to, asof], value_columns).db()?;
1022 let mut out = Vec::new();
1023 for r in rows {
1024 out.push(value_of(r.db()?)?.0);
1025 }
1026 Ok(out)
1027 });
1028 }
1029 let p = Pattern::of(&v)?;
1030 self.run("slice_values", |c| {
1031 let rows = rows(&p, c, asof, [L::SliceValuesAll, L::SliceValuesDrive], value_columns)?;
1032 let mut values = Vec::with_capacity(rows.len());
1033 for row in rows {
1034 values.push(value_of(row)?);
1035 }
1036 in_key_order_parsed(&p, values, |m| &m.version.cell_id)
1037 })
1038 }
1039
1040 fn slice_changed_after(&self, slice: &Slice, asof: &str) -> Result<bool> {
1041 check_instant(asof, "asof")?;
1042 let v = slice.to_value();
1043 if let Some((cell_id, from, to)) = version_slice(slice, &v)? {
1044 return self.run("slice_changed_after", |c| {
1045 c.prepare_cached(L::VersionsChangedAfter.sql())
1046 .db()?
1047 .query_row(rusqlite::params![cell_id, from, to, asof], |r| r.get(0))
1048 .db()
1049 });
1050 }
1051 let p = Pattern::of(&v)?;
1052 self.run("slice_changed_after", |c| changed_after(&p, c, asof))
1053 }
1054}
1055
1056type ValueColumns = (String, String, String, String, String, String, String, String, String, String);
1059
1060fn value_columns(r: &rusqlite::Row<'_>) -> rusqlite::Result<ValueColumns> {
1061 Ok((r.get(0)?, r.get(1)?, r.get(2)?, r.get(3)?, r.get(4)?, r.get(5)?, r.get(6)?, r.get(7)?, r.get(8)?, r.get(9)?))
1062}
1063
1064fn value_of(c: ValueColumns) -> Result<(SliceValue, Args)> {
1066 let (tx, receipt_hash, asof, ty, value, cell_id, args, ns, fn_name, fn_hash) = c;
1067 let ty = Type::from_name(&ty).ok_or_else(|| bad("type"))?;
1068 let args = args_of(&args)?;
1069 let cell = Cell { args: args.clone(), fn_hash, fn_name, ns };
1070 Ok((
1071 SliceValue {
1072 version: Version { cell_id, tx, receipt_hash, asof },
1073 value: Scalar::from_lexical(ty, &value)?,
1074 cell,
1075 },
1076 args,
1077 ))
1078}
1079
1080fn constraints(p: &Pattern) -> String {
1083 let items: Vec<Value> = p
1084 .args
1085 .iter()
1086 .map(|a| {
1087 let (lo, hi) = match a.kind {
1088 Kind::Range => (sort_key(&a.ty, &a.lo), sort_key(&a.ty, &a.hi)),
1089 _ => (a.lo.clone(), a.hi.clone()),
1090 };
1091 Value::Array(
1092 [a.name.clone(), a.kind.as_str().to_string(), a.ty.clone(), lo, hi]
1093 .into_iter()
1094 .map(Value::String)
1095 .collect(),
1096 )
1097 })
1098 .collect();
1099 String::from_utf8(Value::Array(items).to_jcs()).unwrap_or_default()
1100}
1101
1102fn bind(st: &mut rusqlite::CachedStatement<'_>, params: &[(&str, SqlValue)]) -> Result<()> {
1104 for (name, value) in params {
1105 if let Some(i) = st.parameter_index(name).db()? {
1106 st.raw_bind_parameter(i, value).db()?;
1107 }
1108 }
1109 Ok(())
1110}
1111
1112fn drive<'p>(p: &'p Pattern, c: &Connection, count: bool) -> Result<(Option<&'p Constraint>, Option<i64>)> {
1116 let probe = |a: Option<&Constraint>| -> Result<i64> {
1117 match a {
1118 None => c.prepare_cached(L::ProbeCells.sql()).db()?.query_row([&p.fn_hash], |r| r.get(0)).db(),
1119 Some(a) => {
1120 let (lo, hi) = (sort_key(&a.ty, &a.lo), sort_key(&a.ty, &a.hi));
1121 c.prepare_cached(L::SliceProbe.sql())
1122 .db()?
1123 .query_row(rusqlite::params![p.fn_hash, a.name, a.ty, lo, hi], |r| r.get(0))
1124 .db()
1125 }
1126 }
1127 };
1128 let constrained: Vec<&Constraint> = p.args.iter().filter(|a| a.kind != Kind::Open).collect();
1129 Ok(match constrained.as_slice() {
1130 [] => (None, if count { Some(probe(None)?) } else { None }),
1131 [one] => (Some(*one), if count { Some(probe(Some(one))?) } else { None }),
1132 several => {
1133 let mut best: Option<(i64, &Constraint)> = None;
1134 for a in several {
1135 let n = probe(Some(a))?;
1136 if best.is_none_or(|(m, _)| n < m) {
1137 best = Some((n, a));
1138 }
1139 }
1140 (best.map(|(_, a)| a), best.map(|(n, _)| n))
1141 }
1142 })
1143}
1144
1145fn params(p: &Pattern, asof: &str, drive: Option<&Constraint>) -> Vec<(&'static str, SqlValue)> {
1147 let text = |s: &str| SqlValue::Text(s.to_string());
1148 let mut out = vec![
1149 (":fn_hash", text(&p.fn_hash)),
1150 (":ns", text(&p.ns)),
1151 (":fn", text(&p.fn_name)),
1152 (":nargs", SqlValue::Integer(p.args.len() as i64)),
1153 (":asof", text(asof)),
1154 (":cons", SqlValue::Text(constraints(p))),
1155 ];
1156 if let Some(a) = drive {
1157 out.push((":d_name", text(&a.name)));
1158 out.push((":d_type", text(&a.ty)));
1159 out.push((":d_lo", SqlValue::Text(sort_key(&a.ty, &a.lo))));
1160 out.push((":d_hi", SqlValue::Text(sort_key(&a.ty, &a.hi))));
1161 }
1162 out
1163}
1164
1165fn rows<T>(
1167 p: &Pattern,
1168 c: &Connection,
1169 asof: &str,
1170 qs: [L; 2],
1171 make: impl Fn(&rusqlite::Row<'_>) -> rusqlite::Result<T>,
1172) -> Result<Vec<T>> {
1173 let (d, _) = drive(p, c, false)?;
1174 let q = if d.is_some() { qs[1] } else { qs[0] };
1175 let mut st = c.prepare_cached(q.sql()).db()?;
1176 bind(&mut st, ¶ms(p, asof, d))?;
1177 let mut rows = st.raw_query();
1178 let mut out = Vec::new();
1179 while let Some(r) = rows.next().db()? {
1180 out.push(make(r).db()?);
1181 }
1182 Ok(out)
1183}
1184
1185fn key_of(p: &Pattern, args: &Args) -> Vec<String> {
1187 p.args
1188 .iter()
1189 .filter(|a| a.kind != Kind::Value)
1190 .filter_map(|a| args.get(&a.name).and_then(Arg::scalar))
1191 .map(|s| format!("{}:{}", s.ty().as_str(), s.lexical()))
1192 .collect()
1193}
1194
1195fn in_key_order<T>(p: &Pattern, rows: Vec<(T, String)>, cell_of: impl Fn(&T) -> &String) -> Result<Vec<T>> {
1197 let mut keyed = Vec::with_capacity(rows.len());
1198 for (m, args) in rows {
1199 keyed.push((key_of(p, &args_of(&args)?), m));
1200 }
1201 keyed.sort_by(|(a, ma), (b, mb)| cmp_keys(a, b).then_with(|| cell_of(ma).cmp(cell_of(mb))));
1202 Ok(keyed.into_iter().map(|(_, m)| m).collect())
1203}
1204
1205fn in_key_order_parsed<T>(p: &Pattern, rows: Vec<(T, Args)>, cell_of: impl Fn(&T) -> &String) -> Result<Vec<T>> {
1207 let mut keyed: Vec<(Vec<String>, T)> = rows.into_iter().map(|(m, args)| (key_of(p, &args), m)).collect();
1208 keyed.sort_by(|(a, ma), (b, mb)| cmp_keys(a, b).then_with(|| cell_of(ma).cmp(cell_of(mb))));
1209 Ok(keyed.into_iter().map(|(_, m)| m).collect())
1210}
1211
1212fn changed_after(p: &Pattern, c: &Connection, asof: &str) -> Result<bool> {
1216 let (d, candidates) = drive(p, c, true)?;
1217 let recent: i64 = c
1218 .prepare_cached(L::ProbeRecent.sql())
1219 .db()?
1220 .query_row(rusqlite::params![p.fn_hash, asof], |r| r.get(0))
1221 .db()?;
1222 let q = if recent <= candidates.unwrap_or(i64::MAX) {
1223 L::ChangedRecent
1224 } else if d.is_some() {
1225 L::ChangedCandDrive
1226 } else {
1227 L::ChangedCandAll
1228 };
1229 let d = if q == L::ChangedCandDrive { d } else { None };
1230 let mut st = c.prepare_cached(q.sql()).db()?;
1231 bind(&mut st, ¶ms(p, asof, d))?;
1232 let mut rows = st.raw_query();
1233 Ok(rows.next().db()?.is_some())
1234}
1235
1236#[cfg(test)]
1237mod tests {
1238 use super::*;
1239 use crate::slices::cmp_decimal;
1240
1241 #[test]
1242 fn decimal_keys_sort_as_the_numbers() {
1243 let mut numbers: Vec<String> = [
1244 "0", "9", "10", "-1.5", "-1", "-1.25", "-1.11", "-1.1", "-0.1", "-0.099", "0.099", "0.1", "0.109", "0.11",
1245 "1", "1.1", "1.109", "1.11", "9.5", "99.999", "100", "-100", "-10", "-9", "123", "-123", "0.0001",
1246 "-0.0001", "1000000", "-1000000", "12.5", "12.05", "-12.5", "-12.05", "120", "-120", "1.2", "-1.2",
1247 ]
1248 .iter()
1249 .map(|s| s.to_string())
1250 .collect();
1251 numbers.push("9".repeat(128));
1252 numbers.push(format!("-{}", "9".repeat(127)));
1253 numbers.push(format!("1{}", "0".repeat(127)));
1254 numbers.push(format!("-1{}", "0".repeat(126)));
1255 numbers.push(format!("0.{}1", "0".repeat(125)));
1256 numbers.push(format!("-0.{}1", "0".repeat(124)));
1257 numbers.push(format!("{}.{}", "1".repeat(64), "1".repeat(63)));
1258 numbers.push(format!("-{}.{}", "1".repeat(63), "1".repeat(63)));
1259 for a in &numbers {
1260 assert!(a.len() <= 128, "{a}");
1261 for b in &numbers {
1262 assert_eq!(decimal_key(a).cmp(&decimal_key(b)), cmp_decimal(a, b), "{a} vs {b}");
1263 }
1264 }
1265 assert_eq!(decimal_key("0"), "1");
1267 assert_eq!(decimal_key("100"), "25031");
1268 assert_eq!(decimal_key("0.0012"), "249812");
1269 assert_eq!(decimal_key("-1.5"), "049884~");
1270 assert_eq!(decimal_key(&"9".repeat(128)), format!("2628{}", "9".repeat(128)));
1271 }
1272
1273 #[test]
1274 fn result_codes_have_names() {
1275 assert_eq!(result_code_name(5), "SQLITE_BUSY");
1276 assert_eq!(result_code_name(1811), "SQLITE_CONSTRAINT_TRIGGER");
1277 assert_eq!(result_code_name(14), "SQLITE_CANTOPEN");
1278 assert_eq!(result_code_name(10 | (99 << 8)), "SQLITE_IOERR(25354)");
1279 }
1280
1281 #[test]
1282 fn sqlite_urls_name_paths() {
1283 assert_eq!(sqlite_path("sqlite:store.db"), "store.db");
1284 assert_eq!(sqlite_path("sqlite://store.db"), "store.db");
1285 assert_eq!(sqlite_path("sqlite:///var/lib/x.db"), "/var/lib/x.db");
1286 assert_eq!(sqlite_path("/var/lib/x.db"), "/var/lib/x.db");
1287 assert_eq!(sqlite_path("x.db"), "x.db");
1288 }
1289
1290 fn plan(c: &Connection, sql: &str) -> Vec<String> {
1292 let mut st = c.prepare(&format!("EXPLAIN QUERY PLAN {sql}")).unwrap();
1293 let n = st.parameter_count();
1294 for i in 1..=n {
1295 st.raw_bind_parameter(i, "x").unwrap();
1296 }
1297 let mut rows = st.raw_query();
1298 let mut out = Vec::new();
1299 while let Some(r) = rows.next().unwrap() {
1300 out.push(r.get::<_, String>(3).unwrap());
1301 }
1302 out
1303 }
1304
1305 #[test]
1310 fn bounded_slices_use_indexes() {
1311 let c = Connection::open_in_memory().unwrap();
1312 for m in crate::migrate::sqlite::MIGRATIONS {
1313 c.execute_batch(m.sql).unwrap();
1314 }
1315 c.execute_batch(
1317 "ANALYZE sqlite_schema;
1318 INSERT INTO sqlite_stat1 (tbl, idx, stat) VALUES
1319 ('cells', 'cells_fn_hash', '100000 100000 1'),
1320 ('cells', 'sqlite_autoindex_cells_1', '100000 1'),
1321 ('cell_args', 'cell_args_key', '200000 200000 100000 100000 1 1'),
1322 ('cell_args', 'sqlite_autoindex_cell_args_1', '200000 2 1'),
1323 ('versions', 'sqlite_autoindex_versions_1', '100000 1 1'),
1324 ('versions', 'versions_receipt_hash', '100000 1'),
1325 ('receipts', 'sqlite_autoindex_receipts_1', '100000 1'),
1326 ('receipts', 'receipts_cell_tx', '100000 1 1'),
1327 ('receipts', 'receipts_fn_hash_asof', '100000 100000 1'),
1328 ('receipts', 'receipts_cell_asof', '100000 1 1');
1329 ANALYZE sqlite_schema;",
1330 )
1331 .unwrap();
1332 let scans = |s: &str| s.starts_with("SCAN ") && !s.starts_with("SCAN k") && !s.starts_with("SCAN (subquery");
1335 for q in [
1336 L::SliceMembersDrive,
1337 L::SliceValuesDrive,
1338 L::ChangedCandDrive,
1339 L::ChangedRecent,
1340 L::SliceProbe,
1341 L::ProbeRecent,
1342 L::ProbeCells,
1343 L::VersionsBetween,
1344 L::VersionsBetweenValues,
1345 L::StoredAsof,
1346 L::ComputedAsof,
1347 L::CellDependents,
1348 ] {
1349 let steps = plan(&c, q.sql());
1350 assert!(!steps.iter().any(|s| scans(s)), "{q:?} scans:\n{steps:#?}");
1351 }
1352 for q in [L::SliceMembersAll, L::SliceValuesAll, L::ChangedCandAll] {
1354 let steps = plan(&c, q.sql());
1355 assert!(
1356 steps.iter().any(|s| s.contains("SEARCH c USING") && s.contains("cells_fn_hash")),
1357 "{q:?}: {steps:#?}"
1358 );
1359 assert!(!steps.iter().any(|s| scans(s)), "{q:?} scans:\n{steps:#?}");
1360 }
1361 }
1362}