uqa_storage/mvcc/session/
read.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7use crate::key_value::{KeyValueRead, KeyValueReadRevision};
8use crate::mvcc::{DatabaseId, VersionError};
9use crate::mvcc::{MergedRecordSnapshot, VersionResult};
10use crate::read_control::{KeyReadVisitor, KeyValueReadVisitor, StorageReadControl};
11use crate::{read_control::ValueReadVisitor, StorageBackendResult};
12use uqa_core::memory::BudgetedVec;
13
14/// Borrowed Key/Value adapter over an already captured committed/private record view. It does not open or advance a session.
15pub struct RecordRead<'a> {
16    pub(super) view: &'a MergedRecordSnapshot,
17    pub(super) database: DatabaseId,
18    pub(super) control: &'a StorageReadControl,
19}
20
21impl<'a> RecordRead<'a> {
22    pub fn new(
23        view: &'a MergedRecordSnapshot,
24        database: DatabaseId,
25        control: &'a StorageReadControl,
26    ) -> Self {
27        Self {
28            view,
29            database,
30            control,
31        }
32    }
33}
34
35impl KeyValueRead for RecordRead<'_> {
36    fn control(&self) -> &StorageReadControl {
37        self.control
38    }
39
40    fn marker_fingerprint(&self, key: &[u8]) -> StorageBackendResult<Option<[u8; 32]>> {
41        self.view
42            .marker_fingerprint(self.database, key, self.control)
43            .map(Some)
44    }
45
46    fn revision(&self, prefixes: &[&[u8]]) -> StorageBackendResult<KeyValueReadRevision> {
47        self.control.check()?;
48        let mut private = None;
49        for prefix in prefixes {
50            let mut after = BudgetedVec::new(self.control.memory());
51            loop {
52                let page = self
53                    .view
54                    .private_keys(
55                        prefix,
56                        (!after.is_empty()).then_some(&*after),
57                        64,
58                        self.control,
59                    )
60                    .map_err(VersionError::into_storage_error)?;
61                let Some(last) = page.last() else { break };
62                after.clear();
63                after.extend_from_slice(last.key())?;
64                for record in page.iter() {
65                    private = private.max(Some(record.revision()));
66                }
67            }
68        }
69        Ok(KeyValueReadRevision::records(
70            self.database,
71            self.view.sequence(),
72            private,
73        ))
74    }
75
76    fn record_revision(&self, key: &[u8]) -> StorageBackendResult<Option<KeyValueReadRevision>> {
77        self.view
78            .record_revision(self.database, key, self.control)
79            .map(|revision| revision.map(KeyValueReadRevision::record))
80            .map_err(VersionError::into_storage_error)
81    }
82
83    fn retained_source(
84        &self,
85        key: &[u8],
86    ) -> StorageBackendResult<Option<std::sync::Arc<dyn KeyValueRead + Send + Sync>>> {
87        self.view
88            .retained_source(key, self.control)
89            .map_err(VersionError::into_storage_error)
90    }
91
92    fn retain(
93        &self,
94        _prefixes: &[&[u8]],
95    ) -> StorageBackendResult<std::sync::Arc<dyn KeyValueRead + Send + Sync>> {
96        self.control.check()?;
97        let memory = self
98            .control
99            .memory()
100            .reserve(std::mem::size_of::<RetainedRecordRead>())?;
101        Ok(std::sync::Arc::new(RetainedRecordRead {
102            view: self
103                .view
104                .try_clone()
105                .map_err(VersionError::into_storage_error)?,
106            database: self.database,
107            control: self.control.clone(),
108            _memory: memory,
109        }))
110    }
111
112    fn visit_value(
113        &self,
114        key: &[u8],
115        visit: &mut ValueReadVisitor<'_>,
116    ) -> StorageBackendResult<()> {
117        self.visit_value_budgeted(key, self.control, visit)
118    }
119
120    fn visit_prefix(
121        &self,
122        prefix: &[u8],
123        visit: &mut KeyValueReadVisitor<'_>,
124    ) -> StorageBackendResult<()> {
125        self.visit_prefix_after(prefix, None, usize::MAX, self.control, visit)
126    }
127
128    fn visit_value_budgeted(
129        &self,
130        key: &[u8],
131        control: &StorageReadControl,
132        visit: &mut ValueReadVisitor<'_>,
133    ) -> StorageBackendResult<()> {
134        self.view
135            .visit_value(key, control, &mut |record| {
136                visit(record.and_then(|record| record.value)).map_err(Into::into)
137            })
138            .map_err(VersionError::into_storage_error)
139    }
140
141    fn visit_value_bounded(
142        &self,
143        key: &[u8],
144        max_bytes: usize,
145        control: &StorageReadControl,
146        visit: &mut ValueReadVisitor<'_>,
147    ) -> StorageBackendResult<()> {
148        self.view
149            .visit_value_bounded(key, max_bytes, control, &mut |record| {
150                visit(record.and_then(|record| record.value)).map_err(Into::into)
151            })
152            .map_err(VersionError::into_storage_error)
153    }
154
155    fn visit_prefix_after(
156        &self,
157        prefix: &[u8],
158        after: Option<&[u8]>,
159        limit: usize,
160        control: &StorageReadControl,
161        visit: &mut KeyValueReadVisitor<'_>,
162    ) -> StorageBackendResult<()> {
163        visit_live(self.view, prefix, after, limit, control, visit)
164            .map_err(VersionError::into_storage_error)
165    }
166
167    fn visit_keys_after(
168        &self,
169        prefix: &[u8],
170        after: Option<&[u8]>,
171        limit: usize,
172        control: &StorageReadControl,
173        visit: &mut KeyReadVisitor<'_>,
174    ) -> StorageBackendResult<()> {
175        control.check()?;
176        if limit == 0 {
177            return Ok(());
178        }
179        let mut count = 0;
180        self.view
181            .visit_keys(prefix, after, usize::MAX, control, &mut |key, record| {
182                if record.live {
183                    visit(key)?;
184                    count += 1;
185                }
186                Ok(count < limit)
187            })
188            .map_err(VersionError::into_storage_error)
189    }
190
191    fn contains_prefix_budgeted(
192        &self,
193        prefix: &[u8],
194        control: &StorageReadControl,
195    ) -> StorageBackendResult<bool> {
196        control.check()?;
197        let mut found = false;
198        self.view
199            .visit_keys(prefix, None, usize::MAX, control, &mut |_, record| {
200                found = record.live;
201                Ok(!found)
202            })
203            .map_err(VersionError::into_storage_error)?;
204        Ok(found)
205    }
206}
207
208pub(super) fn visit_live(
209    view: &MergedRecordSnapshot,
210    prefix: &[u8],
211    after: Option<&[u8]>,
212    limit: usize,
213    control: &StorageReadControl,
214    visit: &mut KeyValueReadVisitor<'_>,
215) -> VersionResult<()> {
216    control.cancellation().check()?;
217    if limit == 0 {
218        return Ok(());
219    }
220    let mut count = 0;
221    view.visit_prefix(prefix, after, usize::MAX, control, &mut |key, record| {
222        if let Some(value) = record.value {
223            visit(key, value)?;
224            count += 1;
225        }
226        Ok(count < limit)
227    })
228}
229
230struct RetainedRecordRead {
231    view: MergedRecordSnapshot,
232    database: DatabaseId,
233    control: StorageReadControl,
234    _memory: uqa_core::memory::MemoryReservation,
235}
236impl RetainedRecordRead {
237    fn read(&self) -> RecordRead<'_> {
238        RecordRead {
239            view: &self.view,
240            database: self.database,
241            control: &self.control,
242        }
243    }
244}
245impl KeyValueRead for RetainedRecordRead {
246    fn control(&self) -> &StorageReadControl {
247        &self.control
248    }
249    fn marker_fingerprint(&self, key: &[u8]) -> StorageBackendResult<Option<[u8; 32]>> {
250        self.read().marker_fingerprint(key)
251    }
252    fn revision(&self, prefixes: &[&[u8]]) -> StorageBackendResult<KeyValueReadRevision> {
253        self.read().revision(prefixes)
254    }
255    fn record_revision(&self, key: &[u8]) -> StorageBackendResult<Option<KeyValueReadRevision>> {
256        self.read().record_revision(key)
257    }
258    fn retained_source(
259        &self,
260        key: &[u8],
261    ) -> StorageBackendResult<Option<std::sync::Arc<dyn KeyValueRead + Send + Sync>>> {
262        self.read().retained_source(key)
263    }
264    fn retain(
265        &self,
266        prefixes: &[&[u8]],
267    ) -> StorageBackendResult<std::sync::Arc<dyn KeyValueRead + Send + Sync>> {
268        self.read().retain(prefixes)
269    }
270    fn visit_value(
271        &self,
272        key: &[u8],
273        visit: &mut ValueReadVisitor<'_>,
274    ) -> StorageBackendResult<()> {
275        self.read().visit_value(key, visit)
276    }
277    fn visit_prefix(
278        &self,
279        prefix: &[u8],
280        visit: &mut KeyValueReadVisitor<'_>,
281    ) -> StorageBackendResult<()> {
282        self.read().visit_prefix(prefix, visit)
283    }
284    fn visit_value_budgeted(
285        &self,
286        key: &[u8],
287        control: &StorageReadControl,
288        visit: &mut ValueReadVisitor<'_>,
289    ) -> StorageBackendResult<()> {
290        self.read().visit_value_budgeted(key, control, visit)
291    }
292    fn visit_value_bounded(
293        &self,
294        key: &[u8],
295        max_bytes: usize,
296        control: &StorageReadControl,
297        visit: &mut ValueReadVisitor<'_>,
298    ) -> StorageBackendResult<()> {
299        self.read()
300            .visit_value_bounded(key, max_bytes, control, visit)
301    }
302    fn visit_prefix_after(
303        &self,
304        prefix: &[u8],
305        after: Option<&[u8]>,
306        limit: usize,
307        control: &StorageReadControl,
308        visit: &mut KeyValueReadVisitor<'_>,
309    ) -> StorageBackendResult<()> {
310        self.read()
311            .visit_prefix_after(prefix, after, limit, control, visit)
312    }
313    fn contains_prefix_budgeted(
314        &self,
315        prefix: &[u8],
316        control: &StorageReadControl,
317    ) -> StorageBackendResult<bool> {
318        self.read().contains_prefix_budgeted(prefix, control)
319    }
320
321    fn visit_keys_after(
322        &self,
323        prefix: &[u8],
324        after: Option<&[u8]>,
325        limit: usize,
326        control: &StorageReadControl,
327        visit: &mut KeyReadVisitor<'_>,
328    ) -> StorageBackendResult<()> {
329        self.read()
330            .visit_keys_after(prefix, after, limit, control, visit)
331    }
332}
uqa_storage/mvcc/session/read.rs

uqa_storage/mvcc/session/
read.rs
