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 KeyValueReadKeyIterator<'a> =
53    dyn Iterator<Item = StorageBackendResult<BudgetedVec<u8>>> + 'a;
54pub type KeyValuePointVisitor<'a> =
55    dyn FnMut(&[u8], Option<&[u8]>) -> StorageBackendResult<bool> + 'a;
56pub type KeyPresenceVisitor<'a> = dyn FnMut(&[u8], bool) -> StorageBackendResult<bool> + 'a;
57pub type KeyValueMutation<'a> =
58    dyn FnMut(&dyn KeyValueRead, &mut dyn KeyValueBatch) -> StorageBackendResult<()> + 'a;
59pub type KeyValueVersionedMutation<'a> = dyn FnMut(
60        crate::mvcc::StorageMutationOrigin,
61        &dyn KeyValueRead,
62        &mut dyn KeyValueBatch,
63    ) -> StorageBackendResult<()>
64    + 'a;
65
66/// A fixed read boundary supplied by the store. Borrowed visitors must not reenter persistence; compound reads make successive calls through this same reader.
67pub trait KeyValueRead {
68    fn control(&self) -> &StorageReadControl;
69    fn revision(&self, prefixes: &[&[u8]]) -> StorageBackendResult<KeyValueReadRevision>;
70    /// 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.
71    fn marker_fingerprint(&self, _key: &[u8]) -> StorageBackendResult<Option<[u8; 32]>> {
72        self.control().check()?;
73        Ok(None)
74    }
75    /// 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.
76    fn record_revision(&self, _key: &[u8]) -> StorageBackendResult<Option<KeyValueReadRevision>> {
77        self.control().check()?;
78        Err(super::codec::other_error(
79            "individual record revisions are not supported",
80        ))
81    }
82    /// 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.
83    fn retained_source(
84        &self,
85        _key: &[u8],
86    ) -> StorageBackendResult<Option<Arc<dyn KeyValueRead + Send + Sync>>> {
87        self.control().check()?;
88        Ok(None)
89    }
90    /// 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.
91    fn retain(
92        &self,
93        prefixes: &[&[u8]],
94    ) -> StorageBackendResult<Arc<dyn KeyValueRead + Send + Sync>> {
95        snapshot::capture(self, prefixes)
96    }
97    fn visit_value(&self, key: &[u8], visit: &mut ValueReadVisitor<'_>)
98        -> StorageBackendResult<()>;
99    fn visit_prefix(
100        &self,
101        prefix: &[u8],
102        visit: &mut KeyValueReadVisitor<'_>,
103    ) -> StorageBackendResult<()>;
104
105    /// Borrow requested values in input order on this boundary. Key production and callbacks are internal and must not reenter persistence. A false visitor or an error stops before producing the next key; defaults require only ordinary point reads.
106    fn visit_values(
107        &self,
108        keys: &mut KeyValueReadKeyIterator<'_>,
109        visit: &mut KeyValuePointVisitor<'_>,
110    ) -> StorageBackendResult<()> {
111        loop {
112            self.control().check()?;
113            let Some(key) = keys.next() else {
114                return Ok(());
115            };
116            let key = key?;
117            let mut more = true;
118            self.visit_value(&key, &mut |value| {
119                more = visit(&key, value)?;
120                Ok(())
121            })?;
122            self.control().check()?;
123            if !more {
124                return Ok(());
125            }
126        }
127    }
128
129    /// Probe requested identities in input order without loading their values. The default uses the existing key-only scan capability; versioned readers reuse private selection cursors.
130    fn visit_key_presence(
131        &self,
132        keys: &mut KeyValueReadKeyIterator<'_>,
133        visit: &mut KeyPresenceVisitor<'_>,
134    ) -> StorageBackendResult<()> {
135        loop {
136            self.control().check()?;
137            let Some(key) = keys.next() else {
138                return Ok(());
139            };
140            let key = key?;
141            let mut present = false;
142            self.visit_keys_after(&key, None, 1, self.control(), &mut |found| {
143                present = found == &*key;
144                Ok(())
145            })?;
146            self.control().check()?;
147            let more = visit(&key, present)?;
148            self.control().check()?;
149            if !more {
150                return Ok(());
151            }
152        }
153    }
154
155    /// Visit one value on this same boundary, charging temporary provider buffers to the supplied query allowance.
156    fn visit_value_budgeted(
157        &self,
158        _key: &[u8],
159        control: &StorageReadControl,
160        _visit: &mut ValueReadVisitor<'_>,
161    ) -> StorageBackendResult<()> {
162        control.check()?;
163        Err(super::codec::other_error(
164            "controlled compound value reads are not supported",
165        ))
166    }
167
168    /// Check a value's encoded size before provider materialization on this same committed/private view.
169    fn visit_value_bounded(
170        &self,
171        _key: &[u8],
172        _max_bytes: usize,
173        control: &StorageReadControl,
174        _visit: &mut ValueReadVisitor<'_>,
175    ) -> StorageBackendResult<()> {
176        control.check()?;
177        Err(super::codec::other_error(
178            "size-bounded compound value reads are not supported",
179        ))
180    }
181
182    /// Visit at most `limit` values in key order, strictly after `after`, without advancing this read boundary.
183    fn visit_prefix_after(
184        &self,
185        _prefix: &[u8],
186        _after: Option<&[u8]>,
187        _limit: usize,
188        control: &StorageReadControl,
189        _visit: &mut KeyValueReadVisitor<'_>,
190    ) -> StorageBackendResult<()> {
191        control.check()?;
192        Err(super::codec::other_error(
193            "controlled compound prefix reads are not supported",
194        ))
195    }
196
197    /// 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.
198    fn visit_keys_after(
199        &self,
200        _prefix: &[u8],
201        _after: Option<&[u8]>,
202        _limit: usize,
203        control: &StorageReadControl,
204        _visit: &mut KeyReadVisitor<'_>,
205    ) -> StorageBackendResult<()> {
206        control.check()?;
207        Err(super::codec::other_error(
208            "compound key-only scans are not supported",
209        ))
210    }
211
212    /// Probe live keys on this boundary without materializing their values.
213    fn contains_prefix_budgeted(
214        &self,
215        _prefix: &[u8],
216        control: &StorageReadControl,
217    ) -> StorageBackendResult<bool> {
218        control.check()?;
219        Err(super::codec::other_error(
220            "compound key-only probes are not supported",
221        ))
222    }
223
224    fn get(&self, key: &[u8]) -> StorageBackendResult<Option<BudgetedVec<u8>>> {
225        let mut result = None;
226        self.visit_value(key, &mut |value| {
227            if let Some(value) = value {
228                let mut bytes = BudgetedVec::new(self.control().memory());
229                bytes.extend_from_slice(value)?;
230                result = Some(bytes);
231            }
232            Ok(())
233        })?;
234        Ok(result)
235    }
236}
237
238#[cfg(test)]
239mod tests;
240
241/// Opaque cache identity. Numeric index revisions alone do not distinguish undo branches or independent private transactions.
242#[derive(Clone)]
243pub struct KeyValueReadRevision(Revision);
244
245#[derive(Clone)]
246enum Revision {
247    Memory(Arc<()>),
248    Record(crate::mvcc::VisibleRecordRevision),
249    Records {
250        database: DatabaseId,
251        committed: CommitSequence,
252        private: Option<PrivateRecordRevision>,
253    },
254}
255
256impl KeyValueReadRevision {
257    pub(crate) fn observed_commit(&self, database: DatabaseId) -> Option<CommitSequence> {
258        match &self.0 {
259            Revision::Record(revision) => revision.committed(database),
260            _ => None,
261        }
262    }
263
264    /// Compare private changes for the same selected prefixes, independently of intervening committed writes. Unversioned or individual-record identities cannot establish this property.
265    pub fn same_private_changes(&self, other: &Self) -> bool {
266        matches!((&self.0, &other.0), (
267            Revision::Records { database: a, private: ap, .. },
268            Revision::Records { database: b, private: bp, .. }
269        ) if a == b && ap == bp)
270    }
271
272    /// 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.
273    pub fn follows_by_one_commit(&self, earlier: &Self) -> bool {
274        match (earlier.committed_state(), self.committed_state()) {
275            (Some((database, before, _)), Some((current, after, false))) => {
276                database == current && before.successor().ok() == Some(after)
277            }
278            _ => false,
279        }
280    }
281
282    /// 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.
283    pub fn same_committed_state(&self, other: &Self) -> bool {
284        match (self.committed_state(), other.committed_state()) {
285            (Some((database, sequence, _)), Some((other_database, other_sequence, _))) => {
286                database == other_database && sequence == other_sequence
287            }
288            _ => false,
289        }
290    }
291
292    /// The database, committed sequence and presence of private changes of a record view.
293    fn committed_state(&self) -> Option<(DatabaseId, CommitSequence, bool)> {
294        match &self.0 {
295            Revision::Record(record) => record
296                .committed_sequence()
297                .map(|sequence| (record.database(), sequence, record.is_private())),
298            Revision::Records {
299                database,
300                committed,
301                private,
302            } => Some((*database, *committed, private.is_some())),
303            Revision::Memory(_) => None,
304        }
305    }
306
307    pub(crate) fn record_database(&self) -> Option<DatabaseId> {
308        match &self.0 {
309            Revision::Record(revision) => Some(revision.database()),
310            Revision::Memory(_) | Revision::Records { .. } => None,
311        }
312    }
313
314    /// Whether the selected record prefixes include changes private to this transaction. Unversioned view identities conservatively report private state because they cannot prove committed provenance.
315    pub fn has_private_changes(&self) -> bool {
316        match &self.0 {
317            Revision::Memory(_) => true,
318            Revision::Record(revision) => revision.is_private(),
319            Revision::Records { private, .. } => private.is_some(),
320        }
321    }
322
323    /// 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.
324    pub fn fresh() -> Self {
325        Self(Revision::Memory(Arc::new(())))
326    }
327
328    pub(crate) fn memory(identity: &Arc<()>) -> Self {
329        Self(Revision::Memory(Arc::clone(identity)))
330    }
331
332    pub(crate) fn record(revision: crate::mvcc::VisibleRecordRevision) -> Self {
333        Self(Revision::Record(revision))
334    }
335
336    pub(crate) fn records(
337        database: DatabaseId,
338        committed: CommitSequence,
339        private: Option<PrivateRecordRevision>,
340    ) -> Self {
341        Self(Revision::Records {
342            database,
343            committed,
344            private,
345        })
346    }
347}
348
349impl PartialEq for KeyValueReadRevision {
350    fn eq(&self, other: &Self) -> bool {
351        match (&self.0, &other.0) {
352            (Revision::Memory(a), Revision::Memory(b)) => Arc::ptr_eq(a, b),
353            (Revision::Record(a), Revision::Record(b)) => a == b,
354            (
355                Revision::Records {
356                    database: a,
357                    committed: ac,
358                    private: ap,
359                },
360                Revision::Records {
361                    database: b,
362                    committed: bc,
363                    private: bp,
364                },
365            ) => a == b && ac == bc && ap == bp,
366            _ => false,
367        }
368    }
369}
370impl Eq for KeyValueReadRevision {}
uqa_storage/key_value/view.rs

uqa_storage/key_value/
view.rs
