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

uqa_storage/mvcc/session/
read.rs
