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;
54
55/// A fixed read boundary supplied by the store. Borrowed visitors must not reenter persistence; compound reads make successive calls through this same reader.
56pub trait KeyValueRead {
57    fn control(&self) -> &StorageReadControl;
58    fn revision(&self, prefixes: &[&[u8]]) -> StorageBackendResult<KeyValueReadRevision>;
59    /// 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.
60    fn retain(
61        &self,
62        prefixes: &[&[u8]],
63    ) -> StorageBackendResult<Arc<dyn KeyValueRead + Send + Sync>> {
64        snapshot::capture(self, prefixes)
65    }
66    fn visit_value(&self, key: &[u8], visit: &mut ValueReadVisitor<'_>)
67        -> StorageBackendResult<()>;
68    fn visit_prefix(
69        &self,
70        prefix: &[u8],
71        visit: &mut KeyValueReadVisitor<'_>,
72    ) -> StorageBackendResult<()>;
73
74    /// Visit one value on this same boundary, charging temporary provider buffers to the supplied query allowance.
75    fn visit_value_budgeted(
76        &self,
77        _key: &[u8],
78        control: &StorageReadControl,
79        _visit: &mut ValueReadVisitor<'_>,
80    ) -> StorageBackendResult<()> {
81        control.check()?;
82        Err(super::codec::other_error(
83            "controlled compound value reads are not supported",
84        ))
85    }
86
87    /// Visit at most `limit` values in key order, strictly after `after`, without advancing this read boundary.
88    fn visit_prefix_after(
89        &self,
90        _prefix: &[u8],
91        _after: Option<&[u8]>,
92        _limit: usize,
93        control: &StorageReadControl,
94        _visit: &mut KeyValueReadVisitor<'_>,
95    ) -> StorageBackendResult<()> {
96        control.check()?;
97        Err(super::codec::other_error(
98            "controlled compound prefix reads are not supported",
99        ))
100    }
101
102    /// 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.
103    fn visit_keys_after(
104        &self,
105        _prefix: &[u8],
106        _after: Option<&[u8]>,
107        _limit: usize,
108        control: &StorageReadControl,
109        _visit: &mut KeyReadVisitor<'_>,
110    ) -> StorageBackendResult<()> {
111        control.check()?;
112        Err(super::codec::other_error(
113            "compound key-only scans are not supported",
114        ))
115    }
116
117    /// Probe live keys on this boundary without materializing their values.
118    fn contains_prefix_budgeted(
119        &self,
120        _prefix: &[u8],
121        control: &StorageReadControl,
122    ) -> StorageBackendResult<bool> {
123        control.check()?;
124        Err(super::codec::other_error(
125            "compound key-only probes are not supported",
126        ))
127    }
128
129    fn get(&self, key: &[u8]) -> StorageBackendResult<Option<BudgetedVec<u8>>> {
130        let mut result = None;
131        self.visit_value(key, &mut |value| {
132            if let Some(value) = value {
133                let mut bytes = BudgetedVec::new(self.control().memory());
134                bytes.extend_from_slice(value)?;
135                result = Some(bytes);
136            }
137            Ok(())
138        })?;
139        Ok(result)
140    }
141}
142
143/// Opaque cache identity. Numeric index revisions alone do not distinguish undo branches or independent private transactions.
144#[derive(Clone)]
145pub struct KeyValueReadRevision(Revision);
146
147#[derive(Clone)]
148enum Revision {
149    Memory(Arc<()>),
150    Records {
151        database: DatabaseId,
152        committed: CommitSequence,
153        private: Option<PrivateRecordRevision>,
154    },
155}
156
157impl KeyValueReadRevision {
158    /// Whether the selected record prefixes include changes private to this transaction. Unversioned view identities conservatively report private state because they cannot prove committed provenance.
159    pub fn has_private_changes(&self) -> bool {
160        match &self.0 {
161            Revision::Memory(_) => true,
162            Revision::Records { private, .. } => private.is_some(),
163        }
164    }
165
166    /// 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.
167    pub fn fresh() -> Self {
168        Self(Revision::Memory(Arc::new(())))
169    }
170
171    pub(crate) fn memory(identity: &Arc<()>) -> Self {
172        Self(Revision::Memory(Arc::clone(identity)))
173    }
174
175    pub(crate) fn records(
176        database: DatabaseId,
177        committed: CommitSequence,
178        private: Option<PrivateRecordRevision>,
179    ) -> Self {
180        Self(Revision::Records {
181            database,
182            committed,
183            private,
184        })
185    }
186}
187
188impl PartialEq for KeyValueReadRevision {
189    fn eq(&self, other: &Self) -> bool {
190        match (&self.0, &other.0) {
191            (Revision::Memory(a), Revision::Memory(b)) => Arc::ptr_eq(a, b),
192            (
193                Revision::Records {
194                    database: a,
195                    committed: ac,
196                    private: ap,
197                },
198                Revision::Records {
199                    database: b,
200                    committed: bc,
201                    private: bp,
202                },
203            ) => a == b && ac == bc && ap == bp,
204            _ => false,
205        }
206    }
207}
208impl Eq for KeyValueReadRevision {}
uqa_storage/key_value/view.rs

uqa_storage/key_value/
view.rs
