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    fn inverted_index(&self, table: &str, analyzer: Analyzer) -> Box<dyn InvertedIndex>;
245
246    /// Upgrade backend-owned inverted-index values before table handles are
247    /// restored. Implementations must make the rewrite atomic and idempotent.
248    fn migrate_inverted_index_storage(&self) -> StorageBackendResult<()> {
249        Ok(())
250    }
251
252    fn vector_index(
253        &self,
254        table: &str,
255        field: &str,
256        dimensions: u32,
257        spec: VectorIndexSpec,
258        mode: VectorIndexOpenMode,
259    ) -> StorageBackendResult<Box<dyn VectorIndex>>;
260
261    fn drop_vector_index_metadata(&self, _table: &str, _field: &str) -> StorageBackendResult<()> {
262        Ok(())
263    }
264
265    /// Whether this backend maps logical `btree` indexes to durable postings.
266    fn persists_btree_indexes(&self) -> bool {
267        false
268    }
269
270    /// `Some(entries)` is a complete persisted index; `None` means it has not
271    /// been built yet and the engine must backfill it from documents once.
272    fn load_btree_index(
273        &self,
274        _table: &str,
275        _field: &str,
276    ) -> StorageBackendResult<Option<Vec<(DocId, Value)>>> {
277        Ok(None)
278    }
279
280    fn btree_index_fields(&self, _table: &str) -> StorageBackendResult<Vec<String>> {
281        Ok(Vec::new())
282    }
283
284    /// Fields whose persisted posting support was found inconsistent during a
285    /// schema migration. The engine repairs these at its explicit open-time
286    /// write boundary and clears each durable retry marker only after success.
287    fn btree_index_repairs(&self) -> StorageBackendResult<Vec<(String, String)>> {
288        Ok(Vec::new())
289    }
290
291    fn clear_btree_index_repair(&self, _table: &str, _field: &str) -> StorageBackendResult<()> {
292        Ok(())
293    }
294
295    fn replace_btree_index(
296        &self,
297        _table: &str,
298        _field: &str,
299        _values: &[(DocId, Value)],
300    ) -> StorageBackendResult<()> {
301        Ok(())
302    }
303
304    /// Repair sparse support differences without requiring capable backends
305    /// to rewrite every already-valid posting. The complete replacement is
306    /// supplied for the storage-neutral fallback.
307    fn repair_btree_index(
308        &self,
309        table: &str,
310        field: &str,
311        complete: &[(DocId, Value)],
312        _stale_doc_ids: &[DocId],
313        _missing: &[(DocId, Value)],
314    ) -> StorageBackendResult<()> {
315        self.replace_btree_index(table, field, complete)
316    }
317
318    /// Replace several complete indexes for one table atomically. Backends
319    /// may override this to share one transaction and prepared statements.
320    fn replace_btree_indexes(
321        &self,
322        table: &str,
323        indexes: &[(&str, &[(DocId, Value)])],
324    ) -> StorageBackendResult<()> {
325        for (field, values) in indexes {
326            self.replace_btree_index(table, field, values)?;
327        }
328        Ok(())
329    }
330
331    fn apply_btree_index_write(
332        &self,
333        _table: &str,
334        _doc_id: DocId,
335        _values: Option<&BTreeMap<String, Value>>,
336    ) -> StorageBackendResult<()> {
337        Ok(())
338    }
339
340    fn drop_btree_index(&self, _table: &str, _field: &str) -> StorageBackendResult<()> {
341        Ok(())
342    }
343
344    fn clear_btree_indexes(&self, _table: &str) -> StorageBackendResult<()> {
345        Ok(())
346    }
347
348    /// 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.
349    fn vacuum(&self) -> StorageBackendResult<()> {
350        Ok(())
351    }
352
353    fn begin_transaction(&self) -> StorageBackendResult<()>;
354
355    /// Begin a transaction whose first operation is expected to be a read.
356    /// Backends with distinct lock modes may defer write-lock acquisition;
357    /// the default preserves existing transaction semantics.
358    fn begin_read_transaction(&self) -> StorageBackendResult<()> {
359        self.begin_transaction()
360    }
361
362    /// Whether this session currently owns a pinned storage transaction.
363    fn in_transaction(&self) -> bool;
364
365    /// Whether the current transaction has performed a physical write.
366    ///
367    /// The engine uses this to enforce read-only statement transactions even
368    /// for writes made through catalog/index helpers it did not classify.
369    fn transaction_has_written(&self) -> StorageBackendResult<bool>;
370
371    /// Backend commit generation visible to this session, when available.
372    /// A changing value invalidates session-local catalog and index caches.
373    fn change_version(&self) -> StorageBackendResult<Option<u64>> {
374        Ok(None)
375    }
376
377    /// Whether reading [`Self::change_version`] can proceed while this
378    /// session owns its write transaction.
379    fn change_version_monitor_is_nonblocking(&self) -> StorageBackendResult<bool> {
380        Ok(true)
381    }
382
383    /// Pin the transaction's read snapshot before cache restoration.
384    fn pin_transaction_snapshot(&self) -> StorageBackendResult<()> {
385        Ok(())
386    }
387
388    fn commit_transaction(&self) -> StorageBackendResult<()>;
389
390    fn rollback_transaction(&self) -> StorageBackendResult<()>;
391
392    fn savepoint(&self, id: StorageSavepointId) -> StorageBackendResult<()>;
393
394    fn release_savepoint(&self, id: StorageSavepointId) -> StorageBackendResult<()>;
395
396    fn rollback_to_savepoint(&self, id: StorageSavepointId) -> StorageBackendResult<()>;
397}
398
399#[derive(Clone)]
400pub struct SQLiteStorageBackend {
401    conn: ManagedConnection,
402}
403
404impl SQLiteStorageBackend {
405    pub fn new(conn: ManagedConnection) -> Self {
406        Self { conn }
407    }
408
409    pub fn connection(&self) -> ManagedConnection {
410        self.conn.clone()
411    }
412
413    /// Create a backend whose stores use an independent transaction session
414    /// over the same physical `SQLite` pool.
415    #[must_use]
416    pub fn new_session(&self) -> Self {
417        Self::new(self.conn.new_session())
418    }
419}
420
421/// Database-level owner that creates isolated `SQLite` engine sessions.
422#[derive(Clone)]
423pub struct SQLiteStorageProvider {
424    connection: ManagedConnection,
425}
426
427impl SQLiteStorageProvider {
428    pub fn new(connection: ManagedConnection) -> Self {
429        Self { connection }
430    }
431}
432
433impl PersistentStorageProvider for SQLiteStorageProvider {
434    fn open_session(&self) -> StorageBackendResult<PersistentStorageSession> {
435        let connection = self.connection.new_session();
436        let catalog: Arc<dyn CatalogFacade> = Arc::new(Catalog::open(connection.clone())?);
437        let backend: Arc<dyn PersistentStorageBackend> =
438            Arc::new(SQLiteStorageBackend::new(connection));
439        Ok(PersistentStorageSession::new(catalog, backend))
440    }
441
442    fn storage_identity(&self) -> StorageBackendResult<Option<PersistentStorageIdentity>> {
443        let Some(path) = self.connection.database_path() else {
444            return Ok(None);
445        };
446        PersistentStorageIdentity::for_database_path(path)
447            .map(Some)
448            .map_err(|error| {
449                StorageBackendError::Other(format!(
450                    "resolve SQLite database identity `{}`: {error}",
451                    path.display()
452                ))
453            })
454    }
455}
456
457impl PersistentStorageBackend for SQLiteStorageBackend {
458    fn storage_identity(&self) -> StorageBackendResult<Option<PersistentStorageIdentity>> {
459        let Some(path) = self.conn.database_path() else {
460            return Ok(None);
461        };
462        PersistentStorageIdentity::for_database_path(path).map(Some)
463    }
464
465    fn open_session(&self) -> StorageBackendResult<PersistentStorageSession> {
466        let connection = self.conn.new_session();
467        let catalog: Arc<dyn CatalogFacade> = Arc::new(Catalog::open(connection.clone())?);
468        let backend: Arc<dyn PersistentStorageBackend> = Arc::new(Self::new(connection));
469        Ok(PersistentStorageSession::new(catalog, backend))
470    }
471
472    fn supports_concurrent_pinned_read_and_write(&self) -> bool {
473        self.conn.supports_concurrent_pinned_read_and_write()
474    }
475
476    fn document_store(&self, table: &str) -> Box<dyn DocumentStore> {
477        Box::new(SQLiteDocumentStore::new(self.conn.clone(), table))
478    }
479
480    fn inverted_index(&self, table: &str, analyzer: Analyzer) -> Box<dyn InvertedIndex> {
481        Box::new(SQLiteInvertedIndex::new(self.conn.clone(), table, analyzer))
482    }
483
484    fn vector_index(
485        &self,
486        table: &str,
487        field: &str,
488        dimensions: u32,
489        spec: VectorIndexSpec,
490        mode: VectorIndexOpenMode,
491    ) -> StorageBackendResult<Box<dyn VectorIndex>> {
492        let index: Box<dyn VectorIndex> = match spec {
493            VectorIndexSpec::BruteForce => Box::new(SQLiteVectorIndex::new(
494                self.conn.clone(),
495                table,
496                field,
497                dimensions,
498            )),
499            VectorIndexSpec::IVF(params) => {
500                params.validate()?;
501                match mode {
502                    VectorIndexOpenMode::Create => Box::new(SQLiteIVFIndex::with_params(
503                        self.conn.clone(),
504                        table,
505                        field,
506                        dimensions,
507                        params.nlist,
508                        params.nprobe,
509                        params.train_threshold,
510                    )),
511                    VectorIndexOpenMode::Restore => Box::new(SQLiteIVFIndex::open_existing(
512                        self.conn.clone(),
513                        table,
514                        field,
515                        dimensions,
516                        params.nlist,
517                        params.nprobe,
518                        params.train_threshold,
519                    )),
520                }
521            }
522            VectorIndexSpec::HNSW(params) => {
523                params.validate()?;
524                match mode {
525                    VectorIndexOpenMode::Create => Box::new(SQLiteHNSWIndex::with_params(
526                        self.conn.clone(),
527                        table,
528                        field,
529                        dimensions,
530                        params,
531                    )),
532                    VectorIndexOpenMode::Restore => {
533                        let index = SQLiteHNSWIndex::open_existing(
534                            self.conn.clone(),
535                            table,
536                            field,
537                            dimensions,
538                            params,
539                        );
540                        index.validate_existing()?;
541                        Box::new(index)
542                    }
543                }
544            }
545        };
546        Ok(index)
547    }
548
549    fn drop_vector_index_metadata(&self, table: &str, field: &str) -> StorageBackendResult<()> {
550        SQLiteIVFIndex::drop_metadata(&self.conn, table, field)?;
551        SQLiteHNSWIndex::drop_metadata(&self.conn, table, field)?;
552        Ok(())
553    }
554
555    fn persists_btree_indexes(&self) -> bool {
556        true
557    }
558
559    fn load_btree_index(
560        &self,
561        table: &str,
562        field: &str,
563    ) -> StorageBackendResult<Option<Vec<(DocId, Value)>>> {
564        Ok(SQLiteBTreeIndexStore::new(self.conn.clone()).load(table, field)?)
565    }
566
567    fn btree_index_fields(&self, table: &str) -> StorageBackendResult<Vec<String>> {
568        Ok(SQLiteBTreeIndexStore::new(self.conn.clone()).fields(table)?)
569    }
570
571    fn btree_index_repairs(&self) -> StorageBackendResult<Vec<(String, String)>> {
572        Ok(SQLiteBTreeIndexStore::new(self.conn.clone()).repairs()?)
573    }
574
575    fn clear_btree_index_repair(&self, table: &str, field: &str) -> StorageBackendResult<()> {
576        SQLiteBTreeIndexStore::new(self.conn.clone()).clear_repair(table, field)?;
577        Ok(())
578    }
579
580    fn replace_btree_index(
581        &self,
582        table: &str,
583        field: &str,
584        values: &[(DocId, Value)],
585    ) -> StorageBackendResult<()> {
586        SQLiteBTreeIndexStore::new(self.conn.clone()).replace(table, field, values)?;
587        Ok(())
588    }
589
590    fn repair_btree_index(
591        &self,
592        table: &str,
593        field: &str,
594        _complete: &[(DocId, Value)],
595        stale_doc_ids: &[DocId],
596        missing: &[(DocId, Value)],
597    ) -> StorageBackendResult<()> {
598        SQLiteBTreeIndexStore::new(self.conn.clone()).repair(
599            table,
600            field,
601            stale_doc_ids,
602            missing,
603        )?;
604        Ok(())
605    }
606
607    fn replace_btree_indexes(
608        &self,
609        table: &str,
610        indexes: &[(&str, &[(DocId, Value)])],
611    ) -> StorageBackendResult<()> {
612        SQLiteBTreeIndexStore::new(self.conn.clone()).replace_many(table, indexes)?;
613        Ok(())
614    }
615
616    fn apply_btree_index_write(
617        &self,
618        table: &str,
619        doc_id: DocId,
620        values: Option<&BTreeMap<String, Value>>,
621    ) -> StorageBackendResult<()> {
622        SQLiteBTreeIndexStore::new(self.conn.clone()).apply_write(table, doc_id, values)?;
623        Ok(())
624    }
625
626    fn drop_btree_index(&self, table: &str, field: &str) -> StorageBackendResult<()> {
627        SQLiteBTreeIndexStore::new(self.conn.clone()).drop_index(table, field)?;
628        Ok(())
629    }
630
631    fn clear_btree_indexes(&self, table: &str) -> StorageBackendResult<()> {
632        SQLiteBTreeIndexStore::new(self.conn.clone()).clear_table(table)?;
633        Ok(())
634    }
635
636    fn vacuum(&self) -> StorageBackendResult<()> {
637        self.conn.vacuum()?;
638        Ok(())
639    }
640
641    fn begin_transaction(&self) -> StorageBackendResult<()> {
642        self.conn.begin_transaction()?;
643        Ok(())
644    }
645
646    fn begin_read_transaction(&self) -> StorageBackendResult<()> {
647        self.conn.begin_deferred_transaction()?;
648        Ok(())
649    }
650
651    fn in_transaction(&self) -> bool {
652        self.conn.in_transaction()
653    }
654
655    fn transaction_has_written(&self) -> StorageBackendResult<bool> {
656        Ok(self.conn.transaction_has_written()?)
657    }
658
659    fn change_version(&self) -> StorageBackendResult<Option<u64>> {
660        Ok(self.conn.data_version()?)
661    }
662
663    fn change_version_monitor_is_nonblocking(&self) -> StorageBackendResult<bool> {
664        Ok(self.conn.data_version_monitor_is_nonblocking()?)
665    }
666
667    fn pin_transaction_snapshot(&self) -> StorageBackendResult<()> {
668        self.conn.pin_transaction_snapshot()?;
669        Ok(())
670    }
671
672    fn commit_transaction(&self) -> StorageBackendResult<()> {
673        self.conn.commit_transaction()?;
674        Ok(())
675    }
676
677    fn rollback_transaction(&self) -> StorageBackendResult<()> {
678        self.conn.rollback_transaction()?;
679        Ok(())
680    }
681
682    fn savepoint(&self, id: StorageSavepointId) -> StorageBackendResult<()> {
683        self.conn.savepoint(&id.backend_name())?;
684        Ok(())
685    }
686
687    fn release_savepoint(&self, id: StorageSavepointId) -> StorageBackendResult<()> {
688        self.conn.release_savepoint(&id.backend_name())?;
689        Ok(())
690    }
691
692    fn rollback_to_savepoint(&self, id: StorageSavepointId) -> StorageBackendResult<()> {
693        self.conn.rollback_to_savepoint(&id.backend_name())?;
694        Ok(())
695    }
696}
697
698#[cfg(test)]
699mod tests {
700    use std::collections::BTreeMap;
701
702    use uqa_analysis::analyzer::standard_analyzer;
703    use uqa_core::Value;
704
705    use super::*;
706    use crate::sqlite::Catalog;
707
708    #[test]
709    fn sqlite_backend_builds_document_index_and_vector_stores() {
710        let conn = ManagedConnection::open_in_memory().unwrap();
711        let _catalog = Catalog::open(conn.clone()).unwrap();
712        let backend = SQLiteStorageBackend::new(conn);
713
714        let mut doc = BTreeMap::new();
715        doc.insert("title".to_string(), Value::Str("rust storage".into()));
716        let mut docs = backend.document_store("articles");
717        docs.put(1, doc).unwrap();
718        assert_eq!(
719            docs.get_field(1, "title").unwrap(),
720            Some(Value::Str("rust storage".into()))
721        );
722
723        let mut inv = backend.inverted_index("articles", standard_analyzer("english"));
724        inv.add_document(
725            1,
726            BTreeMap::from([("title".to_string(), "rust storage".to_string())]),
727        )
728        .unwrap();
729        assert_eq!(inv.doc_freq("title", "rust").unwrap(), 1);
730
731        let mut vectors = backend
732            .vector_index(
733                "articles",
734                "embedding",
735                2,
736                VectorIndexSpec::IVF(crate::IVFIndexParams {
737                    nlist: 2,
738                    nprobe: 1,
739                    train_threshold: 2,
740                }),
741                VectorIndexOpenMode::Create,
742            )
743            .unwrap();
744        vectors.add(1, vec![1.0, 0.0]).unwrap();
745        let hits = vectors.search_knn(&[1.0, 0.0], 1).unwrap();
746        assert_eq!(hits.entries().len(), 1);
747        assert_eq!(hits.entries()[0].doc_id, 1);
748    }
749
750    #[test]
751    fn sqlite_backend_transaction_rolls_back_cross_store_writes() {
752        let conn = ManagedConnection::open_in_memory().unwrap();
753        let _catalog = Catalog::open(conn.clone()).unwrap();
754        let backend = SQLiteStorageBackend::new(conn);
755        let mut docs = backend.document_store("articles");
756        let mut inv = backend.inverted_index("articles", standard_analyzer("english"));
757
758        backend.begin_transaction().unwrap();
759        docs.put(
760            1,
761            BTreeMap::from([("title".to_string(), Value::Str("rollback".into()))]),
762        )
763        .unwrap();
764        inv.add_document(
765            1,
766            BTreeMap::from([("title".to_string(), "rollback".to_string())]),
767        )
768        .unwrap();
769        backend.rollback_transaction().unwrap();
770
771        assert_eq!(docs.len().unwrap(), 0);
772        assert_eq!(inv.doc_freq("title", "rollback").unwrap(), 0);
773    }
774
775    #[test]
776    fn sqlite_sessions_isolate_and_atomically_commit_cross_store_writes() {
777        let dir = tempfile::tempdir().unwrap();
778        let path = dir.path().join("cross-store-isolation.sqlite3");
779        let conn = ManagedConnection::open(&path).unwrap();
780        let catalog = Catalog::open(conn.clone()).unwrap();
781        let writer = SQLiteStorageBackend::new(conn.clone());
782        let observer_conn = conn.new_session();
783        let observer_catalog = Catalog::open(observer_conn.clone()).unwrap();
784        let observer = SQLiteStorageBackend::new(observer_conn);
785
786        let mut writer_docs = writer.document_store("articles");
787        let mut writer_inv = writer.inverted_index("articles", standard_analyzer("english"));
788        let mut writer_vectors = writer
789            .vector_index(
790                "articles",
791                "embedding",
792                2,
793                VectorIndexSpec::BruteForce,
794                VectorIndexOpenMode::Create,
795            )
796            .unwrap();
797        let observer_docs = observer.document_store("articles");
798        let observer_inv = observer.inverted_index("articles", standard_analyzer("english"));
799        let observer_vectors = observer
800            .vector_index(
801                "articles",
802                "embedding",
803                2,
804                VectorIndexSpec::BruteForce,
805                VectorIndexOpenMode::Restore,
806            )
807            .unwrap();
808
809        writer.begin_transaction().unwrap();
810        writer_docs
811            .put(
812                1,
813                BTreeMap::from([("title".to_string(), Value::Str("atomic rust".into()))]),
814            )
815            .unwrap();
816        writer_inv
817            .add_document(
818                1,
819                BTreeMap::from([("title".to_string(), "atomic rust".to_string())]),
820            )
821            .unwrap();
822        writer_vectors.add(1, vec![1.0, 0.0]).unwrap();
823        catalog
824            .save_scoring_params("transactional", r#"{"alpha":1.0}"#)
825            .unwrap();
826
827        assert_eq!(writer_docs.len().unwrap(), 1);
828        assert_eq!(writer_inv.doc_freq("title", "rust").unwrap(), 1);
829        assert_eq!(writer_vectors.count().unwrap(), 1);
830        assert!(catalog
831            .load_scoring_params("transactional")
832            .unwrap()
833            .is_some());
834
835        assert_eq!(observer_docs.len().unwrap(), 0);
836        assert_eq!(observer_inv.doc_freq("title", "rust").unwrap(), 0);
837        assert_eq!(observer_vectors.count().unwrap(), 0);
838        assert!(observer_catalog
839            .load_scoring_params("transactional")
840            .unwrap()
841            .is_none());
842
843        writer.commit_transaction().unwrap();
844        assert_eq!(observer_docs.len().unwrap(), 1);
845        assert_eq!(observer_inv.doc_freq("title", "rust").unwrap(), 1);
846        assert_eq!(observer_vectors.count().unwrap(), 1);
847        assert!(observer_catalog
848            .load_scoring_params("transactional")
849            .unwrap()
850            .is_some());
851    }
852
853    #[test]
854    fn ignored_legacy_index_error_cannot_commit_partial_document_write() {
855        let dir = tempfile::tempdir().unwrap();
856        let path = dir.path().join("ignored-index-error.sqlite3");
857        let conn = ManagedConnection::open(&path).unwrap();
858        let _catalog = Catalog::open(conn.clone()).unwrap();
859        let backend = SQLiteStorageBackend::new(conn.clone());
860        let observer = conn.new_session();
861        let mut docs = backend.document_store("articles");
862        let mut vectors = backend
863            .vector_index(
864                "articles",
865                "embedding",
866                2,
867                VectorIndexSpec::BruteForce,
868                VectorIndexOpenMode::Create,
869            )
870            .unwrap();
871
872        backend.begin_transaction().unwrap();
873        docs.put(
874            1,
875            BTreeMap::from([("title".to_string(), Value::Str("must roll back".into()))]),
876        )
877        .unwrap();
878        conn.with(|connection| {
879            connection.execute("DROP TABLE _vectors", [])?;
880            Ok(())
881        })
882        .unwrap();
883        // The vector write reports its error directly. Even if a caller
884        // ignores that Result, the managed transaction is poisoned and the
885        // partial document write cannot commit.
886        let ignored = vectors.add(1, vec![1.0, 0.0]);
887        assert!(ignored.is_err());
888        assert!(matches!(
889            backend.commit_transaction(),
890            Err(StorageBackendError::SQLite(
891                SQLiteError::TransactionAborted(_)
892            ))
893        ));
894
895        let stored_docs: i64 = observer
896            .with(|connection| {
897                Ok(connection.query_row(
898                    "SELECT COUNT(*) FROM _documents WHERE table_name = 'articles'",
899                    [],
900                    |row| row.get(0),
901                )?)
902            })
903            .unwrap();
904        let vector_table_exists: i64 = observer
905            .with(|connection| {
906                Ok(connection.query_row(
907                    "SELECT COUNT(*) FROM sqlite_master WHERE type = 'table' AND name = '_vectors'",
908                    [],
909                    |row| row.get(0),
910                )?)
911            })
912            .unwrap();
913        assert_eq!(stored_docs, 0);
914        assert_eq!(vector_table_exists, 1);
915    }
916}
uqa_storage/backend.rs

uqa_storage/
backend.rs
