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;
10mod last;
11mod selected;
12pub use selected::SelectedRecordRead;
13
14use std::cmp::Ordering;
15use std::sync::Arc;
16
17use uqa_core::memory::{BudgetedVec, MemoryReservation};
18
19use crate::read_control::StorageReadControl;
20
21use super::{
22    CommitSequence, PreparedRecordWrite, PrivateRecordSnapshot, RecordVersion, ScannedRecord,
23    SharedRecordValue, VersionResult,
24};
25
26#[derive(Clone, Copy)]
27pub struct BorrowedRecord<'a> {
28    pub revision: Option<CommitSequence>,
29    pub value: Option<&'a [u8]>,
30}
31
32/// Visible revision metadata, including tombstones, without retaining the encoded payload.
33#[derive(Clone, Copy, Debug, PartialEq, Eq)]
34pub struct RecordMetadata {
35    pub revision: Option<CommitSequence>,
36    pub live: bool,
37}
38
39/// 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.
40#[derive(Clone, Copy, Debug, PartialEq, Eq)]
41pub struct VisibleRecordRevision {
42    database: super::DatabaseId,
43    committed: Option<CommitSequence>,
44    private: Option<super::PrivateRecordRevision>,
45}
46
47impl VisibleRecordRevision {
48    pub(crate) fn committed(self, database: super::DatabaseId) -> Option<CommitSequence> {
49        (self.database == database && self.private.is_none())
50            .then_some(self.committed)
51            .flatten()
52    }
53
54    /// The committed sequence this view includes, whether or not private changes accompany it.
55    pub(crate) fn committed_sequence(self) -> Option<CommitSequence> {
56        self.committed
57    }
58
59    pub(crate) fn database(self) -> super::DatabaseId {
60        self.database
61    }
62
63    pub fn is_private(self) -> bool {
64        self.private.is_some()
65    }
66}
67
68impl From<BorrowedRecord<'_>> for RecordMetadata {
69    fn from(record: BorrowedRecord<'_>) -> Self {
70        Self {
71            revision: record.revision,
72            live: record.value.is_some(),
73        }
74    }
75}
76
77pub type RecordValueVisitor<'a> = dyn FnMut(Option<BorrowedRecord<'_>>) -> VersionResult<()> + 'a;
78pub type RecordPointVisitor<'a> =
79    dyn FnMut(&[u8], Option<BorrowedRecord<'_>>) -> VersionResult<bool> + 'a;
80pub type RecordKeyIterator<'a> = dyn Iterator<Item = VersionResult<BudgetedVec<u8>>> + 'a;
81pub type RecordScanVisitor<'a> = dyn FnMut(&[u8], BorrowedRecord<'_>) -> VersionResult<bool> + 'a;
82pub type RecordKeyVisitor<'a> = dyn FnMut(&[u8], RecordMetadata) -> VersionResult<bool> + 'a;
83
84/// A lazy sequence of metadata requests. A successful `advance` exposes one key until `accept` receives its result, including absent records and tombstones. These methods are storage-internal: they must not reenter a provider or execute user callbacks. Returning `false` closes the read window; an error stops before producing another key.
85pub trait RecordMetadataRequests {
86    fn advance(&mut self) -> VersionResult<bool>;
87    fn key(&self) -> &[u8];
88    fn accept(&mut self, metadata: Option<RecordMetadata>) -> VersionResult<()>;
89}
90
91/// 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.
92pub trait CommittedRecordSnapshot: Send + Sync {
93    fn sequence(&self) -> CommitSequence;
94
95    /// 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.
96    fn reclamation_epoch(&self) -> Option<u64> {
97        None
98    }
99
100    /// The provider's own snapshot type, which its native read paths may serve from physical projections at the same boundary. Wrappers must forward the original provider snapshot.
101    fn provider_snapshot(&self) -> Option<&dyn std::any::Any> {
102        None
103    }
104
105    /// The value [`VersionedPersistence::commit_monitor_version`](super::VersionedPersistence::commit_monitor_version) returned before this snapshot was captured. While the monitor still returns it, nothing was committed after the capture began, so this snapshot is the latest one. `None` when the provider has no monitor. Wrappers must forward the original value.
106    fn commit_monitor(&self) -> Option<u64> {
107        None
108    }
109
110    /// Take the monitor value another snapshot of this same sequence was captured at. Nothing was committed between the two captures, so this snapshot is the latest one for as long as the monitor returns that value. Returns false when the snapshot cannot change its value, and its holder then keeps the other snapshot instead. Keeping this one keeps what it has already read of its records.
111    fn adopt_commit_monitor(&self, _monitor: u64) -> bool {
112        false
113    }
114
115    /// Read a revision and tombstone marker without materializing its value when the provider supports key-only access.
116    fn metadata(
117        &self,
118        key: &[u8],
119        control: &StorageReadControl,
120    ) -> VersionResult<Option<RecordMetadata>> {
121        let mut found = None;
122        self.visit_value(key, control, &mut |record| {
123            found = record.map(Into::into);
124            Ok(())
125        })?;
126        Ok(found)
127    }
128
129    /// Resolve requested revisions in input order through `metadata`, without fetching payloads on key-only providers. Providers may share one physical read window, admitted only after the first key; the snapshot boundary, cancellation and each point-read diagnostic remain unchanged.
130    fn visit_metadata(
131        &self,
132        requests: &mut dyn RecordMetadataRequests,
133        control: &StorageReadControl,
134    ) -> VersionResult<()> {
135        loop {
136            control.check()?;
137            if !requests.advance()? {
138                return control.check().map_err(Into::into);
139            }
140            control.check()?;
141            let metadata = self.metadata(requests.key(), control)?;
142            requests.accept(metadata)?;
143        }
144    }
145
146    /// Visit ordered record identities, including tombstones, without fetching values from providers with key-only access.
147    fn visit_keys(
148        &self,
149        prefix: &[u8],
150        after: Option<&[u8]>,
151        limit: usize,
152        control: &StorageReadControl,
153        visit: &mut RecordKeyVisitor<'_>,
154    ) -> VersionResult<()> {
155        self.visit_prefix(prefix, after, limit, control, &mut |key, record| {
156            visit(key, record.into())
157        })
158    }
159
160    /// Visit the greatest visible identity with this prefix strictly before `before`, including a tombstone. Invoke the visitor at most once; providers may seek directly without reading preceding keys or values. The default preserves semantics for providers that only implement forward scans.
161    fn visit_last_key(
162        &self,
163        prefix: &[u8],
164        before: Option<&[u8]>,
165        control: &StorageReadControl,
166        visit: &mut RecordKeyVisitor<'_>,
167    ) -> VersionResult<()> {
168        last::by_scan(self, prefix, before, control, visit)
169    }
170
171    /// Return the newest revision at or before this snapshot, including tombstones. Missing identities return `None`.
172    fn get(
173        &self,
174        key: &[u8],
175        control: &StorageReadControl,
176    ) -> VersionResult<Option<RecordVersion<SharedRecordValue>>>;
177
178    /// 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.
179    fn scan(
180        &self,
181        prefix: &[u8],
182        after: Option<&[u8]>,
183        limit: usize,
184        control: &StorageReadControl,
185    ) -> VersionResult<BudgetedVec<ScannedRecord>>;
186
187    /// Borrow provider-owned bytes for an internal storage visitor. Visitors must not reenter this snapshot or execute SQL/user callbacks.
188    fn visit_value(
189        &self,
190        key: &[u8],
191        control: &StorageReadControl,
192        visit: &mut RecordValueVisitor<'_>,
193    ) -> VersionResult<()> {
194        let record = self.get(key, control)?;
195        visit(record.as_ref().map(|record| BorrowedRecord {
196            revision: Some(record.sequence()),
197            value: record.value().map(|value| &***value),
198        }))?;
199        control.cancellation().check()?;
200        Ok(())
201    }
202
203    /// 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`.
204    fn visit_value_bounded(
205        &self,
206        _key: &[u8],
207        _max_bytes: usize,
208        control: &StorageReadControl,
209        _visit: &mut RecordValueVisitor<'_>,
210    ) -> VersionResult<()> {
211        control.check()?;
212        Err(crate::StorageBackendError::Other(
213            "size-bounded committed record reads are not supported".into(),
214        )
215        .into())
216    }
217
218    /// Borrow requested values in input order under one optional provider read window. Key production and visitors are internal and must not reenter storage or invoke user code. Produce one key at a time; a false visitor or an error stops before requesting the next key, including duplicate and missing identities.
219    fn visit_values(
220        &self,
221        keys: &mut RecordKeyIterator<'_>,
222        control: &StorageReadControl,
223        visit: &mut RecordPointVisitor<'_>,
224    ) -> VersionResult<()> {
225        loop {
226            control.check()?;
227            let Some(key) = keys.next() else {
228                return Ok(());
229            };
230            let key = key?;
231            let mut more = true;
232            self.visit_value(&key, control, &mut |record| {
233                more = visit(&key, record)?;
234                Ok(())
235            })?;
236            control.check()?;
237            if !more {
238                return Ok(());
239            }
240        }
241    }
242
243    /// 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.
244    fn visit_prefix(
245        &self,
246        prefix: &[u8],
247        after: Option<&[u8]>,
248        limit: usize,
249        control: &StorageReadControl,
250        visit: &mut RecordScanVisitor<'_>,
251    ) -> VersionResult<()> {
252        control.cancellation().check()?;
253        let mut cursor: Option<BudgetedVec<u8>> = None;
254        let mut remaining = limit;
255        while remaining != 0 {
256            let mut page = self.scan(
257                prefix,
258                cursor.as_deref().or(after),
259                remaining.min(64),
260                control,
261            )?;
262            if page.is_empty() {
263                break;
264            }
265            for record in page.iter() {
266                control.cancellation().check()?;
267                if !visit(
268                    &record.key,
269                    BorrowedRecord {
270                        revision: Some(record.version.sequence()),
271                        value: record.version.value().map(|value| &***value),
272                    },
273                )? {
274                    control.cancellation().check()?;
275                    return Ok(());
276                }
277                remaining -= 1;
278            }
279            cursor = page.pop().map(|record| record.key);
280        }
281        control.cancellation().check()?;
282        Ok(())
283    }
284}
285
286struct RetainedSnapshot<T> {
287    snapshot: T,
288    _memory: MemoryReservation,
289}
290
291/// 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.
292pub fn retain_record_snapshot<T: CommittedRecordSnapshot + 'static>(
293    snapshot: T,
294    control: &StorageReadControl,
295) -> VersionResult<Arc<dyn CommittedRecordSnapshot>> {
296    control.cancellation().check()?;
297    let memory = control
298        .memory()
299        .reserve(std::mem::size_of::<RetainedSnapshot<T>>())?;
300    Ok(Arc::new(RetainedSnapshot {
301        snapshot,
302        _memory: memory,
303    }))
304}
305
306impl<T: CommittedRecordSnapshot> CommittedRecordSnapshot for RetainedSnapshot<T> {
307    fn sequence(&self) -> CommitSequence {
308        self.snapshot.sequence()
309    }
310    fn reclamation_epoch(&self) -> Option<u64> {
311        self.snapshot.reclamation_epoch()
312    }
313    fn provider_snapshot(&self) -> Option<&dyn std::any::Any> {
314        self.snapshot.provider_snapshot()
315    }
316    fn commit_monitor(&self) -> Option<u64> {
317        self.snapshot.commit_monitor()
318    }
319    fn adopt_commit_monitor(&self, monitor: u64) -> bool {
320        self.snapshot.adopt_commit_monitor(monitor)
321    }
322    fn visit_last_key(
323        &self,
324        prefix: &[u8],
325        before: Option<&[u8]>,
326        control: &StorageReadControl,
327        visit: &mut RecordKeyVisitor<'_>,
328    ) -> VersionResult<()> {
329        self.snapshot.visit_last_key(prefix, before, control, visit)
330    }
331    fn metadata(
332        &self,
333        key: &[u8],
334        control: &StorageReadControl,
335    ) -> VersionResult<Option<RecordMetadata>> {
336        self.snapshot.metadata(key, control)
337    }
338    fn visit_metadata(
339        &self,
340        requests: &mut dyn RecordMetadataRequests,
341        control: &StorageReadControl,
342    ) -> VersionResult<()> {
343        self.snapshot.visit_metadata(requests, control)
344    }
345    fn visit_keys(
346        &self,
347        prefix: &[u8],
348        after: Option<&[u8]>,
349        limit: usize,
350        control: &StorageReadControl,
351        visit: &mut RecordKeyVisitor<'_>,
352    ) -> VersionResult<()> {
353        self.snapshot
354            .visit_keys(prefix, after, limit, control, visit)
355    }
356    fn get(
357        &self,
358        key: &[u8],
359        control: &StorageReadControl,
360    ) -> VersionResult<Option<RecordVersion<SharedRecordValue>>> {
361        self.snapshot.get(key, control)
362    }
363    fn scan(
364        &self,
365        prefix: &[u8],
366        after: Option<&[u8]>,
367        limit: usize,
368        control: &StorageReadControl,
369    ) -> VersionResult<BudgetedVec<ScannedRecord>> {
370        self.snapshot.scan(prefix, after, limit, control)
371    }
372    fn visit_value(
373        &self,
374        key: &[u8],
375        control: &StorageReadControl,
376        visit: &mut RecordValueVisitor<'_>,
377    ) -> VersionResult<()> {
378        self.snapshot.visit_value(key, control, visit)
379    }
380    fn visit_values(
381        &self,
382        keys: &mut RecordKeyIterator<'_>,
383        control: &StorageReadControl,
384        visit: &mut RecordPointVisitor<'_>,
385    ) -> VersionResult<()> {
386        self.snapshot.visit_values(keys, control, visit)
387    }
388    fn visit_value_bounded(
389        &self,
390        key: &[u8],
391        max_bytes: usize,
392        control: &StorageReadControl,
393        visit: &mut RecordValueVisitor<'_>,
394    ) -> VersionResult<()> {
395        self.snapshot
396            .visit_value_bounded(key, max_bytes, control, visit)
397    }
398    fn visit_prefix(
399        &self,
400        prefix: &[u8],
401        after: Option<&[u8]>,
402        limit: usize,
403        control: &StorageReadControl,
404        visit: &mut RecordScanVisitor<'_>,
405    ) -> VersionResult<()> {
406        self.snapshot
407            .visit_prefix(prefix, after, limit, control, visit)
408    }
409}
410
411/// Visible payload plus its original committed precondition. Private deletions remain observable as tombstones rather than falling back to the committed value.
412pub struct VisibleRecord {
413    original: Option<CommitSequence>,
414    value: Option<SharedRecordValue>,
415    private: bool,
416}
417
418impl VisibleRecord {
419    pub fn original_revision(&self) -> Option<CommitSequence> {
420        self.original
421    }
422
423    pub fn value(&self) -> Option<&[u8]> {
424        self.value.as_ref().map(|value| &***value)
425    }
426
427    pub fn is_private(&self) -> bool {
428        self.private
429    }
430
431    fn committed(version: RecordVersion<SharedRecordValue>) -> Self {
432        let (sequence, value) = version.into_parts();
433        Self {
434            original: Some(sequence),
435            value,
436            private: false,
437        }
438    }
439
440    fn private(write: &PreparedRecordWrite) -> Self {
441        Self {
442            original: write.expected(),
443            value: write.shared_value(),
444            private: true,
445        }
446    }
447}
448
449pub struct ScannedVisibleRecord {
450    pub key: BudgetedVec<u8>,
451    pub record: VisibleRecord,
452}
453
454/// 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.
455pub struct MergedRecordSnapshot {
456    committed: Arc<dyn CommittedRecordSnapshot>,
457    private: PrivateRecordSnapshot,
458    serializable: Option<super::SerializableReadContext>,
459}
460
461impl MergedRecordSnapshot {
462    pub(in crate::mvcc) fn retained_source(
463        &self,
464        key: &[u8],
465        control: &StorageReadControl,
466    ) -> VersionResult<Option<Arc<dyn crate::key_value::KeyValueRead + Send + Sync>>> {
467        self.private.retained_source(key, control)
468    }
469
470    /// 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.
471    pub fn record_revision(
472        &self,
473        database: super::DatabaseId,
474        key: &[u8],
475        control: &StorageReadControl,
476    ) -> VersionResult<Option<VisibleRecordRevision>> {
477        control.check()?;
478        let (metadata, private) = match self.private.revision_metadata(key, control)? {
479            Some((metadata, revision)) => (Some(metadata), Some(revision)),
480            None => (self.committed.metadata(key, control)?, None),
481        };
482        let Some(metadata) = metadata.filter(|record| record.live) else {
483            control.check()?;
484            return Ok(None);
485        };
486        control.check()?;
487        Ok(Some(VisibleRecordRevision {
488            database,
489            committed: metadata.revision,
490            private,
491        }))
492    }
493
494    /// Share the committed owner and retain the original private revision under its existing allowance.
495    pub fn try_clone(&self) -> VersionResult<Self> {
496        Ok(Self {
497            committed: Arc::clone(&self.committed),
498            private: self.private.try_clone()?,
499            serializable: self.serializable.clone(),
500        })
501    }
502
503    pub fn new(
504        committed: Arc<dyn CommittedRecordSnapshot>,
505        private: PrivateRecordSnapshot,
506    ) -> Self {
507        Self {
508            committed,
509            private,
510            serializable: None,
511        }
512    }
513
514    pub(crate) fn with_serializable(
515        mut self,
516        context: Option<super::SerializableReadContext>,
517    ) -> Self {
518        self.serializable = context;
519        self
520    }
521
522    /// Original logical reader attribution. Clones retain the same participant and fixed data boundaries.
523    pub fn serializable(&self) -> Option<&super::SerializableReadContext> {
524        self.serializable.as_ref()
525    }
526
527    pub fn sequence(&self) -> CommitSequence {
528        self.committed.sequence()
529    }
530
531    /// Whole private command-root identity, restored together with its records on undo.
532    pub fn private_revision(&self) -> Option<super::PrivateRecordRevision> {
533        self.private.revision()
534    }
535
536    /// The pinned committed boundary underlying this command view, without private replacements.
537    pub fn committed(&self) -> &dyn CommittedRecordSnapshot {
538        self.committed.as_ref()
539    }
540
541    pub(super) fn retain_committed(&self) -> Arc<dyn CommittedRecordSnapshot> {
542        Arc::clone(&self.committed)
543    }
544
545    /// Bounded private key metadata for provider-derived cache invalidation; values are never materialized.
546    pub fn private_keys(
547        &self,
548        prefix: &[u8],
549        after: Option<&[u8]>,
550        limit: usize,
551        control: &StorageReadControl,
552    ) -> VersionResult<BudgetedVec<super::PrivateRecordKey>> {
553        self.private.scan_keys(prefix, after, limit, control)
554    }
555
556    /// Newest private revisions per provider-selected cache scope, or exact changed keys when the provider supplies no classifier. The summary shares this view's undo boundary and spills under the same allowance.
557    pub fn private_revision_scopes(
558        &self,
559        after: Option<&[u8]>,
560        limit: usize,
561        control: &StorageReadControl,
562    ) -> VersionResult<BudgetedVec<super::PrivateRecordKey>> {
563        self.private.revision_scopes(after, limit, control)
564    }
565
566    /// Read one complete provider-selected revision scope without scanning its changed records. The caller must pass a scope emitted by its provider's classifier, not a narrower or broader prefix. Providers without a classifier scan that prefix's key metadata using one retained cursor.
567    pub fn private_scope_revision(
568        &self,
569        scope: &[u8],
570        control: &StorageReadControl,
571    ) -> VersionResult<Option<super::PrivateRecordRevision>> {
572        self.private.scope_revision(scope, control)
573    }
574
575    /// Stream private entries on this command boundary without repeated key lookups. Providers can merge their committed rows with each entry's metadata or requested payload.
576    pub fn private_cursor(
577        &self,
578        prefix: &[u8],
579        after: Option<&[u8]>,
580        control: &StorageReadControl,
581    ) -> VersionResult<super::PrivateRecordCursor<'_>> {
582        self.private.cursor(prefix, after, control)
583    }
584
585    pub fn visit_value(
586        &self,
587        key: &[u8],
588        control: &StorageReadControl,
589        visit: &mut RecordValueVisitor<'_>,
590    ) -> VersionResult<()> {
591        if let Some(write) = self.private.get(key, control)? {
592            visit(Some(BorrowedRecord {
593                revision: write.expected(),
594                value: write.value(),
595            }))?;
596            control.cancellation().check()?;
597            return Ok(());
598        }
599        self.committed.visit_value(key, control, visit)
600    }
601
602    /// Keep private replacements ahead of committed values while allowing an unchanged command view to reuse the provider's point-read window.
603    pub fn visit_values(
604        &self,
605        keys: &mut RecordKeyIterator<'_>,
606        control: &StorageReadControl,
607        visit: &mut RecordPointVisitor<'_>,
608    ) -> VersionResult<()> {
609        self.selected(control).visit_values(keys, control, visit)
610    }
611
612    /// Preserve private replacement/tombstone precedence while enforcing the physical source's encoded-value cap.
613    pub fn visit_value_bounded(
614        &self,
615        key: &[u8],
616        max_bytes: usize,
617        control: &StorageReadControl,
618        visit: &mut RecordValueVisitor<'_>,
619    ) -> VersionResult<()> {
620        control.check()?;
621        if let Some(write) = self.private.get(key, control)? {
622            if let Some(value) = write.value() {
623                control.check_value_size(value.len(), max_bytes)?;
624            }
625            visit(Some(BorrowedRecord {
626                revision: write.expected(),
627                value: write.value(),
628            }))?;
629            control.check()?;
630            return Ok(());
631        }
632        self.committed
633            .visit_value_bounded(key, max_bytes, control, visit)
634    }
635
636    pub fn visit_prefix(
637        &self,
638        prefix: &[u8],
639        after: Option<&[u8]>,
640        limit: usize,
641        control: &StorageReadControl,
642        visit: &mut RecordScanVisitor<'_>,
643    ) -> VersionResult<()> {
644        self.private.visit_merged::<super::projection::Values>(
645            &*self.committed,
646            prefix,
647            after,
648            limit,
649            control,
650            visit,
651        )
652    }
653
654    pub fn metadata(
655        &self,
656        key: &[u8],
657        control: &StorageReadControl,
658    ) -> VersionResult<Option<RecordMetadata>> {
659        if let Some(metadata) = self.private.metadata(key, control)? {
660            return Ok(Some(metadata));
661        }
662        self.committed.metadata(key, control)
663    }
664
665    pub fn visit_keys(
666        &self,
667        prefix: &[u8],
668        after: Option<&[u8]>,
669        limit: usize,
670        control: &StorageReadControl,
671        visit: &mut RecordKeyVisitor<'_>,
672    ) -> VersionResult<()> {
673        self.private.visit_merged::<super::projection::Keys>(
674            &*self.committed,
675            prefix,
676            after,
677            limit,
678            control,
679            visit,
680        )
681    }
682
683    /// Whether this view holds a private change of `key`, whose committed record is then not what a read of the key returns.
684    pub fn has_private_change(
685        &self,
686        key: &[u8],
687        control: &StorageReadControl,
688    ) -> VersionResult<bool> {
689        Ok(self.private.get(key, control)?.is_some())
690    }
691
692    pub fn get(
693        &self,
694        key: &[u8],
695        control: &StorageReadControl,
696    ) -> VersionResult<Option<VisibleRecord>> {
697        if let Some(write) = self.private.get(key, control)? {
698            return Ok(Some(VisibleRecord::private(&write)));
699        }
700        Ok(self
701            .committed
702            .get(key, control)?
703            .map(VisibleRecord::committed))
704    }
705
706    /// Merge bounded pages without duplicate identities or payload copies. Tombstones are returned so callers can preserve revision observations even when no row is visible.
707    pub fn scan(
708        &self,
709        prefix: &[u8],
710        after: Option<&[u8]>,
711        limit: usize,
712        control: &StorageReadControl,
713    ) -> VersionResult<BudgetedVec<ScannedVisibleRecord>> {
714        control.cancellation().check()?;
715        let mut result = BudgetedVec::new(control.memory());
716        if limit == 0 {
717            return Ok(result);
718        }
719        let (committed, _committed_memory) = self
720            .committed
721            .scan(prefix, after, limit, control)?
722            .into_parts();
723        let (private, _private_memory) = self
724            .private
725            .scan(prefix, after, limit, control)?
726            .into_parts();
727        let mut committed = committed.into_iter().peekable();
728        let mut private = private.into_iter().peekable();
729        while result.len() < limit {
730            control.cancellation().check()?;
731            let order = match (committed.peek(), private.peek()) {
732                (Some(committed), Some(private)) => committed.key.as_ref().cmp(private.key()),
733                (Some(_), None) => Ordering::Less,
734                (None, Some(_)) => Ordering::Greater,
735                (None, None) => break,
736            };
737            let record = match order {
738                Ordering::Less => {
739                    let committed = committed.next().expect("peeked committed record");
740                    ScannedVisibleRecord {
741                        key: committed.key,
742                        record: VisibleRecord::committed(committed.version),
743                    }
744                }
745                Ordering::Equal => {
746                    let committed = committed.next().expect("peeked committed record");
747                    let private = private.next().expect("peeked private record");
748                    ScannedVisibleRecord {
749                        key: committed.key,
750                        record: VisibleRecord::private(&private),
751                    }
752                }
753                Ordering::Greater => {
754                    let private = private.next().expect("peeked private record");
755                    let mut key = BudgetedVec::new(control.memory());
756                    key.extend_from_slice(private.key())?;
757                    ScannedVisibleRecord {
758                        key,
759                        record: VisibleRecord::private(&private),
760                    }
761                }
762            };
763            result.push(record)?;
764        }
765        Ok(result)
766    }
767}
uqa_storage/mvcc/view.rs

uqa_storage/mvcc/
view.rs
