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

uqa_storage/
backend.rs
