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/// Opaque cache identity. Numeric index revisions alone do not distinguish undo branches or independent private transactions.
184#[derive(Clone)]
185pub struct KeyValueReadRevision(Revision);
186
187#[derive(Clone)]
188enum Revision {
189    Memory(Arc<()>),
190    Record(crate::mvcc::VisibleRecordRevision),
191    Records {
192        database: DatabaseId,
193        committed: CommitSequence,
194        private: Option<PrivateRecordRevision>,
195    },
196}
197
198impl KeyValueReadRevision {
199    pub(crate) fn observed_commit(&self, database: DatabaseId) -> Option<CommitSequence> {
200        match &self.0 {
201            Revision::Record(revision) => revision.committed(database),
202            _ => None,
203        }
204    }
205
206    /// Compare private changes for the same selected prefixes, independently of intervening committed writes. Unversioned or individual-record identities cannot establish this property.
207    pub fn same_private_changes(&self, other: &Self) -> bool {
208        matches!((&self.0, &other.0), (
209            Revision::Records { database: a, private: ap, .. },
210            Revision::Records { database: b, private: bp, .. }
211        ) if a == b && ap == bp)
212    }
213
214    pub(crate) fn record_database(&self) -> Option<DatabaseId> {
215        match &self.0 {
216            Revision::Record(revision) => Some(revision.database()),
217            Revision::Memory(_) | Revision::Records { .. } => None,
218        }
219    }
220
221    /// Whether the selected record prefixes include changes private to this transaction. Unversioned view identities conservatively report private state because they cannot prove committed provenance.
222    pub fn has_private_changes(&self) -> bool {
223        match &self.0 {
224            Revision::Memory(_) => true,
225            Revision::Record(revision) => revision.is_private(),
226            Revision::Records { private, .. } => private.is_some(),
227        }
228    }
229
230    /// 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.
231    pub fn fresh() -> Self {
232        Self(Revision::Memory(Arc::new(())))
233    }
234
235    pub(crate) fn memory(identity: &Arc<()>) -> Self {
236        Self(Revision::Memory(Arc::clone(identity)))
237    }
238
239    pub(crate) fn record(revision: crate::mvcc::VisibleRecordRevision) -> Self {
240        Self(Revision::Record(revision))
241    }
242
243    pub(crate) fn records(
244        database: DatabaseId,
245        committed: CommitSequence,
246        private: Option<PrivateRecordRevision>,
247    ) -> Self {
248        Self(Revision::Records {
249            database,
250            committed,
251            private,
252        })
253    }
254}
255
256impl PartialEq for KeyValueReadRevision {
257    fn eq(&self, other: &Self) -> bool {
258        match (&self.0, &other.0) {
259            (Revision::Memory(a), Revision::Memory(b)) => Arc::ptr_eq(a, b),
260            (Revision::Record(a), Revision::Record(b)) => a == b,
261            (
262                Revision::Records {
263                    database: a,
264                    committed: ac,
265                    private: ap,
266                },
267                Revision::Records {
268                    database: b,
269                    committed: bc,
270                    private: bp,
271                },
272            ) => a == b && ac == bc && ap == bp,
273            _ => false,
274        }
275    }
276}
277impl Eq for KeyValueReadRevision {}
uqa_storage/key_value/view.rs

uqa_storage/key_value/
view.rs
