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
123
124
125
126
127
128
129
130
// SPDX-License-Identifier: Apache-2.0
// Copyright (c) 2024-present, fjall-rs
// Copyright (c) 2026-present, Dmitry Prudnikov
use alloc::sync::Arc;
use crate::{InternalValue, Table, table::Scanner, version::Run};
/// Scans through a disjoint run
///
/// Optimized for compaction, by using a `TableScanner` instead of `TableReader`.
pub struct RunScanner {
tables: Arc<Run<Table>>,
lo: usize,
hi: usize,
lo_reader: Option<Scanner>,
/// Told how long each table's reads take, for every table of the run.
pace: Option<crate::table::util::Pacer>,
/// Where each table's scanner records the row groups or data blocks it
/// reads, for a compaction that copies them whole.
carry: Option<crate::table::group_carry::CarryTarget>,
}
impl RunScanner {
/// Scans tables `lo..=hi` of `run`, telling `pace` how long the reads of
/// every one of them take.
pub fn culled(
run: Arc<Run<Table>>,
(lo, hi): (Option<usize>, Option<usize>),
pace: Option<crate::table::util::Pacer>,
) -> crate::Result<Self> {
let lo = lo.unwrap_or_default();
let hi = hi.unwrap_or(run.len() - 1);
#[expect(
clippy::expect_used,
reason = "we trust the caller to pass valid indexes"
)]
let lo_table = run.get(lo).expect("should exist");
let lo_reader = lo_table.scan_paced(pace.as_ref())?;
Ok(Self {
tables: run,
lo,
hi,
lo_reader: Some(lo_reader),
pace,
carry: None,
})
}
/// Has the scanner of every table of the run `target` takes record its
/// groups or blocks there (see
/// [`crate::table::group_carry::CarryTarget::takes`]).
#[must_use]
pub(crate) fn with_carry(mut self, target: crate::table::group_carry::CarryTarget) -> Self {
// Called on a scanner just built: its first reader is open and its
// table is in the run.
#[expect(
clippy::expect_used,
reason = "culled opened the reader of table lo, which the run holds"
)]
let table = self.tables.get(self.lo).expect("lo is within the run");
self.lo_reader = self
.lo_reader
.take()
.map(|reader| carrying(reader, table, &target));
self.carry = Some(target);
self
}
fn scan_table(&self, index: usize) -> crate::Result<Scanner> {
// `index` is within `lo..=hi`, every slot of which names a table: a
// missing one fails the scan rather than ending it early, which would
// hand a merge the run without its tail.
let Some(table) = self.tables.get(index) else {
return Err(crate::Error::from(crate::io::Error::other(
"run scanner: table index past the run",
)));
};
let scanner = table.scan_paced(self.pace.as_ref())?;
if let Some(target) = &self.carry {
return Ok(carrying(scanner, table, target));
}
Ok(scanner)
}
}
/// `scanner` of `table`, recording its groups or blocks in `target` when it
/// takes them.
fn carrying(
scanner: Scanner,
table: &Table,
target: &crate::table::group_carry::CarryTarget,
) -> Scanner {
if target.takes(table) {
scanner.with_carry(target.carry_for(table))
} else {
scanner
}
}
impl Iterator for RunScanner {
type Item = crate::Result<InternalValue>;
fn next(&mut self) -> Option<Self::Item> {
loop {
if let Some(lo_reader) = &mut self.lo_reader {
if let Some(item) = lo_reader.next() {
return Some(item);
}
// NOTE: Lo reader is empty, get next one
self.lo_reader = None;
self.lo += 1;
if self.lo <= self.hi {
self.lo_reader = Some(fail_iter!(self.scan_table(self.lo)));
}
} else {
return None;
}
}
}
}
#[cfg(test)]
#[expect(clippy::unwrap_used)]
mod tests;