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

uqa_storage/mvcc/session/
read.rs
