uqa_storage/mvcc/
view.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! Provider-independent reads combine one pinned committed boundary with a fixed private command view.
8
9use std::cmp::Ordering;
10use std::sync::Arc;
11
12use uqa_core::memory::{BudgetedVec, MemoryReservation};
13
14use crate::read_control::StorageReadControl;
15
16use super::{
17    CommitSequence, PreparedRecordWrite, PrivateRecordSnapshot, RecordVersion, ScannedRecord,
18    SharedRecordValue, VersionResult,
19};
20
21#[derive(Clone, Copy)]
22pub struct BorrowedRecord<'a> {
23    pub revision: Option<CommitSequence>,
24    pub value: Option<&'a [u8]>,
25}
26
27/// Visible revision metadata, including tombstones, without retaining the encoded payload.
28#[derive(Clone, Copy, Debug, PartialEq, Eq)]
29pub struct RecordMetadata {
30    pub revision: Option<CommitSequence>,
31    pub live: bool,
32}
33
34impl From<BorrowedRecord<'_>> for RecordMetadata {
35    fn from(record: BorrowedRecord<'_>) -> Self {
36        Self {
37            revision: record.revision,
38            live: record.value.is_some(),
39        }
40    }
41}
42
43pub type RecordValueVisitor<'a> = dyn FnMut(Option<BorrowedRecord<'_>>) -> VersionResult<()> + 'a;
44pub type RecordScanVisitor<'a> = dyn FnMut(&[u8], BorrowedRecord<'_>) -> VersionResult<bool> + 'a;
45pub type RecordKeyVisitor<'a> = dyn FnMut(&[u8], RecordMetadata) -> VersionResult<bool> + 'a;
46
47/// A retained committed view whose lease protects visible versions until this owner is dropped. Provider read windows must finish inside each call, without retaining a physical writer or calling user code.
48pub trait CommittedRecordSnapshot: Send + Sync {
49    fn sequence(&self) -> CommitSequence;
50
51    /// Read a revision and tombstone marker without materializing its value when the provider supports key-only access.
52    fn metadata(
53        &self,
54        key: &[u8],
55        control: &StorageReadControl,
56    ) -> VersionResult<Option<RecordMetadata>> {
57        let mut found = None;
58        self.visit_value(key, control, &mut |record| {
59            found = record.map(Into::into);
60            Ok(())
61        })?;
62        Ok(found)
63    }
64
65    /// Visit ordered record identities, including tombstones, without fetching values from providers with key-only access.
66    fn visit_keys(
67        &self,
68        prefix: &[u8],
69        after: Option<&[u8]>,
70        limit: usize,
71        control: &StorageReadControl,
72        visit: &mut RecordKeyVisitor<'_>,
73    ) -> VersionResult<()> {
74        self.visit_prefix(prefix, after, limit, control, &mut |key, record| {
75            visit(key, record.into())
76        })
77    }
78
79    /// Return the newest revision at or before this snapshot, including tombstones. Missing identities return `None`.
80    fn get(
81        &self,
82        key: &[u8],
83        control: &StorageReadControl,
84    ) -> VersionResult<Option<RecordVersion<SharedRecordValue>>>;
85
86    /// Return at most `limit` unique keys in ascending order, strictly after `after`, including tombstones. A short page means that the prefix is exhausted at this snapshot; all pages keep the same boundary.
87    fn scan(
88        &self,
89        prefix: &[u8],
90        after: Option<&[u8]>,
91        limit: usize,
92        control: &StorageReadControl,
93    ) -> VersionResult<BudgetedVec<ScannedRecord>>;
94
95    /// Borrow provider-owned bytes for an internal storage visitor. Visitors must not reenter this snapshot or execute SQL/user callbacks.
96    fn visit_value(
97        &self,
98        key: &[u8],
99        control: &StorageReadControl,
100        visit: &mut RecordValueVisitor<'_>,
101    ) -> VersionResult<()> {
102        let record = self.get(key, control)?;
103        visit(record.as_ref().map(|record| BorrowedRecord {
104            revision: Some(record.sequence()),
105            value: record.value().map(|value| &***value),
106        }))?;
107        control.cancellation().check()?;
108        Ok(())
109    }
110
111    /// Visit ordered versions until the limit, exhaustion or a visitor returning `false`. Implementations with borrowed pages avoid charging their encoded payloads to the caller's decode allowance.
112    fn visit_prefix(
113        &self,
114        prefix: &[u8],
115        after: Option<&[u8]>,
116        limit: usize,
117        control: &StorageReadControl,
118        visit: &mut RecordScanVisitor<'_>,
119    ) -> VersionResult<()> {
120        control.cancellation().check()?;
121        let mut cursor: Option<BudgetedVec<u8>> = None;
122        let mut remaining = limit;
123        while remaining != 0 {
124            let mut page = self.scan(
125                prefix,
126                cursor.as_deref().or(after),
127                remaining.min(64),
128                control,
129            )?;
130            if page.is_empty() {
131                break;
132            }
133            for record in page.iter() {
134                control.cancellation().check()?;
135                if !visit(
136                    &record.key,
137                    BorrowedRecord {
138                        revision: Some(record.version.sequence()),
139                        value: record.version.value().map(|value| &***value),
140                    },
141                )? {
142                    control.cancellation().check()?;
143                    return Ok(());
144                }
145                remaining -= 1;
146            }
147            cursor = page.pop().map(|record| record.key);
148        }
149        control.cancellation().check()?;
150        Ok(())
151    }
152}
153
154struct RetainedSnapshot<T> {
155    snapshot: T,
156    _memory: MemoryReservation,
157}
158
159/// Charge a provider snapshot's retained metadata before allocating its shared owner. The provider snapshot itself owns any version-retention lease required by its format.
160pub fn retain_record_snapshot<T: CommittedRecordSnapshot + 'static>(
161    snapshot: T,
162    control: &StorageReadControl,
163) -> VersionResult<Arc<dyn CommittedRecordSnapshot>> {
164    control.cancellation().check()?;
165    let memory = control
166        .memory()
167        .reserve(std::mem::size_of::<RetainedSnapshot<T>>())?;
168    Ok(Arc::new(RetainedSnapshot {
169        snapshot,
170        _memory: memory,
171    }))
172}
173
174impl<T: CommittedRecordSnapshot> CommittedRecordSnapshot for RetainedSnapshot<T> {
175    fn sequence(&self) -> CommitSequence {
176        self.snapshot.sequence()
177    }
178    fn metadata(
179        &self,
180        key: &[u8],
181        control: &StorageReadControl,
182    ) -> VersionResult<Option<RecordMetadata>> {
183        self.snapshot.metadata(key, control)
184    }
185    fn visit_keys(
186        &self,
187        prefix: &[u8],
188        after: Option<&[u8]>,
189        limit: usize,
190        control: &StorageReadControl,
191        visit: &mut RecordKeyVisitor<'_>,
192    ) -> VersionResult<()> {
193        self.snapshot
194            .visit_keys(prefix, after, limit, control, visit)
195    }
196    fn get(
197        &self,
198        key: &[u8],
199        control: &StorageReadControl,
200    ) -> VersionResult<Option<RecordVersion<SharedRecordValue>>> {
201        self.snapshot.get(key, control)
202    }
203    fn scan(
204        &self,
205        prefix: &[u8],
206        after: Option<&[u8]>,
207        limit: usize,
208        control: &StorageReadControl,
209    ) -> VersionResult<BudgetedVec<ScannedRecord>> {
210        self.snapshot.scan(prefix, after, limit, control)
211    }
212    fn visit_value(
213        &self,
214        key: &[u8],
215        control: &StorageReadControl,
216        visit: &mut RecordValueVisitor<'_>,
217    ) -> VersionResult<()> {
218        self.snapshot.visit_value(key, control, visit)
219    }
220    fn visit_prefix(
221        &self,
222        prefix: &[u8],
223        after: Option<&[u8]>,
224        limit: usize,
225        control: &StorageReadControl,
226        visit: &mut RecordScanVisitor<'_>,
227    ) -> VersionResult<()> {
228        self.snapshot
229            .visit_prefix(prefix, after, limit, control, visit)
230    }
231}
232
233/// Visible payload plus its original committed precondition. Private deletions remain observable as tombstones rather than falling back to the committed value.
234pub struct VisibleRecord {
235    original: Option<CommitSequence>,
236    value: Option<SharedRecordValue>,
237    private: bool,
238}
239
240impl VisibleRecord {
241    pub fn original_revision(&self) -> Option<CommitSequence> {
242        self.original
243    }
244
245    pub fn value(&self) -> Option<&[u8]> {
246        self.value.as_ref().map(|value| &***value)
247    }
248
249    pub fn is_private(&self) -> bool {
250        self.private
251    }
252
253    fn committed(version: RecordVersion<SharedRecordValue>) -> Self {
254        let (sequence, value) = version.into_parts();
255        Self {
256            original: Some(sequence),
257            value,
258            private: false,
259        }
260    }
261
262    fn private(write: &PreparedRecordWrite) -> Self {
263        Self {
264            original: write.expected(),
265            value: write.shared_value(),
266            private: true,
267        }
268    }
269}
270
271pub struct ScannedVisibleRecord {
272    pub key: BudgetedVec<u8>,
273    pub record: VisibleRecord,
274}
275
276/// Retain both boundaries across reads and paging. The transaction owner chooses when a later SQL command acquires a new committed snapshot; this reader never advances either boundary implicitly.
277pub struct MergedRecordSnapshot {
278    committed: Arc<dyn CommittedRecordSnapshot>,
279    private: PrivateRecordSnapshot,
280    serializable: Option<super::SerializableReadContext>,
281}
282
283impl MergedRecordSnapshot {
284    /// Share the committed owner and retain the original private revision under its existing allowance.
285    pub fn try_clone(&self) -> VersionResult<Self> {
286        Ok(Self {
287            committed: Arc::clone(&self.committed),
288            private: self.private.try_clone()?,
289            serializable: self.serializable.clone(),
290        })
291    }
292
293    pub fn new(
294        committed: Arc<dyn CommittedRecordSnapshot>,
295        private: PrivateRecordSnapshot,
296    ) -> Self {
297        Self {
298            committed,
299            private,
300            serializable: None,
301        }
302    }
303
304    pub(crate) fn with_serializable(
305        mut self,
306        context: Option<super::SerializableReadContext>,
307    ) -> Self {
308        self.serializable = context;
309        self
310    }
311
312    /// Original logical reader attribution. Clones retain the same participant and fixed data boundaries.
313    pub fn serializable(&self) -> Option<&super::SerializableReadContext> {
314        self.serializable.as_ref()
315    }
316
317    pub fn sequence(&self) -> CommitSequence {
318        self.committed.sequence()
319    }
320
321    /// The pinned committed boundary underlying this command view, without private replacements.
322    pub fn committed(&self) -> &dyn CommittedRecordSnapshot {
323        self.committed.as_ref()
324    }
325
326    pub(super) fn retain_committed(&self) -> Arc<dyn CommittedRecordSnapshot> {
327        Arc::clone(&self.committed)
328    }
329
330    /// Bounded private key metadata for provider-derived cache invalidation; values are never materialized.
331    pub fn private_keys(
332        &self,
333        prefix: &[u8],
334        after: Option<&[u8]>,
335        limit: usize,
336        control: &StorageReadControl,
337    ) -> VersionResult<BudgetedVec<super::PrivateRecordKey>> {
338        self.private.scan_keys(prefix, after, limit, control)
339    }
340
341    pub fn visit_value(
342        &self,
343        key: &[u8],
344        control: &StorageReadControl,
345        visit: &mut RecordValueVisitor<'_>,
346    ) -> VersionResult<()> {
347        if let Some(write) = self.private.get(key, control)? {
348            visit(Some(BorrowedRecord {
349                revision: write.expected(),
350                value: write.value(),
351            }))?;
352            control.cancellation().check()?;
353            return Ok(());
354        }
355        self.committed.visit_value(key, control, visit)
356    }
357
358    pub fn visit_prefix(
359        &self,
360        prefix: &[u8],
361        after: Option<&[u8]>,
362        limit: usize,
363        control: &StorageReadControl,
364        visit: &mut RecordScanVisitor<'_>,
365    ) -> VersionResult<()> {
366        self.private.visit_merged::<super::projection::Values>(
367            &*self.committed,
368            prefix,
369            after,
370            limit,
371            control,
372            visit,
373        )
374    }
375
376    pub fn metadata(
377        &self,
378        key: &[u8],
379        control: &StorageReadControl,
380    ) -> VersionResult<Option<RecordMetadata>> {
381        if let Some(write) = self.private.get(key, control)? {
382            return Ok(Some(RecordMetadata {
383                revision: write.expected(),
384                live: write.value().is_some(),
385            }));
386        }
387        self.committed.metadata(key, control)
388    }
389
390    pub fn visit_keys(
391        &self,
392        prefix: &[u8],
393        after: Option<&[u8]>,
394        limit: usize,
395        control: &StorageReadControl,
396        visit: &mut RecordKeyVisitor<'_>,
397    ) -> VersionResult<()> {
398        self.private.visit_merged::<super::projection::Keys>(
399            &*self.committed,
400            prefix,
401            after,
402            limit,
403            control,
404            visit,
405        )
406    }
407
408    pub fn get(
409        &self,
410        key: &[u8],
411        control: &StorageReadControl,
412    ) -> VersionResult<Option<VisibleRecord>> {
413        if let Some(write) = self.private.get(key, control)? {
414            return Ok(Some(VisibleRecord::private(&write)));
415        }
416        Ok(self
417            .committed
418            .get(key, control)?
419            .map(VisibleRecord::committed))
420    }
421
422    /// Merge bounded pages without duplicate identities or payload copies. Tombstones are returned so callers can preserve revision observations even when no row is visible.
423    pub fn scan(
424        &self,
425        prefix: &[u8],
426        after: Option<&[u8]>,
427        limit: usize,
428        control: &StorageReadControl,
429    ) -> VersionResult<BudgetedVec<ScannedVisibleRecord>> {
430        control.cancellation().check()?;
431        let mut result = BudgetedVec::new(control.memory());
432        if limit == 0 {
433            return Ok(result);
434        }
435        let (committed, _committed_memory) = self
436            .committed
437            .scan(prefix, after, limit, control)?
438            .into_parts();
439        let (private, _private_memory) = self
440            .private
441            .scan(prefix, after, limit, control)?
442            .into_parts();
443        let mut committed = committed.into_iter().peekable();
444        let mut private = private.into_iter().peekable();
445        while result.len() < limit {
446            control.cancellation().check()?;
447            let order = match (committed.peek(), private.peek()) {
448                (Some(committed), Some(private)) => committed.key.as_ref().cmp(private.key()),
449                (Some(_), None) => Ordering::Less,
450                (None, Some(_)) => Ordering::Greater,
451                (None, None) => break,
452            };
453            let record = match order {
454                Ordering::Less => {
455                    let committed = committed.next().expect("peeked committed record");
456                    ScannedVisibleRecord {
457                        key: committed.key,
458                        record: VisibleRecord::committed(committed.version),
459                    }
460                }
461                Ordering::Equal => {
462                    let committed = committed.next().expect("peeked committed record");
463                    let private = private.next().expect("peeked private record");
464                    ScannedVisibleRecord {
465                        key: committed.key,
466                        record: VisibleRecord::private(&private),
467                    }
468                }
469                Ordering::Greater => {
470                    let private = private.next().expect("peeked private record");
471                    let mut key = BudgetedVec::new(control.memory());
472                    key.extend_from_slice(private.key())?;
473                    ScannedVisibleRecord {
474                        key,
475                        record: VisibleRecord::private(&private),
476                    }
477                }
478            };
479            result.push(record)?;
480        }
481        Ok(result)
482    }
483}
uqa_storage/mvcc/view.rs

uqa_storage/mvcc/
view.rs
