uqa_storage/mvcc/commit/
cursor.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! Ordered reads of the writes of a prepared commit.
8
9use std::ops::Bound;
10use std::sync::Arc;
11
12use crate::mvcc::overlay::run::{RunCursor, RunEntry, SpilledRun};
13use crate::mvcc::VersionResult;
14use crate::read_control::StorageReadControl;
15
16use super::metadata::PreparedWriteMetadata;
17use super::PreparedRecordWrite;
18
19#[cfg(test)]
20mod tests;
21
22enum Source<'a> {
23    Resident(std::slice::Iter<'a, PreparedRecordWrite>),
24    Spilled(RunCursor),
25}
26
27/// Visits the writes of a prepared commit in order. A spilled commit is read one block at a time, and each value is loaded only when its write is returned.
28pub struct PreparedWriteCursor<'a> {
29    source: Source<'a>,
30}
31
32impl<'a> PreparedWriteCursor<'a> {
33    pub(super) fn resident(writes: &'a [PreparedRecordWrite]) -> Self {
34        Self {
35            source: Source::Resident(writes.iter()),
36        }
37    }
38
39    pub(super) fn spilled(run: &Arc<SpilledRun>) -> Self {
40        Self {
41            source: Source::Spilled(run.cursor(Bound::Unbounded)),
42        }
43    }
44
45    pub(super) fn spilled_from(run: &Arc<SpilledRun>, start: &[u8]) -> Self {
46        Self {
47            source: Source::Spilled(run.cursor(Bound::Included(start))),
48        }
49    }
50
51    /// The next write with its value, or `None` after the last.
52    pub fn next(
53        &mut self,
54        control: &StorageReadControl,
55    ) -> VersionResult<Option<PreparedRecordWrite>> {
56        self.next_matching(None, control)
57    }
58
59    /// The next selected write, testing metadata before loading its value. Skipped
60    /// spilled values are neither read nor admitted to the caller's allowance.
61    pub fn next_where(
62        &mut self,
63        control: &StorageReadControl,
64        mut select: impl FnMut(&PreparedWriteMetadata) -> VersionResult<bool>,
65    ) -> VersionResult<Option<PreparedRecordWrite>> {
66        match &mut self.source {
67            Source::Resident(writes) => {
68                for write in writes.by_ref() {
69                    control.cancellation().check()?;
70                    let metadata = PreparedWriteMetadata::new(
71                        write.shared_key(),
72                        write.expected(),
73                        write.kind(),
74                        write.value().map(|value| value.len() as u64),
75                    );
76                    if select(&metadata)? {
77                        return Ok(Some(write.clone()));
78                    }
79                }
80                control.cancellation().check()?;
81                Ok(None)
82            }
83            Source::Spilled(cursor) => {
84                while let Some(entry) = cursor.next(control)? {
85                    let metadata = PreparedWriteMetadata::new(
86                        entry.key.clone(),
87                        entry.expected,
88                        entry.kind,
89                        entry.value.map(|location| location.len),
90                    );
91                    if select(&metadata)? {
92                        return spilled_write(cursor.run(), entry, control).map(Some);
93                    }
94                }
95                Ok(None)
96            }
97        }
98    }
99
100    /// Read only one record family, skipping other payloads without loading them.
101    pub(in crate::mvcc) fn next_with_kind(
102        &mut self,
103        kind: super::RecordWriteKind,
104        control: &StorageReadControl,
105    ) -> VersionResult<Option<PreparedRecordWrite>> {
106        self.next_matching(Some(kind), control)
107    }
108
109    fn next_matching(
110        &mut self,
111        kind: Option<super::RecordWriteKind>,
112        control: &StorageReadControl,
113    ) -> VersionResult<Option<PreparedRecordWrite>> {
114        match &mut self.source {
115            Source::Resident(writes) => {
116                for write in writes.by_ref() {
117                    control.cancellation().check()?;
118                    if kind.is_none_or(|kind| write.kind() == kind) {
119                        return Ok(Some(write.clone()));
120                    }
121                }
122                control.cancellation().check()?;
123                Ok(None)
124            }
125            Source::Spilled(cursor) => {
126                while let Some(entry) = cursor.next(control)? {
127                    if kind.is_none_or(|kind| entry.kind == kind) {
128                        return spilled_write(cursor.run(), entry, control).map(Some);
129                    }
130                }
131                Ok(None)
132            }
133        }
134    }
135
136    /// What the next write changes, without loading its value, or `None` after the last.
137    pub fn next_metadata(
138        &mut self,
139        control: &StorageReadControl,
140    ) -> VersionResult<Option<PreparedWriteMetadata>> {
141        match &mut self.source {
142            Source::Resident(writes) => {
143                control.cancellation().check()?;
144                Ok(writes.next().map(|write| {
145                    PreparedWriteMetadata::new(
146                        write.shared_key(),
147                        write.expected(),
148                        write.kind(),
149                        write.value().map(|value| value.len() as u64),
150                    )
151                }))
152            }
153            Source::Spilled(cursor) => Ok(cursor.next(control)?.map(|entry| {
154                let value_len = entry.value.map(|location| location.len);
155                PreparedWriteMetadata::new(entry.key, entry.expected, entry.kind, value_len)
156            })),
157        }
158    }
159}
160
161fn spilled_write(
162    run: &SpilledRun,
163    entry: RunEntry,
164    control: &StorageReadControl,
165) -> VersionResult<PreparedRecordWrite> {
166    let value = entry
167        .value
168        .map(|location| run.load_value(location, control))
169        .transpose()?;
170    Ok(PreparedRecordWrite::from_shared(entry.key, entry.expected, value).with_kind(entry.kind))
171}
uqa_storage/mvcc/commit/cursor.rs

uqa_storage/mvcc/commit/
cursor.rs
