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
19enum Source<'a> {
20    Resident(std::slice::Iter<'a, PreparedRecordWrite>),
21    Spilled(RunCursor),
22}
23
24/// 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.
25pub struct PreparedWriteCursor<'a> {
26    source: Source<'a>,
27}
28
29impl<'a> PreparedWriteCursor<'a> {
30    pub(super) fn resident(writes: &'a [PreparedRecordWrite]) -> Self {
31        Self {
32            source: Source::Resident(writes.iter()),
33        }
34    }
35
36    pub(super) fn spilled(run: &Arc<SpilledRun>) -> Self {
37        Self {
38            source: Source::Spilled(run.cursor(Bound::Unbounded)),
39        }
40    }
41
42    pub(super) fn spilled_from(run: &Arc<SpilledRun>, start: &[u8]) -> Self {
43        Self {
44            source: Source::Spilled(run.cursor(Bound::Included(start))),
45        }
46    }
47
48    /// The next write with its value, or `None` after the last.
49    pub fn next(
50        &mut self,
51        control: &StorageReadControl,
52    ) -> VersionResult<Option<PreparedRecordWrite>> {
53        match &mut self.source {
54            Source::Resident(writes) => {
55                control.cancellation().check()?;
56                Ok(writes.next().cloned())
57            }
58            Source::Spilled(cursor) => {
59                let Some(entry) = cursor.next(control)? else {
60                    return Ok(None);
61                };
62                spilled_write(cursor.run(), entry, control).map(Some)
63            }
64        }
65    }
66
67    /// What the next write changes, without loading its value, or `None` after the last.
68    pub fn next_metadata(
69        &mut self,
70        control: &StorageReadControl,
71    ) -> VersionResult<Option<PreparedWriteMetadata>> {
72        match &mut self.source {
73            Source::Resident(writes) => {
74                control.cancellation().check()?;
75                Ok(writes.next().map(|write| {
76                    PreparedWriteMetadata::new(
77                        write.shared_key(),
78                        write.expected(),
79                        write.kind(),
80                        write.value().map(|value| value.len() as u64),
81                    )
82                }))
83            }
84            Source::Spilled(cursor) => Ok(cursor.next(control)?.map(|entry| {
85                let value_len = entry.value.map(|location| location.len);
86                PreparedWriteMetadata::new(entry.key, entry.expected, entry.kind, value_len)
87            })),
88        }
89    }
90}
91
92fn spilled_write(
93    run: &SpilledRun,
94    entry: RunEntry,
95    control: &StorageReadControl,
96) -> VersionResult<PreparedRecordWrite> {
97    let value = entry
98        .value
99        .map(|location| run.load_value(location, control))
100        .transpose()?;
101    Ok(PreparedRecordWrite::from_shared(entry.key, entry.expected, value).with_kind(entry.kind))
102}
uqa_storage/mvcc/commit/cursor.rs

uqa_storage/mvcc/commit/
cursor.rs
