1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
//! Range reads inside a transaction: pages of the snapshot merged with the transaction's own
//! writes.
use std::cmp::Ordering;
use tessarikv_types::KeyspaceId;
use super::Transaction;
use super::write_set::Own;
use crate::{ReadOptions, Result, ScanPage, ScanRange};
impl Transaction<'_> {
/// Up to `limit` live entries of `keyspace` with keys in `[start, end)` (unbounded above
/// when `end` is `None`) as this transaction sees them: the database as it was when the
/// transaction began, with the transaction's own puts, deletes and range deletes applied. A
/// full page names the key the next page starts at, as [`crate::Db::scan`] does.
///
/// # Errors
///
/// As [`crate::Db::scan`].
pub async fn scan(
&mut self,
keyspace: KeyspaceId,
start: &[u8],
end: Option<&[u8]>,
limit: usize,
options: &ReadOptions,
) -> Result<ScanPage> {
self.db.check_keyspace(keyspace)?;
let page = self
.merged_page(keyspace, start, end, limit, options)
.await?;
if let Some(reads) = &mut self.reads {
// The range this page answered for: up to where the next page starts.
reads.range(keyspace, start, page.next.as_deref().or(end));
}
Ok(page)
}
/// One page for each of `ranges` in `keyspace`, in the order given, as [`Transaction::scan`]
/// returns each: at the transaction's snapshot, with its own writes, every range read
/// recorded for a serializable transaction.
///
/// # Errors
///
/// As [`Transaction::scan`].
pub async fn multi_scan<'a>(
&mut self,
keyspace: KeyspaceId,
ranges: impl IntoIterator<Item = ScanRange<'a>>,
options: &ReadOptions,
) -> Result<Vec<ScanPage>> {
let ranges = ranges.into_iter();
let mut pages = Vec::with_capacity(ranges.size_hint().0);
for ScanRange { start, end, limit } in ranges {
pages.push(self.scan(keyspace, start, end, limit, options).await?);
}
Ok(pages)
}
/// The page [`Transaction::scan`] returns.
async fn merged_page(
&self,
keyspace: KeyspaceId,
start: &[u8],
end: Option<&[u8]>,
limit: usize,
options: &ReadOptions,
) -> Result<ScanPage> {
let mut page = ScanPage::default();
let mut cursor = start.to_vec();
loop {
// Each snapshot page covers `[cursor, its next)`, or up to `end` when it is the last.
let snapshot = self
.snapshot
.scan(keyspace, &cursor, end, limit.max(1), options)
.await?;
let covered = snapshot.next.as_deref().or(end);
let own = self.writes.values_in(keyspace, &cursor, covered);
let mut theirs = snapshot.entries.into_iter().peekable();
let mut ours = own.into_iter().peekable();
loop {
let candidate = match (theirs.peek(), ours.peek()) {
(None, None) => break,
(Some((key, _)), Some((own_key, _))) => key.as_slice().cmp(own_key),
(Some(_), None) => Ordering::Less,
(None, Some(_)) => Ordering::Greater,
};
let (key, value) = match candidate {
Ordering::Less => match theirs.next() {
Some((key, value)) => match self.writes.lookup(keyspace, &key) {
Own::Untouched => (key, value),
// A key this transaction wrote is among its own values, or deleted.
Own::Value(_) | Own::Deleted => continue,
},
None => break,
},
Ordering::Equal => {
theirs.next();
match ours.next() {
Some((key, value)) => (key.to_vec(), value.to_vec()),
None => break,
}
}
Ordering::Greater => match ours.next() {
Some((key, value)) => (key.to_vec(), value.to_vec()),
None => break,
},
};
if page.entries.len() >= limit {
page.next = Some(key);
return Ok(page);
}
page.entries.push((key, value));
}
match snapshot.next {
Some(next) => cursor = next,
None => return Ok(page),
}
}
}
}