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::vector_index::{VectorIndex, VectorIndexOpenMode, VectorIndexSpec};
24use crate::CatalogFacade;
25
26#[derive(Debug, thiserror::Error)]
27pub enum StorageBackendError {
28    #[error(transparent)]
29    Memory(#[from] uqa_core::memory::MemoryError),
30    #[error(transparent)]
31    Cancelled(#[from] uqa_core::QueryCancelled),
32    #[error("text analysis failed: {0}")]
33    Analysis(#[from] uqa_analysis::AnalysisError),
34    #[error("payload serialization failed: {0}")]
35    Serde(#[from] serde_json::Error),
36    #[error("{backend} storage failed: {source}")]
37    Backend {
38        backend: &'static str,
39        #[source]
40        source: Box<dyn std::error::Error + Send + Sync>,
41    },
42    #[error("{0}")]
43    Other(String),
44}
45
46impl StorageBackendError {
47    pub fn backend(
48        backend: &'static str,
49        source: impl std::error::Error + Send + Sync + 'static,
50    ) -> Self {
51        Self::Backend {
52            backend,
53            source: Box::new(source),
54        }
55    }
56}
57
58pub type StorageBackendResult<T> = std::result::Result<T, StorageBackendError>;
59
60/// Opaque transaction checkpoint identity. SQL savepoint names remain engine metadata and are never forwarded into a backend namespace.
61#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
62pub struct StorageSavepointId(u64);
63
64impl StorageSavepointId {
65    #[must_use]
66    pub fn allocate() -> Self {
67        static NEXT_ID: AtomicU64 = AtomicU64::new(1);
68        let id = NEXT_ID
69            .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| {
70                current.checked_add(1)
71            })
72            .expect("storage savepoint identity space exhausted");
73        Self(id)
74    }
75
76    pub fn backend_name(self) -> String {
77        self.0.to_string()
78    }
79}
80
81/// Session-bound catalog and physical storage handles created together.
82///
83/// Both handles must share the same transaction context. Keeping their
84/// construction behind one provider prevents a catalog write from escaping
85/// through a different connection or transaction than document/index writes.
86pub struct PersistentStorageSession {
87    pub catalog: Arc<dyn CatalogFacade>,
88    pub backend: Arc<dyn PersistentStorageBackend>,
89}
90
91/// 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.
92#[derive(Clone, Debug, PartialEq, Eq, Hash)]
93pub enum PersistentStorageIdentity {
94    File(PathBuf),
95    Opaque(String),
96}
97
98impl PersistentStorageIdentity {
99    /// 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.
100    pub fn for_database_path(path: &Path) -> StorageBackendResult<Self> {
101        let path = resolve_final_symlinks(path)?;
102        match std::fs::canonicalize(&path) {
103            Ok(canonical) => Ok(Self::File(canonical)),
104            Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
105                let file_name = path.file_name().ok_or_else(|| {
106                    StorageBackendError::Other(format!(
107                        "database path `{}` has no file name",
108                        path.display()
109                    ))
110                })?;
111                let parent = path
112                    .parent()
113                    .filter(|parent| !parent.as_os_str().is_empty())
114                    .map_or_else(|| PathBuf::from("."), Path::to_path_buf);
115                let parent = std::fs::canonicalize(&parent).map_err(|error| {
116                    StorageBackendError::Other(format!(
117                        "canonicalize database directory `{}`: {error}",
118                        parent.display()
119                    ))
120                })?;
121                Ok(Self::File(parent.join(file_name)))
122            }
123            Err(error) => Err(StorageBackendError::Other(format!(
124                "canonicalize database `{}`: {error}",
125                path.display()
126            ))),
127        }
128    }
129}
130
131fn resolve_final_symlinks(path: &Path) -> StorageBackendResult<PathBuf> {
132    let mut current = path.to_path_buf();
133    let mut followed = 0usize;
134    loop {
135        match std::fs::symlink_metadata(&current) {
136            Ok(metadata) if metadata.file_type().is_symlink() => {
137                followed += 1;
138                if followed > 40 {
139                    return Err(StorageBackendError::Other(format!(
140                        "database path `{}` has too many symbolic-link levels",
141                        path.display()
142                    )));
143                }
144                let target = std::fs::read_link(&current).map_err(|error| {
145                    StorageBackendError::Other(format!(
146                        "read database symbolic link `{}`: {error}",
147                        current.display()
148                    ))
149                })?;
150                current = if target.is_absolute() {
151                    target
152                } else {
153                    current
154                        .parent()
155                        .filter(|parent| !parent.as_os_str().is_empty())
156                        .unwrap_or_else(|| Path::new("."))
157                        .join(target)
158                };
159            }
160            Ok(_) => return Ok(current),
161            Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(current),
162            Err(error) => {
163                return Err(StorageBackendError::Other(format!(
164                    "inspect database path `{}`: {error}",
165                    current.display()
166                )))
167            }
168        }
169    }
170}
171
172impl PersistentStorageSession {
173    pub fn new(
174        catalog: Arc<dyn CatalogFacade>,
175        backend: Arc<dyn PersistentStorageBackend>,
176    ) -> Self {
177        Self { catalog, backend }
178    }
179}
180
181/// Factory for independent sessions over one durable database.
182///
183/// A provider owns the database-level resource while each returned session
184/// owns its transaction state. This is the engine-facing extension point for
185/// `SQLite`, redb, and application-defined Key/Value stores.
186pub trait PersistentStorageProvider: Send + Sync {
187    fn open_session(&self) -> StorageBackendResult<PersistentStorageSession>;
188
189    /// Open handles for initial Engine restoration. Providers may defer catalog schema preparation until `CatalogFacade::initialize_storage` runs inside the owning transaction; ordinary session factories must return an initialized catalog.
190    fn open_initial_session(&self) -> StorageBackendResult<PersistentStorageSession> {
191        self.open_session()
192    }
193
194    /// 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.
195    fn storage_identity(&self) -> StorageBackendResult<Option<PersistentStorageIdentity>> {
196        Ok(None)
197    }
198}
199
200/// Factory plus transaction surface for persistent table/index storage.
201pub trait PersistentStorageBackend: Send + Sync {
202    /// Return the stable database identity for independently constructed engines over this backend. File identities also enable cross-process row-lock coordination.
203    fn storage_identity(&self) -> StorageBackendResult<Option<PersistentStorageIdentity>> {
204        Ok(None)
205    }
206
207    /// 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.
208    fn open_session(&self) -> StorageBackendResult<PersistentStorageSession> {
209        Err(StorageBackendError::Other(
210            "independent sessions are not implemented for this persistent backend".into(),
211        ))
212    }
213
214    /// 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.
215    fn supports_concurrent_pinned_read_and_write(&self) -> bool {
216        false
217    }
218
219    fn document_store(&self, table: &str) -> Box<dyn DocumentStore>;
220
221    /// Upgrade backend-owned document records before table handles are restored. Implementations must make the rewrite atomic and idempotent.
222    fn migrate_document_storage(&self) -> StorageBackendResult<()> {
223        Ok(())
224    }
225
226    fn inverted_index(&self, table: &str, analyzer: Analyzer) -> Box<dyn InvertedIndex>;
227
228    /// Upgrade backend-owned inverted-index values before table handles are
229    /// restored. Implementations must make the rewrite atomic and idempotent.
230    fn migrate_inverted_index_storage(&self) -> StorageBackendResult<()> {
231        Ok(())
232    }
233
234    fn vector_index(
235        &self,
236        table: &str,
237        field: &str,
238        dimensions: u32,
239        spec: VectorIndexSpec,
240        mode: VectorIndexOpenMode,
241    ) -> StorageBackendResult<Box<dyn VectorIndex>>;
242
243    fn drop_vector_index_metadata(&self, _table: &str, _field: &str) -> StorageBackendResult<()> {
244        Ok(())
245    }
246
247    /// Whether this backend maps logical `btree` indexes to durable postings.
248    fn persists_btree_indexes(&self) -> bool {
249        false
250    }
251
252    /// `Some(entries)` is a complete persisted index; `None` means it has not
253    /// been built yet and the engine must backfill it from documents once.
254    fn load_btree_index(
255        &self,
256        _table: &str,
257        _field: &crate::ValueIndexKey,
258    ) -> StorageBackendResult<Option<Vec<(DocId, Value)>>> {
259        Ok(None)
260    }
261
262    fn btree_index_fields(&self, _table: &str) -> StorageBackendResult<Vec<crate::ValueIndexKey>> {
263        Ok(Vec::new())
264    }
265
266    /// Fields whose persisted posting support was found inconsistent during a
267    /// schema migration. The engine repairs these at its explicit open-time
268    /// write boundary and clears each durable retry marker only after success.
269    fn btree_index_repairs(&self) -> StorageBackendResult<Vec<(String, crate::ValueIndexKey)>> {
270        Ok(Vec::new())
271    }
272
273    fn clear_btree_index_repair(
274        &self,
275        _table: &str,
276        _field: &crate::ValueIndexKey,
277    ) -> StorageBackendResult<()> {
278        Ok(())
279    }
280
281    fn replace_btree_index(
282        &self,
283        _table: &str,
284        _field: &crate::ValueIndexKey,
285        _values: &[(DocId, Value)],
286    ) -> StorageBackendResult<()> {
287        Ok(())
288    }
289
290    /// Repair sparse support differences without requiring capable backends
291    /// to rewrite every already-valid posting. The complete replacement is
292    /// supplied for the storage-neutral fallback.
293    fn repair_btree_index(
294        &self,
295        table: &str,
296        field: &crate::ValueIndexKey,
297        complete: &[(DocId, Value)],
298        _stale_doc_ids: &[DocId],
299        _missing: &[(DocId, Value)],
300    ) -> StorageBackendResult<()> {
301        self.replace_btree_index(table, field, complete)
302    }
303
304    /// Replace several complete indexes for one table atomically. Backends
305    /// may override this to share one transaction and prepared statements.
306    fn replace_btree_indexes(
307        &self,
308        table: &str,
309        indexes: &[(&crate::ValueIndexKey, &[(DocId, Value)])],
310    ) -> StorageBackendResult<()> {
311        for (field, values) in indexes {
312            self.replace_btree_index(table, field, values)?;
313        }
314        Ok(())
315    }
316
317    fn apply_btree_index_write(
318        &self,
319        _table: &str,
320        _doc_id: DocId,
321        _values: Option<&BTreeMap<crate::ValueIndexKey, Value>>,
322    ) -> StorageBackendResult<()> {
323        Ok(())
324    }
325
326    fn drop_btree_index(
327        &self,
328        _table: &str,
329        _field: &crate::ValueIndexKey,
330    ) -> StorageBackendResult<()> {
331        Ok(())
332    }
333
334    fn clear_btree_indexes(&self, _table: &str) -> StorageBackendResult<()> {
335        Ok(())
336    }
337
338    /// 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.
339    fn vacuum(&self) -> StorageBackendResult<()> {
340        Ok(())
341    }
342
343    fn begin_transaction(&self) -> StorageBackendResult<()>;
344
345    /// Begin a transaction whose first operation is expected to be a read.
346    /// Backends with distinct lock modes may defer write-lock acquisition;
347    /// the default preserves existing transaction semantics.
348    fn begin_read_transaction(&self) -> StorageBackendResult<()> {
349        self.begin_transaction()
350    }
351
352    /// 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.
353    fn begin_upgradeable_transaction(&self) -> StorageBackendResult<()> {
354        self.begin_transaction()
355    }
356
357    /// Whether this session currently owns a pinned storage transaction.
358    fn in_transaction(&self) -> bool;
359
360    /// Whether the current transaction has performed a physical write.
361    ///
362    /// The engine uses this to enforce read-only statement transactions even
363    /// for writes made through catalog/index helpers it did not classify.
364    fn transaction_has_written(&self) -> StorageBackendResult<bool>;
365
366    /// Backend commit generation visible to this session, when available.
367    /// A changing value invalidates session-local catalog and index caches.
368    fn change_version(&self) -> StorageBackendResult<Option<u64>> {
369        Ok(None)
370    }
371
372    /// Whether reading [`Self::change_version`] can proceed while this session owns its pinned transaction. An independent monitor can also be unsafe for a reader when a pending writer is waiting for that reader's lock.
373    fn change_version_monitor_is_nonblocking(&self) -> StorageBackendResult<bool> {
374        Ok(true)
375    }
376
377    /// Pin the transaction's read snapshot before cache restoration.
378    fn pin_transaction_snapshot(&self) -> StorageBackendResult<()> {
379        Ok(())
380    }
381
382    fn commit_transaction(&self) -> StorageBackendResult<()>;
383
384    fn rollback_transaction(&self) -> StorageBackendResult<()>;
385
386    fn savepoint(&self, id: StorageSavepointId) -> StorageBackendResult<()>;
387
388    fn release_savepoint(&self, id: StorageSavepointId) -> StorageBackendResult<()>;
389
390    fn rollback_to_savepoint(&self, id: StorageSavepointId) -> StorageBackendResult<()>;
391}
392
393#[cfg(test)]
394mod identity_tests {
395    use super::*;
396
397    #[cfg(unix)]
398    #[test]
399    fn dangling_database_symlink_keeps_the_target_identity_after_creation() {
400        use std::os::unix::fs::symlink;
401
402        let directory = tempfile::tempdir().unwrap();
403        let target = directory.path().join("target.db");
404        let link = directory.path().join("database.db");
405        symlink("target.db", &link).unwrap();
406
407        let before = PersistentStorageIdentity::for_database_path(&link).unwrap();
408        std::fs::File::create(&target).unwrap();
409        let after = PersistentStorageIdentity::for_database_path(&link).unwrap();
410
411        assert_eq!(before, after);
412        assert_eq!(
413            after,
414            PersistentStorageIdentity::File(target.canonicalize().unwrap())
415        );
416    }
417}
uqa_storage/backend.rs

uqa_storage/
backend.rs
