uqa_storage/
backend.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! Persistent storage backend factory.
8//!
9//! This boundary keeps the engine from constructing SQLite-backed stores
10//! directly. Alternative persistent backends can implement the same factory
11//! without changing query execution code.
12
13use std::collections::BTreeMap;
14use std::path::{Path, PathBuf};
15use std::sync::atomic::{AtomicU64, Ordering};
16use std::sync::Arc;
17
18use uqa_analysis::Analyzer;
19use uqa_core::{DocId, Value};
20
21use crate::document_store::DocumentStore;
22use crate::inverted_index::InvertedIndex;
23use crate::sqlite::{
24    Catalog, ManagedConnection, SQLiteBTreeIndexStore, SQLiteDocumentStore, SQLiteError,
25    SQLiteHNSWIndex, SQLiteIVFIndex, SQLiteInvertedIndex, SQLiteVectorIndex,
26};
27use crate::vector_index::{VectorIndex, VectorIndexOpenMode, VectorIndexSpec};
28use crate::CatalogFacade;
29
30#[derive(Debug, thiserror::Error)]
31pub enum StorageBackendError {
32    #[error("text analysis failed: {0}")]
33    Analysis(#[from] uqa_analysis::AnalysisError),
34    #[error(transparent)]
35    SQLite(#[from] SQLiteError),
36    #[error("payload serialization failed: {0}")]
37    Serde(#[from] serde_json::Error),
38    #[error("{backend} storage failed: {source}")]
39    Backend {
40        backend: &'static str,
41        #[source]
42        source: Box<dyn std::error::Error + Send + Sync>,
43    },
44    #[error("{0}")]
45    Other(String),
46}
47
48impl StorageBackendError {
49    pub fn backend(
50        backend: &'static str,
51        source: impl std::error::Error + Send + Sync + 'static,
52    ) -> Self {
53        Self::Backend {
54            backend,
55            source: Box::new(source),
56        }
57    }
58}
59
60pub type StorageBackendResult<T> = std::result::Result<T, StorageBackendError>;
61
62/// Opaque transaction checkpoint identity. SQL savepoint names remain engine metadata and are never forwarded into a backend namespace.
63#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
64pub struct StorageSavepointId(u64);
65
66impl StorageSavepointId {
67    #[must_use]
68    pub fn allocate() -> Self {
69        static NEXT_ID: AtomicU64 = AtomicU64::new(1);
70        let id = NEXT_ID
71            .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| {
72                current.checked_add(1)
73            })
74            .expect("storage savepoint identity space exhausted");
75        Self(id)
76    }
77
78    pub(crate) fn backend_name(self) -> String {
79        self.0.to_string()
80    }
81}
82
83/// Session-bound catalog and physical storage handles created together.
84///
85/// Both handles must share the same transaction context. Keeping their
86/// construction behind one provider prevents a catalog write from escaping
87/// through a different connection or transaction than document/index writes.
88pub struct PersistentStorageSession {
89    pub catalog: Arc<dyn CatalogFacade>,
90    pub backend: Arc<dyn PersistentStorageBackend>,
91}
92
93/// Stable identity of one durable database. File identities allow engine coordination to extend across independently constructed providers and OS processes; opaque identities coordinate providers inside one process.
94#[derive(Clone, Debug, PartialEq, Eq, Hash)]
95pub enum PersistentStorageIdentity {
96    File(PathBuf),
97    Opaque(String),
98}
99
100impl PersistentStorageIdentity {
101    /// Resolve the stable file identity for a database path. The database file itself may not exist yet because backends materialize it on first write, so a missing file anchors the identity on its canonicalized parent directory instead of failing.
102    pub fn for_database_path(path: &Path) -> StorageBackendResult<Self> {
103        let path = resolve_final_symlinks(path)?;
104        match std::fs::canonicalize(&path) {
105            Ok(canonical) => Ok(Self::File(canonical)),
106            Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
107                let file_name = path.file_name().ok_or_else(|| {
108                    StorageBackendError::Other(format!(
109                        "database path `{}` has no file name",
110                        path.display()
111                    ))
112                })?;
113                let parent = path
114                    .parent()
115                    .filter(|parent| !parent.as_os_str().is_empty())
116                    .map_or_else(|| PathBuf::from("."), Path::to_path_buf);
117                let parent = std::fs::canonicalize(&parent).map_err(|error| {
118                    StorageBackendError::Other(format!(
119                        "canonicalize database directory `{}`: {error}",
120                        parent.display()
121                    ))
122                })?;
123                Ok(Self::File(parent.join(file_name)))
124            }
125            Err(error) => Err(StorageBackendError::Other(format!(
126                "canonicalize database `{}`: {error}",
127                path.display()
128            ))),
129        }
130    }
131}
132
133fn resolve_final_symlinks(path: &Path) -> StorageBackendResult<PathBuf> {
134    let mut current = path.to_path_buf();
135    let mut followed = 0usize;
136    loop {
137        match std::fs::symlink_metadata(&current) {
138            Ok(metadata) if metadata.file_type().is_symlink() => {
139                followed += 1;
140                if followed > 40 {
141                    return Err(StorageBackendError::Other(format!(
142                        "database path `{}` has too many symbolic-link levels",
143                        path.display()
144                    )));
145                }
146                let target = std::fs::read_link(&current).map_err(|error| {
147                    StorageBackendError::Other(format!(
148                        "read database symbolic link `{}`: {error}",
149                        current.display()
150                    ))
151                })?;
152                current = if target.is_absolute() {
153                    target
154                } else {
155                    current
156                        .parent()
157                        .filter(|parent| !parent.as_os_str().is_empty())
158                        .unwrap_or_else(|| Path::new("."))
159                        .join(target)
160                };
161            }
162            Ok(_) => return Ok(current),
163            Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(current),
164            Err(error) => {
165                return Err(StorageBackendError::Other(format!(
166                    "inspect database path `{}`: {error}",
167                    current.display()
168                )))
169            }
170        }
171    }
172}
173
174#[cfg(test)]
175mod identity_tests {
176    use super::*;
177
178    #[cfg(unix)]
179    #[test]
180    fn dangling_database_symlink_keeps_the_target_identity_after_creation() {
181        use std::os::unix::fs::symlink;
182
183        let directory = tempfile::tempdir().unwrap();
184        let target = directory.path().join("target.db");
185        let link = directory.path().join("database.db");
186        symlink("target.db", &link).unwrap();
187
188        let before = PersistentStorageIdentity::for_database_path(&link).unwrap();
189        std::fs::File::create(&target).unwrap();
190        let after = PersistentStorageIdentity::for_database_path(&link).unwrap();
191
192        assert_eq!(before, after);
193        assert_eq!(
194            after,
195            PersistentStorageIdentity::File(target.canonicalize().unwrap())
196        );
197    }
198}
199
200impl PersistentStorageSession {
201    pub fn new(
202        catalog: Arc<dyn CatalogFacade>,
203        backend: Arc<dyn PersistentStorageBackend>,
204    ) -> Self {
205        Self { catalog, backend }
206    }
207}
208
209/// Factory for independent sessions over one durable database.
210///
211/// A provider owns the database-level resource while each returned session
212/// owns its transaction state. This is the engine-facing extension point for
213/// `SQLite`, redb, and application-defined Key/Value stores.
214pub trait PersistentStorageProvider: Send + Sync {
215    fn open_session(&self) -> StorageBackendResult<PersistentStorageSession>;
216
217    /// Return the database identity shared by every session this provider opens. Custom providers that cannot expose a stable identity may keep the default; engines built from the same `Arc` provider still share an in-process coordinator.
218    fn storage_identity(&self) -> StorageBackendResult<Option<PersistentStorageIdentity>> {
219        Ok(None)
220    }
221}
222
223/// Factory plus transaction surface for persistent table/index storage.
224pub trait PersistentStorageBackend: Send + Sync {
225    /// Return the stable database identity for independently constructed engines over this backend. File identities also enable cross-process row-lock coordination.
226    fn storage_identity(&self) -> StorageBackendResult<Option<PersistentStorageIdentity>> {
227        Ok(None)
228    }
229
230    /// Open a transaction-isolated catalog/backend pair over the same durable database. Engines constructed from already-open backends retain this factory so a row-lock recheck can read the latest committed tuple while the caller's statement snapshot remains pinned.
231    fn open_session(&self) -> StorageBackendResult<PersistentStorageSession> {
232        Err(StorageBackendError::Other(
233            "independent sessions are not implemented for this persistent backend".into(),
234        ))
235    }
236
237    /// Whether an independently pinned read session can remain open while another session writes the same database. Backends that return `false` use a detached fixed snapshot for `REPEATABLE READ` and `SERIALIZABLE` transactions.
238    fn supports_concurrent_pinned_read_and_write(&self) -> bool {
239        false
240    }
241
242    fn document_store(&self, table: &str) -> Box<dyn DocumentStore>;
243
244    /// Upgrade backend-owned document records before table handles are restored. Implementations must make the rewrite atomic and idempotent.
245    fn migrate_document_storage(&self) -> StorageBackendResult<()> {
246        Ok(())
247    }
248
249    fn inverted_index(&self, table: &str, analyzer: Analyzer) -> Box<dyn InvertedIndex>;
250
251    /// Upgrade backend-owned inverted-index values before table handles are
252    /// restored. Implementations must make the rewrite atomic and idempotent.
253    fn migrate_inverted_index_storage(&self) -> StorageBackendResult<()> {
254        Ok(())
255    }
256
257    fn vector_index(
258        &self,
259        table: &str,
260        field: &str,
261        dimensions: u32,
262        spec: VectorIndexSpec,
263        mode: VectorIndexOpenMode,
264    ) -> StorageBackendResult<Box<dyn VectorIndex>>;
265
266    fn drop_vector_index_metadata(&self, _table: &str, _field: &str) -> StorageBackendResult<()> {
267        Ok(())
268    }
269
270    /// Whether this backend maps logical `btree` indexes to durable postings.
271    fn persists_btree_indexes(&self) -> bool {
272        false
273    }
274
275    /// `Some(entries)` is a complete persisted index; `None` means it has not
276    /// been built yet and the engine must backfill it from documents once.
277    fn load_btree_index(
278        &self,
279        _table: &str,
280        _field: &crate::ValueIndexKey,
281    ) -> StorageBackendResult<Option<Vec<(DocId, Value)>>> {
282        Ok(None)
283    }
284
285    fn btree_index_fields(&self, _table: &str) -> StorageBackendResult<Vec<crate::ValueIndexKey>> {
286        Ok(Vec::new())
287    }
288
289    /// Fields whose persisted posting support was found inconsistent during a
290    /// schema migration. The engine repairs these at its explicit open-time
291    /// write boundary and clears each durable retry marker only after success.
292    fn btree_index_repairs(&self) -> StorageBackendResult<Vec<(String, crate::ValueIndexKey)>> {
293        Ok(Vec::new())
294    }
295
296    fn clear_btree_index_repair(
297        &self,
298        _table: &str,
299        _field: &crate::ValueIndexKey,
300    ) -> StorageBackendResult<()> {
301        Ok(())
302    }
303
304    fn replace_btree_index(
305        &self,
306        _table: &str,
307        _field: &crate::ValueIndexKey,
308        _values: &[(DocId, Value)],
309    ) -> StorageBackendResult<()> {
310        Ok(())
311    }
312
313    /// Repair sparse support differences without requiring capable backends
314    /// to rewrite every already-valid posting. The complete replacement is
315    /// supplied for the storage-neutral fallback.
316    fn repair_btree_index(
317        &self,
318        table: &str,
319        field: &crate::ValueIndexKey,
320        complete: &[(DocId, Value)],
321        _stale_doc_ids: &[DocId],
322        _missing: &[(DocId, Value)],
323    ) -> StorageBackendResult<()> {
324        self.replace_btree_index(table, field, complete)
325    }
326
327    /// Replace several complete indexes for one table atomically. Backends
328    /// may override this to share one transaction and prepared statements.
329    fn replace_btree_indexes(
330        &self,
331        table: &str,
332        indexes: &[(&crate::ValueIndexKey, &[(DocId, Value)])],
333    ) -> StorageBackendResult<()> {
334        for (field, values) in indexes {
335            self.replace_btree_index(table, field, values)?;
336        }
337        Ok(())
338    }
339
340    fn apply_btree_index_write(
341        &self,
342        _table: &str,
343        _doc_id: DocId,
344        _values: Option<&BTreeMap<crate::ValueIndexKey, Value>>,
345    ) -> StorageBackendResult<()> {
346        Ok(())
347    }
348
349    fn drop_btree_index(
350        &self,
351        _table: &str,
352        _field: &crate::ValueIndexKey,
353    ) -> StorageBackendResult<()> {
354        Ok(())
355    }
356
357    fn clear_btree_indexes(&self, _table: &str) -> StorageBackendResult<()> {
358        Ok(())
359    }
360
361    /// Reclaim backend-owned storage outside a transaction. Backends whose logical stores eagerly remove obsolete values may keep the no-op default; durable backends with file-level compaction should override it.
362    fn vacuum(&self) -> StorageBackendResult<()> {
363        Ok(())
364    }
365
366    fn begin_transaction(&self) -> StorageBackendResult<()>;
367
368    /// Begin a transaction whose first operation is expected to be a read.
369    /// Backends with distinct lock modes may defer write-lock acquisition;
370    /// the default preserves existing transaction semantics.
371    fn begin_read_transaction(&self) -> StorageBackendResult<()> {
372        self.begin_transaction()
373    }
374
375    /// Begin an atomic transaction that may remain read-only or perform writes after its initial reads. Backends without read-to-write promotion acquire a writer transaction immediately.
376    fn begin_upgradeable_transaction(&self) -> StorageBackendResult<()> {
377        self.begin_transaction()
378    }
379
380    /// Whether this session currently owns a pinned storage transaction.
381    fn in_transaction(&self) -> bool;
382
383    /// Whether the current transaction has performed a physical write.
384    ///
385    /// The engine uses this to enforce read-only statement transactions even
386    /// for writes made through catalog/index helpers it did not classify.
387    fn transaction_has_written(&self) -> StorageBackendResult<bool>;
388
389    /// Backend commit generation visible to this session, when available.
390    /// A changing value invalidates session-local catalog and index caches.
391    fn change_version(&self) -> StorageBackendResult<Option<u64>> {
392        Ok(None)
393    }
394
395    /// Whether reading [`Self::change_version`] can proceed while this
396    /// session owns its write transaction.
397    fn change_version_monitor_is_nonblocking(&self) -> StorageBackendResult<bool> {
398        Ok(true)
399    }
400
401    /// Pin the transaction's read snapshot before cache restoration.
402    fn pin_transaction_snapshot(&self) -> StorageBackendResult<()> {
403        Ok(())
404    }
405
406    fn commit_transaction(&self) -> StorageBackendResult<()>;
407
408    fn rollback_transaction(&self) -> StorageBackendResult<()>;
409
410    fn savepoint(&self, id: StorageSavepointId) -> StorageBackendResult<()>;
411
412    fn release_savepoint(&self, id: StorageSavepointId) -> StorageBackendResult<()>;
413
414    fn rollback_to_savepoint(&self, id: StorageSavepointId) -> StorageBackendResult<()>;
415}
416
417#[derive(Clone)]
418pub struct SQLiteStorageBackend {
419    conn: ManagedConnection,
420}
421
422impl SQLiteStorageBackend {
423    pub fn new(conn: ManagedConnection) -> Self {
424        Self { conn }
425    }
426
427    pub fn connection(&self) -> ManagedConnection {
428        self.conn.clone()
429    }
430
431    /// Create a backend whose stores use an independent transaction session
432    /// over the same physical `SQLite` pool.
433    #[must_use]
434    pub fn new_session(&self) -> Self {
435        Self::new(self.conn.new_session())
436    }
437}
438
439/// Database-level owner that creates isolated `SQLite` engine sessions.
440#[derive(Clone)]
441pub struct SQLiteStorageProvider {
442    connection: ManagedConnection,
443}
444
445impl SQLiteStorageProvider {
446    pub fn new(connection: ManagedConnection) -> Self {
447        Self { connection }
448    }
449}
450
451impl PersistentStorageProvider for SQLiteStorageProvider {
452    fn open_session(&self) -> StorageBackendResult<PersistentStorageSession> {
453        let connection = self.connection.new_session();
454        let catalog: Arc<dyn CatalogFacade> = Arc::new(Catalog::open(connection.clone())?);
455        let backend: Arc<dyn PersistentStorageBackend> =
456            Arc::new(SQLiteStorageBackend::new(connection));
457        Ok(PersistentStorageSession::new(catalog, backend))
458    }
459
460    fn storage_identity(&self) -> StorageBackendResult<Option<PersistentStorageIdentity>> {
461        let Some(path) = self.connection.database_path() else {
462            return Ok(None);
463        };
464        PersistentStorageIdentity::for_database_path(path)
465            .map(Some)
466            .map_err(|error| {
467                StorageBackendError::Other(format!(
468                    "resolve SQLite database identity `{}`: {error}",
469                    path.display()
470                ))
471            })
472    }
473}
474
475impl PersistentStorageBackend for SQLiteStorageBackend {
476    fn storage_identity(&self) -> StorageBackendResult<Option<PersistentStorageIdentity>> {
477        let Some(path) = self.conn.database_path() else {
478            return Ok(None);
479        };
480        PersistentStorageIdentity::for_database_path(path).map(Some)
481    }
482
483    fn open_session(&self) -> StorageBackendResult<PersistentStorageSession> {
484        let connection = self.conn.new_session();
485        let catalog: Arc<dyn CatalogFacade> = Arc::new(Catalog::open(connection.clone())?);
486        let backend: Arc<dyn PersistentStorageBackend> = Arc::new(Self::new(connection));
487        Ok(PersistentStorageSession::new(catalog, backend))
488    }
489
490    fn supports_concurrent_pinned_read_and_write(&self) -> bool {
491        self.conn.supports_concurrent_pinned_read_and_write()
492    }
493
494    fn document_store(&self, table: &str) -> Box<dyn DocumentStore> {
495        Box::new(SQLiteDocumentStore::new(self.conn.clone(), table))
496    }
497
498    fn inverted_index(&self, table: &str, analyzer: Analyzer) -> Box<dyn InvertedIndex> {
499        Box::new(SQLiteInvertedIndex::new(self.conn.clone(), table, analyzer))
500    }
501
502    fn vector_index(
503        &self,
504        table: &str,
505        field: &str,
506        dimensions: u32,
507        spec: VectorIndexSpec,
508        mode: VectorIndexOpenMode,
509    ) -> StorageBackendResult<Box<dyn VectorIndex>> {
510        let index: Box<dyn VectorIndex> = match spec {
511            VectorIndexSpec::BruteForce => Box::new(SQLiteVectorIndex::new(
512                self.conn.clone(),
513                table,
514                field,
515                dimensions,
516            )),
517            VectorIndexSpec::IVF(params) => {
518                params.validate()?;
519                match mode {
520                    VectorIndexOpenMode::Create => Box::new(SQLiteIVFIndex::with_params(
521                        self.conn.clone(),
522                        table,
523                        field,
524                        dimensions,
525                        params.nlist,
526                        params.nprobe,
527                        params.train_threshold,
528                    )),
529                    VectorIndexOpenMode::Restore => Box::new(SQLiteIVFIndex::open_existing(
530                        self.conn.clone(),
531                        table,
532                        field,
533                        dimensions,
534                        params.nlist,
535                        params.nprobe,
536                        params.train_threshold,
537                    )),
538                }
539            }
540            VectorIndexSpec::HNSW(params) => {
541                params.validate()?;
542                match mode {
543                    VectorIndexOpenMode::Create => Box::new(SQLiteHNSWIndex::with_params(
544                        self.conn.clone(),
545                        table,
546                        field,
547                        dimensions,
548                        params,
549                    )),
550                    VectorIndexOpenMode::Restore => {
551                        let index = SQLiteHNSWIndex::open_existing(
552                            self.conn.clone(),
553                            table,
554                            field,
555                            dimensions,
556                            params,
557                        );
558                        index.validate_existing()?;
559                        Box::new(index)
560                    }
561                }
562            }
563        };
564        Ok(index)
565    }
566
567    fn drop_vector_index_metadata(&self, table: &str, field: &str) -> StorageBackendResult<()> {
568        SQLiteIVFIndex::drop_metadata(&self.conn, table, field)?;
569        SQLiteHNSWIndex::drop_metadata(&self.conn, table, field)?;
570        Ok(())
571    }
572
573    fn persists_btree_indexes(&self) -> bool {
574        true
575    }
576
577    fn load_btree_index(
578        &self,
579        table: &str,
580        field: &crate::ValueIndexKey,
581    ) -> StorageBackendResult<Option<Vec<(DocId, Value)>>> {
582        Ok(SQLiteBTreeIndexStore::new(self.conn.clone()).load(table, field)?)
583    }
584
585    fn btree_index_fields(&self, table: &str) -> StorageBackendResult<Vec<crate::ValueIndexKey>> {
586        Ok(SQLiteBTreeIndexStore::new(self.conn.clone()).fields(table)?)
587    }
588
589    fn btree_index_repairs(&self) -> StorageBackendResult<Vec<(String, crate::ValueIndexKey)>> {
590        Ok(SQLiteBTreeIndexStore::new(self.conn.clone()).repairs()?)
591    }
592
593    fn clear_btree_index_repair(
594        &self,
595        table: &str,
596        field: &crate::ValueIndexKey,
597    ) -> StorageBackendResult<()> {
598        SQLiteBTreeIndexStore::new(self.conn.clone()).clear_repair(table, field)?;
599        Ok(())
600    }
601
602    fn replace_btree_index(
603        &self,
604        table: &str,
605        field: &crate::ValueIndexKey,
606        values: &[(DocId, Value)],
607    ) -> StorageBackendResult<()> {
608        SQLiteBTreeIndexStore::new(self.conn.clone()).replace(table, field, values)?;
609        Ok(())
610    }
611
612    fn repair_btree_index(
613        &self,
614        table: &str,
615        field: &crate::ValueIndexKey,
616        _complete: &[(DocId, Value)],
617        stale_doc_ids: &[DocId],
618        missing: &[(DocId, Value)],
619    ) -> StorageBackendResult<()> {
620        SQLiteBTreeIndexStore::new(self.conn.clone()).repair(
621            table,
622            field,
623            stale_doc_ids,
624            missing,
625        )?;
626        Ok(())
627    }
628
629    fn replace_btree_indexes(
630        &self,
631        table: &str,
632        indexes: &[(&crate::ValueIndexKey, &[(DocId, Value)])],
633    ) -> StorageBackendResult<()> {
634        SQLiteBTreeIndexStore::new(self.conn.clone()).replace_many(table, indexes)?;
635        Ok(())
636    }
637
638    fn apply_btree_index_write(
639        &self,
640        table: &str,
641        doc_id: DocId,
642        values: Option<&BTreeMap<crate::ValueIndexKey, Value>>,
643    ) -> StorageBackendResult<()> {
644        SQLiteBTreeIndexStore::new(self.conn.clone()).apply_write(table, doc_id, values)?;
645        Ok(())
646    }
647
648    fn drop_btree_index(
649        &self,
650        table: &str,
651        field: &crate::ValueIndexKey,
652    ) -> StorageBackendResult<()> {
653        SQLiteBTreeIndexStore::new(self.conn.clone()).drop_index(table, field)?;
654        Ok(())
655    }
656
657    fn clear_btree_indexes(&self, table: &str) -> StorageBackendResult<()> {
658        SQLiteBTreeIndexStore::new(self.conn.clone()).clear_table(table)?;
659        Ok(())
660    }
661
662    fn vacuum(&self) -> StorageBackendResult<()> {
663        self.conn.vacuum()?;
664        Ok(())
665    }
666
667    fn begin_transaction(&self) -> StorageBackendResult<()> {
668        self.conn.begin_transaction()?;
669        Ok(())
670    }
671
672    fn begin_read_transaction(&self) -> StorageBackendResult<()> {
673        self.conn.begin_deferred_transaction()?;
674        Ok(())
675    }
676
677    fn begin_upgradeable_transaction(&self) -> StorageBackendResult<()> {
678        self.conn.begin_deferred_transaction()?;
679        Ok(())
680    }
681
682    fn in_transaction(&self) -> bool {
683        self.conn.in_transaction()
684    }
685
686    fn transaction_has_written(&self) -> StorageBackendResult<bool> {
687        Ok(self.conn.transaction_has_written()?)
688    }
689
690    fn change_version(&self) -> StorageBackendResult<Option<u64>> {
691        Ok(self.conn.data_version()?)
692    }
693
694    fn change_version_monitor_is_nonblocking(&self) -> StorageBackendResult<bool> {
695        Ok(self.conn.data_version_monitor_is_nonblocking()?)
696    }
697
698    fn pin_transaction_snapshot(&self) -> StorageBackendResult<()> {
699        self.conn.pin_transaction_snapshot()?;
700        Ok(())
701    }
702
703    fn commit_transaction(&self) -> StorageBackendResult<()> {
704        self.conn.commit_transaction()?;
705        Ok(())
706    }
707
708    fn rollback_transaction(&self) -> StorageBackendResult<()> {
709        self.conn.rollback_transaction()?;
710        Ok(())
711    }
712
713    fn savepoint(&self, id: StorageSavepointId) -> StorageBackendResult<()> {
714        self.conn.savepoint(&id.backend_name())?;
715        Ok(())
716    }
717
718    fn release_savepoint(&self, id: StorageSavepointId) -> StorageBackendResult<()> {
719        self.conn.release_savepoint(&id.backend_name())?;
720        Ok(())
721    }
722
723    fn rollback_to_savepoint(&self, id: StorageSavepointId) -> StorageBackendResult<()> {
724        self.conn.rollback_to_savepoint(&id.backend_name())?;
725        Ok(())
726    }
727}
728
729#[cfg(test)]
730mod tests {
731    use std::collections::BTreeMap;
732
733    use uqa_analysis::analyzer::standard_analyzer;
734    use uqa_core::Value;
735
736    use super::*;
737    use crate::sqlite::Catalog;
738
739    #[test]
740    fn sqlite_backend_builds_document_index_and_vector_stores() {
741        let conn = ManagedConnection::open_in_memory().unwrap();
742        let _catalog = Catalog::open(conn.clone()).unwrap();
743        let backend = SQLiteStorageBackend::new(conn);
744
745        let mut doc = BTreeMap::new();
746        doc.insert("title".to_string(), Value::Str("rust storage".into()));
747        let mut docs = backend.document_store("articles");
748        docs.put(1, doc).unwrap();
749        assert_eq!(
750            docs.get_field(1, "title").unwrap(),
751            Some(Value::Str("rust storage".into()))
752        );
753
754        let mut inv = backend.inverted_index("articles", standard_analyzer("english"));
755        inv.add_document(
756            1,
757            BTreeMap::from([("title".to_string(), "rust storage".to_string())]),
758        )
759        .unwrap();
760        assert_eq!(inv.doc_freq("title", "rust").unwrap(), 1);
761
762        let mut vectors = backend
763            .vector_index(
764                "articles",
765                "embedding",
766                2,
767                VectorIndexSpec::IVF(crate::IVFIndexParams {
768                    nlist: 2,
769                    nprobe: 1,
770                    train_threshold: 2,
771                }),
772                VectorIndexOpenMode::Create,
773            )
774            .unwrap();
775        vectors.add(1, vec![1.0, 0.0]).unwrap();
776        let hits = vectors.search_knn(&[1.0, 0.0], 1).unwrap();
777        assert_eq!(hits.entries().len(), 1);
778        assert_eq!(hits.entries()[0].doc_id, 1);
779    }
780
781    #[test]
782    fn sqlite_backend_transaction_rolls_back_cross_store_writes() {
783        let conn = ManagedConnection::open_in_memory().unwrap();
784        let _catalog = Catalog::open(conn.clone()).unwrap();
785        let backend = SQLiteStorageBackend::new(conn);
786        let mut docs = backend.document_store("articles");
787        let mut inv = backend.inverted_index("articles", standard_analyzer("english"));
788
789        backend.begin_transaction().unwrap();
790        docs.put(
791            1,
792            BTreeMap::from([("title".to_string(), Value::Str("rollback".into()))]),
793        )
794        .unwrap();
795        inv.add_document(
796            1,
797            BTreeMap::from([("title".to_string(), "rollback".to_string())]),
798        )
799        .unwrap();
800        backend.rollback_transaction().unwrap();
801
802        assert_eq!(docs.len().unwrap(), 0);
803        assert_eq!(inv.doc_freq("title", "rollback").unwrap(), 0);
804    }
805
806    #[test]
807    fn sqlite_sessions_isolate_and_atomically_commit_cross_store_writes() {
808        let dir = tempfile::tempdir().unwrap();
809        let path = dir.path().join("cross-store-isolation.sqlite3");
810        let conn = ManagedConnection::open(&path).unwrap();
811        let catalog = Catalog::open(conn.clone()).unwrap();
812        let writer = SQLiteStorageBackend::new(conn.clone());
813        let observer_conn = conn.new_session();
814        let observer_catalog = Catalog::open(observer_conn.clone()).unwrap();
815        let observer = SQLiteStorageBackend::new(observer_conn);
816
817        let mut writer_docs = writer.document_store("articles");
818        let mut writer_inv = writer.inverted_index("articles", standard_analyzer("english"));
819        let mut writer_vectors = writer
820            .vector_index(
821                "articles",
822                "embedding",
823                2,
824                VectorIndexSpec::BruteForce,
825                VectorIndexOpenMode::Create,
826            )
827            .unwrap();
828        let observer_docs = observer.document_store("articles");
829        let observer_inv = observer.inverted_index("articles", standard_analyzer("english"));
830        let observer_vectors = observer
831            .vector_index(
832                "articles",
833                "embedding",
834                2,
835                VectorIndexSpec::BruteForce,
836                VectorIndexOpenMode::Restore,
837            )
838            .unwrap();
839
840        writer.begin_transaction().unwrap();
841        writer_docs
842            .put(
843                1,
844                BTreeMap::from([("title".to_string(), Value::Str("atomic rust".into()))]),
845            )
846            .unwrap();
847        writer_inv
848            .add_document(
849                1,
850                BTreeMap::from([("title".to_string(), "atomic rust".to_string())]),
851            )
852            .unwrap();
853        writer_vectors.add(1, vec![1.0, 0.0]).unwrap();
854        catalog
855            .save_scoring_params("transactional", r#"{"alpha":1.0}"#)
856            .unwrap();
857
858        assert_eq!(writer_docs.len().unwrap(), 1);
859        assert_eq!(writer_inv.doc_freq("title", "rust").unwrap(), 1);
860        assert_eq!(writer_vectors.count().unwrap(), 1);
861        assert!(catalog
862            .load_scoring_params("transactional")
863            .unwrap()
864            .is_some());
865
866        assert_eq!(observer_docs.len().unwrap(), 0);
867        assert_eq!(observer_inv.doc_freq("title", "rust").unwrap(), 0);
868        assert_eq!(observer_vectors.count().unwrap(), 0);
869        assert!(observer_catalog
870            .load_scoring_params("transactional")
871            .unwrap()
872            .is_none());
873
874        writer.commit_transaction().unwrap();
875        assert_eq!(observer_docs.len().unwrap(), 1);
876        assert_eq!(observer_inv.doc_freq("title", "rust").unwrap(), 1);
877        assert_eq!(observer_vectors.count().unwrap(), 1);
878        assert!(observer_catalog
879            .load_scoring_params("transactional")
880            .unwrap()
881            .is_some());
882    }
883
884    #[test]
885    fn ignored_legacy_index_error_cannot_commit_partial_document_write() {
886        let dir = tempfile::tempdir().unwrap();
887        let path = dir.path().join("ignored-index-error.sqlite3");
888        let conn = ManagedConnection::open(&path).unwrap();
889        let _catalog = Catalog::open(conn.clone()).unwrap();
890        let backend = SQLiteStorageBackend::new(conn.clone());
891        let observer = conn.new_session();
892        let mut docs = backend.document_store("articles");
893        let mut vectors = backend
894            .vector_index(
895                "articles",
896                "embedding",
897                2,
898                VectorIndexSpec::BruteForce,
899                VectorIndexOpenMode::Create,
900            )
901            .unwrap();
902
903        backend.begin_transaction().unwrap();
904        docs.put(
905            1,
906            BTreeMap::from([("title".to_string(), Value::Str("must roll back".into()))]),
907        )
908        .unwrap();
909        conn.with(|connection| {
910            connection.execute("DROP TABLE _vectors", [])?;
911            Ok(())
912        })
913        .unwrap();
914        // The vector write reports its error directly. Even if a caller
915        // ignores that Result, the managed transaction is poisoned and the
916        // partial document write cannot commit.
917        let ignored = vectors.add(1, vec![1.0, 0.0]);
918        assert!(ignored.is_err());
919        assert!(matches!(
920            backend.commit_transaction(),
921            Err(StorageBackendError::SQLite(
922                SQLiteError::TransactionAborted(_)
923            ))
924        ));
925
926        let stored_docs: i64 = observer
927            .with(|connection| {
928                Ok(connection.query_row(
929                    "SELECT COUNT(*) FROM _documents WHERE table_name = 'articles'",
930                    [],
931                    |row| row.get(0),
932                )?)
933            })
934            .unwrap();
935        let vector_table_exists: i64 = observer
936            .with(|connection| {
937                Ok(connection.query_row(
938                    "SELECT COUNT(*) FROM sqlite_master WHERE type = 'table' AND name = '_vectors'",
939                    [],
940                    |row| row.get(0),
941                )?)
942            })
943            .unwrap();
944        assert_eq!(stored_docs, 0);
945        assert_eq!(vector_table_exists, 1);
946    }
947}
uqa_storage/backend.rs

uqa_storage/
backend.rs
