uqa_storage/key_value/
storage_backend.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! Persistent-storage facade that assembles the key/value adapters.
8
9use super::{
10    btree_index, Analyzer, Arc, BTreeMap, DocId, DocumentStore, InvertedIndex, KeyValueCatalog,
11    KeyValueDocumentStore, KeyValueHNSWIndex, KeyValueIVFIndex, KeyValueInvertedIndex,
12    KeyValueStore, KeyValueVectorIndex, PersistentStorageBackend, PersistentStorageIdentity,
13    StorageBackendResult, Value, VectorIndex, VectorIndexOpenMode, VectorIndexSpec,
14};
15use crate::{CatalogFacade, PersistentStorageSession};
16
17/// Persistent storage factory implemented over [`KeyValueStore`].
18#[derive(Clone)]
19pub struct KeyValueStorageBackend {
20    store: Arc<dyn KeyValueStore>,
21}
22
23impl KeyValueStorageBackend {
24    pub fn new(store: Arc<dyn KeyValueStore>) -> Self {
25        Self { store }
26    }
27
28    pub fn store(&self) -> Arc<dyn KeyValueStore> {
29        Arc::clone(&self.store)
30    }
31}
32
33impl PersistentStorageBackend for KeyValueStorageBackend {
34    fn auxiliary_encryption_key(&self) -> Option<crate::StorageEncryptionKey> {
35        self.store.auxiliary_encryption_key()
36    }
37
38    fn retention_control(&self) -> Option<crate::read_control::StorageReadControl> {
39        self.store.retention_control()
40    }
41
42    fn notification_publications(
43        &self,
44    ) -> Option<&dyn crate::notifications::NotificationPublicationStore> {
45        self.store.notification_publications()
46    }
47
48    fn serializable_session(&self) -> Option<&dyn crate::mvcc::SerializableSession> {
49        self.store.serializable_session()
50    }
51
52    fn vacuum(&self) -> StorageBackendResult<()> {
53        self.store.vacuum()
54    }
55
56    fn write_cancellation(&self) -> Option<uqa_core::CancellationToken> {
57        self.store.write_cancellation()
58    }
59
60    fn open_session_with_cancellation(
61        &self,
62        cancellation: &uqa_core::CancellationToken,
63    ) -> StorageBackendResult<PersistentStorageSession> {
64        let store = self.store.open_session_with_cancellation(cancellation)?;
65        let catalog: Arc<dyn CatalogFacade> = Arc::new(KeyValueCatalog::new(Arc::clone(&store)));
66        let backend: Arc<dyn PersistentStorageBackend> = Arc::new(Self::new(store));
67        Ok(PersistentStorageSession::new(catalog, backend))
68    }
69
70    fn open_retained_read_session(
71        &self,
72        cancellation: &uqa_core::CancellationToken,
73    ) -> StorageBackendResult<PersistentStorageSession> {
74        let store = self.store.open_retained_read_session(cancellation)?;
75        let catalog: Arc<dyn CatalogFacade> = Arc::new(KeyValueCatalog::new(Arc::clone(&store)));
76        let backend: Arc<dyn PersistentStorageBackend> = Arc::new(Self::new(store));
77        Ok(PersistentStorageSession::new(catalog, backend))
78    }
79
80    fn transaction_model(&self) -> crate::StorageTransactionModel {
81        self.store.transaction_model()
82    }
83
84    fn supports_concurrent_pinned_read_and_write(&self) -> bool {
85        self.store.transaction_model().is_versioned()
86    }
87
88    fn identifier_allocator(&self) -> Option<&dyn crate::mvcc::IdentifierAllocator> {
89        self.store.identifier_allocator()
90    }
91
92    fn transaction_affinity(&self) -> Option<crate::StorageSessionAffinity> {
93        self.store.transaction_affinity()
94    }
95
96    fn storage_identity(&self) -> StorageBackendResult<Option<PersistentStorageIdentity>> {
97        self.store.storage_identity()
98    }
99
100    fn open_session(&self) -> StorageBackendResult<PersistentStorageSession> {
101        let store = self.store.open_session()?;
102        let catalog: Arc<dyn CatalogFacade> = Arc::new(KeyValueCatalog::new(Arc::clone(&store)));
103        let backend: Arc<dyn PersistentStorageBackend> = Arc::new(Self::new(store));
104        Ok(PersistentStorageSession::new(catalog, backend))
105    }
106
107    fn document_store(&self, table: &str) -> Box<dyn DocumentStore> {
108        Box::new(KeyValueDocumentStore::new(Arc::clone(&self.store), table))
109    }
110
111    fn migrate_document_storage(&self) -> StorageBackendResult<()> {
112        KeyValueDocumentStore::migrate_legacy_storage(self.store.as_ref())
113    }
114
115    fn inverted_index(&self, table: &str, analyzer: Analyzer) -> Box<dyn InvertedIndex> {
116        Box::new(KeyValueInvertedIndex::new(
117            Arc::clone(&self.store),
118            table,
119            analyzer,
120        ))
121    }
122
123    fn migrate_inverted_index_storage(&self) -> StorageBackendResult<()> {
124        KeyValueInvertedIndex::migrate_legacy_storage(self.store.as_ref())
125    }
126
127    fn vector_index(
128        &self,
129        table: &str,
130        field: &str,
131        dimensions: u32,
132        spec: VectorIndexSpec,
133        mode: VectorIndexOpenMode,
134    ) -> StorageBackendResult<Box<dyn VectorIndex>> {
135        match spec {
136            VectorIndexSpec::BruteForce => Ok(Box::new(KeyValueVectorIndex::new(
137                Arc::clone(&self.store),
138                table,
139                field,
140                dimensions,
141            ))),
142            VectorIndexSpec::IVF(params) => match mode {
143                VectorIndexOpenMode::Create => Ok(Box::new(KeyValueIVFIndex::create(
144                    Arc::clone(&self.store),
145                    table,
146                    field,
147                    dimensions,
148                    params,
149                )?)),
150                VectorIndexOpenMode::Restore => Ok(Box::new(KeyValueIVFIndex::restore(
151                    Arc::clone(&self.store),
152                    table,
153                    field,
154                    dimensions,
155                    params,
156                )?)),
157            },
158            VectorIndexSpec::HNSW(params) => match mode {
159                VectorIndexOpenMode::Create => Ok(Box::new(KeyValueHNSWIndex::create(
160                    Arc::clone(&self.store),
161                    table,
162                    field,
163                    dimensions,
164                    params,
165                )?)),
166                VectorIndexOpenMode::Restore => Ok(Box::new(KeyValueHNSWIndex::restore(
167                    Arc::clone(&self.store),
168                    table,
169                    field,
170                    dimensions,
171                    params,
172                )?)),
173            },
174        }
175    }
176
177    fn drop_vector_index_metadata(&self, table: &str, field: &str) -> StorageBackendResult<()> {
178        KeyValueIVFIndex::drop_metadata(self.store.as_ref(), table, field)
179    }
180
181    fn persists_btree_indexes(&self) -> bool {
182        true
183    }
184
185    fn load_btree_index(
186        &self,
187        table: &str,
188        field: &crate::ValueIndexKey,
189    ) -> StorageBackendResult<Option<Vec<(DocId, Value)>>> {
190        btree_index::load(self.store.as_ref(), table, field)
191    }
192
193    fn btree_index_fields(&self, table: &str) -> StorageBackendResult<Vec<crate::ValueIndexKey>> {
194        btree_index::fields(self.store.as_ref(), table)
195    }
196
197    fn read_btree_index_entry(
198        &self,
199        table: &str,
200        field: &crate::ValueIndexKey,
201        doc_id: DocId,
202    ) -> StorageBackendResult<crate::ValueIndexEntry> {
203        btree_index::read_entry(self.store.as_ref(), table, field, doc_id)
204    }
205
206    fn replace_btree_index(
207        &self,
208        table: &str,
209        field: &crate::ValueIndexKey,
210        values: &[(DocId, Value)],
211    ) -> StorageBackendResult<()> {
212        btree_index::replace(self.store.as_ref(), table, field, values)
213    }
214
215    fn replace_btree_indexes(
216        &self,
217        table: &str,
218        indexes: &[(&crate::ValueIndexKey, &[(DocId, Value)])],
219    ) -> StorageBackendResult<()> {
220        btree_index::replace_many(self.store.as_ref(), table, indexes)
221    }
222
223    fn apply_btree_index_write(
224        &self,
225        table: &str,
226        doc_id: DocId,
227        values: Option<&BTreeMap<crate::ValueIndexKey, Value>>,
228    ) -> StorageBackendResult<()> {
229        btree_index::apply_write(self.store.as_ref(), table, doc_id, values)
230    }
231
232    fn drop_btree_index(
233        &self,
234        table: &str,
235        field: &crate::ValueIndexKey,
236    ) -> StorageBackendResult<()> {
237        btree_index::drop_index(self.store.as_ref(), table, field)
238    }
239
240    fn clear_btree_indexes(&self, table: &str) -> StorageBackendResult<()> {
241        btree_index::clear_entries(self.store.as_ref(), table)
242    }
243
244    fn begin_transaction(&self) -> StorageBackendResult<()> {
245        self.store.begin_transaction()
246    }
247
248    fn begin_read_transaction(&self) -> StorageBackendResult<()> {
249        // The backend contract is a read-first hint; the byte-store's read transaction is strictly read-only.
250        self.store.begin_upgradeable_transaction()
251    }
252
253    fn begin_upgradeable_transaction(&self) -> StorageBackendResult<()> {
254        self.store.begin_upgradeable_transaction()
255    }
256
257    fn refresh_transaction_snapshot(
258        &self,
259        cancellation: &uqa_core::CancellationToken,
260    ) -> StorageBackendResult<()> {
261        self.store.refresh_transaction_snapshot(cancellation)
262    }
263
264    fn in_transaction(&self) -> bool {
265        self.store.in_transaction()
266    }
267
268    fn transaction_has_written(&self) -> StorageBackendResult<bool> {
269        self.store.transaction_has_written()
270    }
271
272    fn change_version(&self) -> StorageBackendResult<Option<u64>> {
273        self.store.change_version()
274    }
275
276    fn change_version_monitor_is_nonblocking(&self) -> StorageBackendResult<bool> {
277        self.store.change_version_monitor_is_nonblocking()
278    }
279
280    fn pin_transaction_snapshot(&self) -> StorageBackendResult<()> {
281        self.store.pin_transaction_snapshot()
282    }
283
284    fn commit_transaction(&self) -> StorageBackendResult<()> {
285        self.store.commit_transaction()
286    }
287
288    fn rollback_transaction(&self) -> StorageBackendResult<()> {
289        self.store.rollback_transaction()
290    }
291
292    fn savepoint(&self, id: crate::StorageSavepointId) -> StorageBackendResult<()> {
293        self.store.savepoint(&id.backend_name())
294    }
295
296    fn release_savepoint(&self, id: crate::StorageSavepointId) -> StorageBackendResult<()> {
297        self.store.release_savepoint(&id.backend_name())
298    }
299
300    fn rollback_to_savepoint(&self, id: crate::StorageSavepointId) -> StorageBackendResult<()> {
301        self.store.rollback_to_savepoint(&id.backend_name())
302    }
303}
uqa_storage/key_value/storage_backend.rs

uqa_storage/key_value/
storage_backend.rs
