uqa_storage/key_value/
view.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! Compound readers and evaluated mutations share one provider-owned visibility boundary.
8
9use std::sync::Arc;
10
11use uqa_core::memory::BudgetedVec;
12
13use crate::mvcc::{CommitSequence, DatabaseId, PrivateRecordRevision};
14use crate::read_control::{
15    KeyReadVisitor, KeyValueReadVisitor, StorageReadControl, ValueReadVisitor,
16};
17use crate::{KeyValueBatch, StorageBackendResult};
18
19mod snapshot;
20
21/// Copy a bounded key page before calling code that may read this same boundary again. Provider visitors can hold physical locks while lending their key bytes.
22pub(crate) fn for_each_key(
23    read: &dyn KeyValueRead,
24    prefix: &[u8],
25    visit: &mut dyn FnMut(&[u8]) -> StorageBackendResult<bool>,
26) -> StorageBackendResult<()> {
27    let mut after = None::<BudgetedVec<u8>>;
28    loop {
29        let mut keys = BudgetedVec::new(read.control().memory());
30        read.visit_keys_after(prefix, after.as_deref(), 128, read.control(), &mut |key| {
31            let mut owned = BudgetedVec::new(read.control().memory());
32            owned.extend_from_slice(key)?;
33            keys.push(owned)?;
34            Ok(())
35        })?;
36        let Some(last) = keys.last() else {
37            return Ok(());
38        };
39        let mut next = BudgetedVec::new(read.control().memory());
40        next.extend_from_slice(last)?;
41        after = Some(next);
42        for key in keys.iter() {
43            read.control().check()?;
44            if !visit(key)? {
45                return Ok(());
46            }
47        }
48    }
49}
50
51pub type KeyValueReadScope<'a> = dyn FnMut(&dyn KeyValueRead) -> StorageBackendResult<()> + 'a;
52pub type KeyValueMutation<'a> =
53    dyn FnMut(&dyn KeyValueRead, &mut dyn KeyValueBatch) -> StorageBackendResult<()> + 'a;
54pub type KeyValueVersionedMutation<'a> = dyn FnMut(
55        crate::mvcc::StorageMutationOrigin,
56        &dyn KeyValueRead,
57        &mut dyn KeyValueBatch,
58    ) -> StorageBackendResult<()>
59    + 'a;
60
61/// A fixed read boundary supplied by the store. Borrowed visitors must not reenter persistence; compound reads make successive calls through this same reader.
62pub trait KeyValueRead {
63    fn control(&self) -> &StorageReadControl;
64    fn revision(&self, prefixes: &[&[u8]]) -> StorageBackendResult<KeyValueReadRevision>;
65    /// Versioned identity of both committed and private inputs to one shared change marker. Unlike a final record revision, a private marker cannot hide a changed committed input. Unsupported providers return None.
66    fn marker_fingerprint(&self, _key: &[u8]) -> StorageBackendResult<Option<[u8; 32]>> {
67        self.control().check()?;
68        Ok(None)
69    }
70    /// Identity of one live record on this fixed view. Unlike a view revision, unrelated commits do not change it. Compare identities only for the same logical key; missing records and tombstones return `None`. Providers without exact committed/private record provenance must reject this capability.
71    fn record_revision(&self, _key: &[u8]) -> StorageBackendResult<Option<KeyValueReadRevision>> {
72        self.control().check()?;
73        Err(super::codec::other_error(
74            "individual record revisions are not supported",
75        ))
76    }
77    /// Read source attached to this exact private metadata replacement. Absence never authorizes opening a newer snapshot. Committed values have no attachment; provider wrappers translate the metadata key but preserve the source's own key space and controls.
78    fn retained_source(
79        &self,
80        _key: &[u8],
81    ) -> StorageBackendResult<Option<Arc<dyn KeyValueRead + Send + Sync>>> {
82        self.control().check()?;
83        Ok(None)
84    }
85    /// Retain this committed/private boundary after the callback returns. Callers may only read the selected prefixes. The default copies selected bytes under this reader's allowance; versioned providers retain their existing visibility owners without loading values.
86    fn retain(
87        &self,
88        prefixes: &[&[u8]],
89    ) -> StorageBackendResult<Arc<dyn KeyValueRead + Send + Sync>> {
90        snapshot::capture(self, prefixes)
91    }
92    fn visit_value(&self, key: &[u8], visit: &mut ValueReadVisitor<'_>)
93        -> StorageBackendResult<()>;
94    fn visit_prefix(
95        &self,
96        prefix: &[u8],
97        visit: &mut KeyValueReadVisitor<'_>,
98    ) -> StorageBackendResult<()>;
99
100    /// Visit one value on this same boundary, charging temporary provider buffers to the supplied query allowance.
101    fn visit_value_budgeted(
102        &self,
103        _key: &[u8],
104        control: &StorageReadControl,
105        _visit: &mut ValueReadVisitor<'_>,
106    ) -> StorageBackendResult<()> {
107        control.check()?;
108        Err(super::codec::other_error(
109            "controlled compound value reads are not supported",
110        ))
111    }
112
113    /// Check a value's encoded size before provider materialization on this same committed/private view.
114    fn visit_value_bounded(
115        &self,
116        _key: &[u8],
117        _max_bytes: usize,
118        control: &StorageReadControl,
119        _visit: &mut ValueReadVisitor<'_>,
120    ) -> StorageBackendResult<()> {
121        control.check()?;
122        Err(super::codec::other_error(
123            "size-bounded compound value reads are not supported",
124        ))
125    }
126
127    /// Visit at most `limit` values in key order, strictly after `after`, without advancing this read boundary.
128    fn visit_prefix_after(
129        &self,
130        _prefix: &[u8],
131        _after: Option<&[u8]>,
132        _limit: usize,
133        control: &StorageReadControl,
134        _visit: &mut KeyValueReadVisitor<'_>,
135    ) -> StorageBackendResult<()> {
136        control.check()?;
137        Err(super::codec::other_error(
138            "controlled compound prefix reads are not supported",
139        ))
140    }
141
142    /// Visit at most `limit` live keys in ascending order, strictly after `after`, without reading their values or advancing this boundary. Capable wrappers must preserve this key-only contract.
143    fn visit_keys_after(
144        &self,
145        _prefix: &[u8],
146        _after: Option<&[u8]>,
147        _limit: usize,
148        control: &StorageReadControl,
149        _visit: &mut KeyReadVisitor<'_>,
150    ) -> StorageBackendResult<()> {
151        control.check()?;
152        Err(super::codec::other_error(
153            "compound key-only scans are not supported",
154        ))
155    }
156
157    /// Probe live keys on this boundary without materializing their values.
158    fn contains_prefix_budgeted(
159        &self,
160        _prefix: &[u8],
161        control: &StorageReadControl,
162    ) -> StorageBackendResult<bool> {
163        control.check()?;
164        Err(super::codec::other_error(
165            "compound key-only probes are not supported",
166        ))
167    }
168
169    fn get(&self, key: &[u8]) -> StorageBackendResult<Option<BudgetedVec<u8>>> {
170        let mut result = None;
171        self.visit_value(key, &mut |value| {
172            if let Some(value) = value {
173                let mut bytes = BudgetedVec::new(self.control().memory());
174                bytes.extend_from_slice(value)?;
175                result = Some(bytes);
176            }
177            Ok(())
178        })?;
179        Ok(result)
180    }
181}
182
183#[cfg(test)]
184mod tests;
185
186/// Opaque cache identity. Numeric index revisions alone do not distinguish undo branches or independent private transactions.
187#[derive(Clone)]
188pub struct KeyValueReadRevision(Revision);
189
190#[derive(Clone)]
191enum Revision {
192    Memory(Arc<()>),
193    Record(crate::mvcc::VisibleRecordRevision),
194    Records {
195        database: DatabaseId,
196        committed: CommitSequence,
197        private: Option<PrivateRecordRevision>,
198    },
199}
200
201impl KeyValueReadRevision {
202    pub(crate) fn observed_commit(&self, database: DatabaseId) -> Option<CommitSequence> {
203        match &self.0 {
204            Revision::Record(revision) => revision.committed(database),
205            _ => None,
206        }
207    }
208
209    /// Compare private changes for the same selected prefixes, independently of intervening committed writes. Unversioned or individual-record identities cannot establish this property.
210    pub fn same_private_changes(&self, other: &Self) -> bool {
211        matches!((&self.0, &other.0), (
212            Revision::Records { database: a, private: ap, .. },
213            Revision::Records { database: b, private: bp, .. }
214        ) if a == b && ap == bp)
215    }
216
217    /// Whether this committed view follows the committed state of `earlier` by exactly one record commit of the same database. `earlier` may include changes private to the transaction that observed it; this view may not. A session whose transaction wrote records after observing `earlier`, and then observes this view after committing, knows that its own commit is the only one between them.
218    pub fn follows_by_one_commit(&self, earlier: &Self) -> bool {
219        match (earlier.committed_state(), self.committed_state()) {
220            (Some((database, before, _)), Some((current, after, false))) => {
221                database == current && before.successor().ok() == Some(after)
222            }
223            _ => false,
224        }
225    }
226
227    /// Whether both views include the same committed record state of one database, whatever private changes accompany either. Differences between them are then private to the observing transaction.
228    pub fn same_committed_state(&self, other: &Self) -> bool {
229        match (self.committed_state(), other.committed_state()) {
230            (Some((database, sequence, _)), Some((other_database, other_sequence, _))) => {
231                database == other_database && sequence == other_sequence
232            }
233            _ => false,
234        }
235    }
236
237    /// The database, committed sequence and presence of private changes of a record view.
238    fn committed_state(&self) -> Option<(DatabaseId, CommitSequence, bool)> {
239        match &self.0 {
240            Revision::Record(record) => record
241                .committed_sequence()
242                .map(|sequence| (record.database(), sequence, record.is_private())),
243            Revision::Records {
244                database,
245                committed,
246                private,
247            } => Some((*database, *committed, private.is_some())),
248            Revision::Memory(_) => None,
249        }
250    }
251
252    pub(crate) fn record_database(&self) -> Option<DatabaseId> {
253        match &self.0 {
254            Revision::Record(revision) => Some(revision.database()),
255            Revision::Memory(_) | Revision::Records { .. } => None,
256        }
257    }
258
259    /// Whether the selected record prefixes include changes private to this transaction. Unversioned view identities conservatively report private state because they cannot prove committed provenance.
260    pub fn has_private_changes(&self) -> bool {
261        match &self.0 {
262            Revision::Memory(_) => true,
263            Revision::Record(revision) => revision.is_private(),
264            Revision::Records { private, .. } => private.is_some(),
265        }
266    }
267
268    /// Allocate a distinct identity for a provider-owned view. Retain and clone it while that view is unchanged; allocate a fresh identity after writes or undo instead of reusing a numeric counter.
269    pub fn fresh() -> Self {
270        Self(Revision::Memory(Arc::new(())))
271    }
272
273    pub(crate) fn memory(identity: &Arc<()>) -> Self {
274        Self(Revision::Memory(Arc::clone(identity)))
275    }
276
277    pub(crate) fn record(revision: crate::mvcc::VisibleRecordRevision) -> Self {
278        Self(Revision::Record(revision))
279    }
280
281    pub(crate) fn records(
282        database: DatabaseId,
283        committed: CommitSequence,
284        private: Option<PrivateRecordRevision>,
285    ) -> Self {
286        Self(Revision::Records {
287            database,
288            committed,
289            private,
290        })
291    }
292}
293
294impl PartialEq for KeyValueReadRevision {
295    fn eq(&self, other: &Self) -> bool {
296        match (&self.0, &other.0) {
297            (Revision::Memory(a), Revision::Memory(b)) => Arc::ptr_eq(a, b),
298            (Revision::Record(a), Revision::Record(b)) => a == b,
299            (
300                Revision::Records {
301                    database: a,
302                    committed: ac,
303                    private: ap,
304                },
305                Revision::Records {
306                    database: b,
307                    committed: bc,
308                    private: bp,
309                },
310            ) => a == b && ac == bc && ap == bp,
311            _ => false,
312        }
313    }
314}
315impl Eq for KeyValueReadRevision {}
uqa_storage/key_value/view.rs

uqa_storage/key_value/
view.rs
