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
9mod fingerprint;
10
11use std::cmp::Ordering;
12use std::sync::Arc;
13
14use uqa_core::memory::{BudgetedVec, MemoryReservation};
15
16use crate::read_control::StorageReadControl;
17
18use super::{
19    CommitSequence, PreparedRecordWrite, PrivateRecordSnapshot, RecordVersion, ScannedRecord,
20    SharedRecordValue, VersionResult,
21};
22
23#[derive(Clone, Copy)]
24pub struct BorrowedRecord<'a> {
25    pub revision: Option<CommitSequence>,
26    pub value: Option<&'a [u8]>,
27}
28
29/// Visible revision metadata, including tombstones, without retaining the encoded payload.
30#[derive(Clone, Copy, Debug, PartialEq, Eq)]
31pub struct RecordMetadata {
32    pub revision: Option<CommitSequence>,
33    pub live: bool,
34}
35
36/// Exact identity of a live record on one committed/private view. Compare only for the same physical key; history and undo-branch identities distinguish equal payloads that were replaced.
37#[derive(Clone, Copy, Debug, PartialEq, Eq)]
38pub struct VisibleRecordRevision {
39    database: super::DatabaseId,
40    committed: Option<CommitSequence>,
41    private: Option<super::PrivateRecordRevision>,
42}
43
44impl VisibleRecordRevision {
45    pub(crate) fn committed(self, database: super::DatabaseId) -> Option<CommitSequence> {
46        (self.database == database && self.private.is_none())
47            .then_some(self.committed)
48            .flatten()
49    }
50
51    pub(crate) fn database(self) -> super::DatabaseId {
52        self.database
53    }
54
55    pub fn is_private(self) -> bool {
56        self.private.is_some()
57    }
58}
59
60impl From<BorrowedRecord<'_>> for RecordMetadata {
61    fn from(record: BorrowedRecord<'_>) -> Self {
62        Self {
63            revision: record.revision,
64            live: record.value.is_some(),
65        }
66    }
67}
68
69pub type RecordValueVisitor<'a> = dyn FnMut(Option<BorrowedRecord<'_>>) -> VersionResult<()> + 'a;
70pub type RecordScanVisitor<'a> = dyn FnMut(&[u8], BorrowedRecord<'_>) -> VersionResult<bool> + 'a;
71pub type RecordKeyVisitor<'a> = dyn FnMut(&[u8], RecordMetadata) -> VersionResult<bool> + 'a;
72
73/// 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.
74pub trait CommittedRecordSnapshot: Send + Sync {
75    fn sequence(&self) -> CommitSequence;
76
77    /// The provider's tombstone-reclamation epoch captured under the same admission as this snapshot. `None` means this provider does not supply reclamation-aware observations. Wrappers must forward the original value, never refresh it from current storage.
78    fn reclamation_epoch(&self) -> Option<u64> {
79        None
80    }
81
82    /// Read a revision and tombstone marker without materializing its value when the provider supports key-only access.
83    fn metadata(
84        &self,
85        key: &[u8],
86        control: &StorageReadControl,
87    ) -> VersionResult<Option<RecordMetadata>> {
88        let mut found = None;
89        self.visit_value(key, control, &mut |record| {
90            found = record.map(Into::into);
91            Ok(())
92        })?;
93        Ok(found)
94    }
95
96    /// Visit ordered record identities, including tombstones, without fetching values from providers with key-only access.
97    fn visit_keys(
98        &self,
99        prefix: &[u8],
100        after: Option<&[u8]>,
101        limit: usize,
102        control: &StorageReadControl,
103        visit: &mut RecordKeyVisitor<'_>,
104    ) -> VersionResult<()> {
105        self.visit_prefix(prefix, after, limit, control, &mut |key, record| {
106            visit(key, record.into())
107        })
108    }
109
110    /// Return the newest revision at or before this snapshot, including tombstones. Missing identities return `None`.
111    fn get(
112        &self,
113        key: &[u8],
114        control: &StorageReadControl,
115    ) -> VersionResult<Option<RecordVersion<SharedRecordValue>>>;
116
117    /// 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.
118    fn scan(
119        &self,
120        prefix: &[u8],
121        after: Option<&[u8]>,
122        limit: usize,
123        control: &StorageReadControl,
124    ) -> VersionResult<BudgetedVec<ScannedRecord>>;
125
126    /// Borrow provider-owned bytes for an internal storage visitor. Visitors must not reenter this snapshot or execute SQL/user callbacks.
127    fn visit_value(
128        &self,
129        key: &[u8],
130        control: &StorageReadControl,
131        visit: &mut RecordValueVisitor<'_>,
132    ) -> VersionResult<()> {
133        let record = self.get(key, control)?;
134        visit(record.as_ref().map(|record| BorrowedRecord {
135            revision: Some(record.sequence()),
136            value: record.value().map(|value| &***value),
137        }))?;
138        control.cancellation().check()?;
139        Ok(())
140    }
141
142    /// Check a live value's encoded size before provider materialization, without changing revision or tombstone visibility. The default rejects unsupported providers rather than materializing through `get`.
143    fn visit_value_bounded(
144        &self,
145        _key: &[u8],
146        _max_bytes: usize,
147        control: &StorageReadControl,
148        _visit: &mut RecordValueVisitor<'_>,
149    ) -> VersionResult<()> {
150        control.check()?;
151        Err(crate::StorageBackendError::Other(
152            "size-bounded committed record reads are not supported".into(),
153        )
154        .into())
155    }
156
157    /// 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.
158    fn visit_prefix(
159        &self,
160        prefix: &[u8],
161        after: Option<&[u8]>,
162        limit: usize,
163        control: &StorageReadControl,
164        visit: &mut RecordScanVisitor<'_>,
165    ) -> VersionResult<()> {
166        control.cancellation().check()?;
167        let mut cursor: Option<BudgetedVec<u8>> = None;
168        let mut remaining = limit;
169        while remaining != 0 {
170            let mut page = self.scan(
171                prefix,
172                cursor.as_deref().or(after),
173                remaining.min(64),
174                control,
175            )?;
176            if page.is_empty() {
177                break;
178            }
179            for record in page.iter() {
180                control.cancellation().check()?;
181                if !visit(
182                    &record.key,
183                    BorrowedRecord {
184                        revision: Some(record.version.sequence()),
185                        value: record.version.value().map(|value| &***value),
186                    },
187                )? {
188                    control.cancellation().check()?;
189                    return Ok(());
190                }
191                remaining -= 1;
192            }
193            cursor = page.pop().map(|record| record.key);
194        }
195        control.cancellation().check()?;
196        Ok(())
197    }
198}
199
200struct RetainedSnapshot<T> {
201    snapshot: T,
202    _memory: MemoryReservation,
203}
204
205/// 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.
206pub fn retain_record_snapshot<T: CommittedRecordSnapshot + 'static>(
207    snapshot: T,
208    control: &StorageReadControl,
209) -> VersionResult<Arc<dyn CommittedRecordSnapshot>> {
210    control.cancellation().check()?;
211    let memory = control
212        .memory()
213        .reserve(std::mem::size_of::<RetainedSnapshot<T>>())?;
214    Ok(Arc::new(RetainedSnapshot {
215        snapshot,
216        _memory: memory,
217    }))
218}
219
220impl<T: CommittedRecordSnapshot> CommittedRecordSnapshot for RetainedSnapshot<T> {
221    fn sequence(&self) -> CommitSequence {
222        self.snapshot.sequence()
223    }
224    fn reclamation_epoch(&self) -> Option<u64> {
225        self.snapshot.reclamation_epoch()
226    }
227    fn metadata(
228        &self,
229        key: &[u8],
230        control: &StorageReadControl,
231    ) -> VersionResult<Option<RecordMetadata>> {
232        self.snapshot.metadata(key, control)
233    }
234    fn visit_keys(
235        &self,
236        prefix: &[u8],
237        after: Option<&[u8]>,
238        limit: usize,
239        control: &StorageReadControl,
240        visit: &mut RecordKeyVisitor<'_>,
241    ) -> VersionResult<()> {
242        self.snapshot
243            .visit_keys(prefix, after, limit, control, visit)
244    }
245    fn get(
246        &self,
247        key: &[u8],
248        control: &StorageReadControl,
249    ) -> VersionResult<Option<RecordVersion<SharedRecordValue>>> {
250        self.snapshot.get(key, control)
251    }
252    fn scan(
253        &self,
254        prefix: &[u8],
255        after: Option<&[u8]>,
256        limit: usize,
257        control: &StorageReadControl,
258    ) -> VersionResult<BudgetedVec<ScannedRecord>> {
259        self.snapshot.scan(prefix, after, limit, control)
260    }
261    fn visit_value(
262        &self,
263        key: &[u8],
264        control: &StorageReadControl,
265        visit: &mut RecordValueVisitor<'_>,
266    ) -> VersionResult<()> {
267        self.snapshot.visit_value(key, control, visit)
268    }
269    fn visit_value_bounded(
270        &self,
271        key: &[u8],
272        max_bytes: usize,
273        control: &StorageReadControl,
274        visit: &mut RecordValueVisitor<'_>,
275    ) -> VersionResult<()> {
276        self.snapshot
277            .visit_value_bounded(key, max_bytes, control, visit)
278    }
279    fn visit_prefix(
280        &self,
281        prefix: &[u8],
282        after: Option<&[u8]>,
283        limit: usize,
284        control: &StorageReadControl,
285        visit: &mut RecordScanVisitor<'_>,
286    ) -> VersionResult<()> {
287        self.snapshot
288            .visit_prefix(prefix, after, limit, control, visit)
289    }
290}
291
292/// Visible payload plus its original committed precondition. Private deletions remain observable as tombstones rather than falling back to the committed value.
293pub struct VisibleRecord {
294    original: Option<CommitSequence>,
295    value: Option<SharedRecordValue>,
296    private: bool,
297}
298
299impl VisibleRecord {
300    pub fn original_revision(&self) -> Option<CommitSequence> {
301        self.original
302    }
303
304    pub fn value(&self) -> Option<&[u8]> {
305        self.value.as_ref().map(|value| &***value)
306    }
307
308    pub fn is_private(&self) -> bool {
309        self.private
310    }
311
312    fn committed(version: RecordVersion<SharedRecordValue>) -> Self {
313        let (sequence, value) = version.into_parts();
314        Self {
315            original: Some(sequence),
316            value,
317            private: false,
318        }
319    }
320
321    fn private(write: &PreparedRecordWrite) -> Self {
322        Self {
323            original: write.expected(),
324            value: write.shared_value(),
325            private: true,
326        }
327    }
328}
329
330pub struct ScannedVisibleRecord {
331    pub key: BudgetedVec<u8>,
332    pub record: VisibleRecord,
333}
334
335/// 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.
336pub struct MergedRecordSnapshot {
337    committed: Arc<dyn CommittedRecordSnapshot>,
338    private: PrivateRecordSnapshot,
339    serializable: Option<super::SerializableReadContext>,
340}
341
342impl MergedRecordSnapshot {
343    pub(in crate::mvcc) fn retained_source(
344        &self,
345        key: &[u8],
346        control: &StorageReadControl,
347    ) -> VersionResult<Option<Arc<dyn crate::key_value::KeyValueRead + Send + Sync>>> {
348        self.private.retained_source(key, control)
349    }
350
351    /// Select one live record's actual revision without materializing its payload. The owning provider supplies this snapshot's transaction-history identity, distinct from any restored data namespace.
352    pub fn record_revision(
353        &self,
354        database: super::DatabaseId,
355        key: &[u8],
356        control: &StorageReadControl,
357    ) -> VersionResult<Option<VisibleRecordRevision>> {
358        control.check()?;
359        let Some(metadata) = self.metadata(key, control)?.filter(|record| record.live) else {
360            control.check()?;
361            return Ok(None);
362        };
363        let private = self.private_keys(key, None, 1, control)?;
364        let private = private
365            .first()
366            .filter(|record| record.key() == key)
367            .map(super::PrivateRecordKey::revision);
368        control.check()?;
369        Ok(Some(VisibleRecordRevision {
370            database,
371            committed: metadata.revision,
372            private,
373        }))
374    }
375
376    /// Share the committed owner and retain the original private revision under its existing allowance.
377    pub fn try_clone(&self) -> VersionResult<Self> {
378        Ok(Self {
379            committed: Arc::clone(&self.committed),
380            private: self.private.try_clone()?,
381            serializable: self.serializable.clone(),
382        })
383    }
384
385    pub fn new(
386        committed: Arc<dyn CommittedRecordSnapshot>,
387        private: PrivateRecordSnapshot,
388    ) -> Self {
389        Self {
390            committed,
391            private,
392            serializable: None,
393        }
394    }
395
396    pub(crate) fn with_serializable(
397        mut self,
398        context: Option<super::SerializableReadContext>,
399    ) -> Self {
400        self.serializable = context;
401        self
402    }
403
404    /// Original logical reader attribution. Clones retain the same participant and fixed data boundaries.
405    pub fn serializable(&self) -> Option<&super::SerializableReadContext> {
406        self.serializable.as_ref()
407    }
408
409    pub fn sequence(&self) -> CommitSequence {
410        self.committed.sequence()
411    }
412
413    /// The pinned committed boundary underlying this command view, without private replacements.
414    pub fn committed(&self) -> &dyn CommittedRecordSnapshot {
415        self.committed.as_ref()
416    }
417
418    pub(super) fn retain_committed(&self) -> Arc<dyn CommittedRecordSnapshot> {
419        Arc::clone(&self.committed)
420    }
421
422    /// Bounded private key metadata for provider-derived cache invalidation; values are never materialized.
423    pub fn private_keys(
424        &self,
425        prefix: &[u8],
426        after: Option<&[u8]>,
427        limit: usize,
428        control: &StorageReadControl,
429    ) -> VersionResult<BudgetedVec<super::PrivateRecordKey>> {
430        self.private.scan_keys(prefix, after, limit, control)
431    }
432
433    pub fn visit_value(
434        &self,
435        key: &[u8],
436        control: &StorageReadControl,
437        visit: &mut RecordValueVisitor<'_>,
438    ) -> VersionResult<()> {
439        if let Some(write) = self.private.get(key, control)? {
440            visit(Some(BorrowedRecord {
441                revision: write.expected(),
442                value: write.value(),
443            }))?;
444            control.cancellation().check()?;
445            return Ok(());
446        }
447        self.committed.visit_value(key, control, visit)
448    }
449
450    /// Preserve private replacement/tombstone precedence while enforcing the physical source's encoded-value cap.
451    pub fn visit_value_bounded(
452        &self,
453        key: &[u8],
454        max_bytes: usize,
455        control: &StorageReadControl,
456        visit: &mut RecordValueVisitor<'_>,
457    ) -> VersionResult<()> {
458        control.check()?;
459        if let Some(write) = self.private.get(key, control)? {
460            if let Some(value) = write.value() {
461                control.check_value_size(value.len(), max_bytes)?;
462            }
463            visit(Some(BorrowedRecord {
464                revision: write.expected(),
465                value: write.value(),
466            }))?;
467            control.check()?;
468            return Ok(());
469        }
470        self.committed
471            .visit_value_bounded(key, max_bytes, control, visit)
472    }
473
474    pub fn visit_prefix(
475        &self,
476        prefix: &[u8],
477        after: Option<&[u8]>,
478        limit: usize,
479        control: &StorageReadControl,
480        visit: &mut RecordScanVisitor<'_>,
481    ) -> VersionResult<()> {
482        self.private.visit_merged::<super::projection::Values>(
483            &*self.committed,
484            prefix,
485            after,
486            limit,
487            control,
488            visit,
489        )
490    }
491
492    pub fn metadata(
493        &self,
494        key: &[u8],
495        control: &StorageReadControl,
496    ) -> VersionResult<Option<RecordMetadata>> {
497        if let Some(write) = self.private.get(key, control)? {
498            return Ok(Some(RecordMetadata {
499                revision: write.expected(),
500                live: write.value().is_some(),
501            }));
502        }
503        self.committed.metadata(key, control)
504    }
505
506    pub fn visit_keys(
507        &self,
508        prefix: &[u8],
509        after: Option<&[u8]>,
510        limit: usize,
511        control: &StorageReadControl,
512        visit: &mut RecordKeyVisitor<'_>,
513    ) -> VersionResult<()> {
514        self.private.visit_merged::<super::projection::Keys>(
515            &*self.committed,
516            prefix,
517            after,
518            limit,
519            control,
520            visit,
521        )
522    }
523
524    pub fn get(
525        &self,
526        key: &[u8],
527        control: &StorageReadControl,
528    ) -> VersionResult<Option<VisibleRecord>> {
529        if let Some(write) = self.private.get(key, control)? {
530            return Ok(Some(VisibleRecord::private(&write)));
531        }
532        Ok(self
533            .committed
534            .get(key, control)?
535            .map(VisibleRecord::committed))
536    }
537
538    /// Merge bounded pages without duplicate identities or payload copies. Tombstones are returned so callers can preserve revision observations even when no row is visible.
539    pub fn scan(
540        &self,
541        prefix: &[u8],
542        after: Option<&[u8]>,
543        limit: usize,
544        control: &StorageReadControl,
545    ) -> VersionResult<BudgetedVec<ScannedVisibleRecord>> {
546        control.cancellation().check()?;
547        let mut result = BudgetedVec::new(control.memory());
548        if limit == 0 {
549            return Ok(result);
550        }
551        let (committed, _committed_memory) = self
552            .committed
553            .scan(prefix, after, limit, control)?
554            .into_parts();
555        let (private, _private_memory) = self
556            .private
557            .scan(prefix, after, limit, control)?
558            .into_parts();
559        let mut committed = committed.into_iter().peekable();
560        let mut private = private.into_iter().peekable();
561        while result.len() < limit {
562            control.cancellation().check()?;
563            let order = match (committed.peek(), private.peek()) {
564                (Some(committed), Some(private)) => committed.key.as_ref().cmp(private.key()),
565                (Some(_), None) => Ordering::Less,
566                (None, Some(_)) => Ordering::Greater,
567                (None, None) => break,
568            };
569            let record = match order {
570                Ordering::Less => {
571                    let committed = committed.next().expect("peeked committed record");
572                    ScannedVisibleRecord {
573                        key: committed.key,
574                        record: VisibleRecord::committed(committed.version),
575                    }
576                }
577                Ordering::Equal => {
578                    let committed = committed.next().expect("peeked committed record");
579                    let private = private.next().expect("peeked private record");
580                    ScannedVisibleRecord {
581                        key: committed.key,
582                        record: VisibleRecord::private(&private),
583                    }
584                }
585                Ordering::Greater => {
586                    let private = private.next().expect("peeked private record");
587                    let mut key = BudgetedVec::new(control.memory());
588                    key.extend_from_slice(private.key())?;
589                    ScannedVisibleRecord {
590                        key,
591                        record: VisibleRecord::private(&private),
592                    }
593                }
594            };
595            result.push(record)?;
596        }
597        Ok(result)
598    }
599}
uqa_storage/mvcc/view.rs

uqa_storage/mvcc/
view.rs
