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
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
use std::ops::Bound;
use crate::error::DbResult;
use crate::map_index::Dir;
use super::builder::clone_bound;
use super::{BulkScan, Scope, StrictRows};
/// Maximum number of keys deleted by one internal cursor pass.
pub(crate) const CHUNK: usize = 512;
/// The result of one [`Scope::delete_up_to`] call.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Progress {
/// Number of keys deleted successfully by this call.
pub deleted: usize,
/// Whether this call observed exhaustion while collecting the batch and its
/// one-key probe.
pub done: bool,
}
impl<'l, S: BulkScan + 'l, const N: usize> Scope<'l, S, N> {
/// Delete every key currently reached by this scope.
///
/// This is an `O(m)` key-only walk for `m` keys in the selected bounds. It is
/// weakly consistent and non-atomic: concurrent writes may be observed or
/// missed, and a newly inserted key behind this operation's cursor survives.
/// Corrupt rows and TTL cleanup are handled by the collection's key scan and
/// delete paths, so this method can return their errors after prior keys were
/// already deleted.
///
/// A source configured with a change stream or unique claim targets is
/// refused before the scan begins with [`DbError::ChangeStreamRequiresTx`] or
/// [`DbError::UniqueIndexRequiresTx`], respectively. Use
/// [`crate::Tx::delete_scope`] for those sources.
pub fn delete_all(&self) -> DbResult<usize> {
self.scan.reject_nontx_bulk()?;
let mut start = clone_bound(&self.start);
let mut end = clone_bound(&self.end);
let mut deleted = 0;
loop {
let (keys, exhausted) = scan_key_chunk(self.scan, start, end, self.dir, CHUNK)?;
let delete_len = keys.len().min(CHUNK);
let last_deleted_key = keys.get(delete_len.saturating_sub(1)).copied();
for key in &keys[..delete_len] {
deleted += usize::from(self.scan.delete_key(key)?);
}
if exhausted {
return Ok(deleted);
}
let last_deleted_key = last_deleted_key.expect("non-exhausted chunk has a key");
match self.dir {
Dir::Asc => start = Bound::Excluded(last_deleted_key),
Dir::Desc => end = Bound::Excluded(last_deleted_key),
}
}
}
/// Delete at most `limit` keys and report the per-call progress.
///
/// This is an `O(min(limit, 512, m))` key-only walk for `m` keys in the
/// selected bounds, deleting no more than 512 keys in one call. It is weakly
/// consistent and non-atomic: concurrent writes may be observed or missed, and
/// a newly inserted key behind this call's cursor survives. Corrupt rows and
/// TTL cleanup use the collection's key scan and delete paths; an error can
/// therefore follow earlier successful deletions.
///
/// A zero limit performs no scan or mutation and returns `done: false`.
/// A source configured with a change stream or unique claim targets is
/// refused before the scan begins with [`DbError::ChangeStreamRequiresTx`] or
/// [`DbError::UniqueIndexRequiresTx`], respectively. Use
/// [`crate::Tx::delete_scope`] for those sources.
pub fn delete_up_to(&self, limit: usize) -> DbResult<Progress> {
if limit == 0 {
return Ok(Progress {
deleted: 0,
done: false,
});
}
self.scan.reject_nontx_bulk()?;
let take_limit = limit.min(CHUNK);
let (keys, exhausted) = scan_key_chunk(
self.scan,
clone_bound(&self.start),
clone_bound(&self.end),
self.dir,
limit,
)?;
let delete_len = keys.len().min(take_limit);
let mut deleted = 0;
for key in &keys[..delete_len] {
deleted += usize::from(self.scan.delete_key(key)?);
}
Ok(Progress {
deleted,
done: exhausted,
})
}
/// Return whether this scope currently has no keys.
///
/// This is a weakly consistent key-only probe. On a Tree it stops at the
/// first key; on a Map it initializes the ordered shard merge and probes
/// every non-exhausted shard's key batch before returning its first merged
/// key or proving the scope empty. It does not decode values, so corrupt rows
/// and rows awaiting TTL cleanup still count as present; a concurrent write
/// can change the answer immediately after it is returned.
pub fn is_empty(&self) -> DbResult<bool> {
let mut iter =
self.scan
.try_keys(clone_bound(&self.start), clone_bound(&self.end), self.dir);
match iter.next() {
Some(Ok(_)) => Ok(false),
Some(Err(error)) => Err(error),
None => Ok(true),
}
}
/// Count keys currently reached by this scope.
///
/// This is an `O(m)` weakly consistent key-only walk for `m` keys in the
/// selected bounds. It does not decode values, so corrupt rows and rows awaiting
/// TTL cleanup count; concurrent writes can make the result stale immediately.
pub fn count(&self) -> DbResult<usize> {
let mut count = 0;
for key in self
.scan
.try_keys(clone_bound(&self.start), clone_bound(&self.end), self.dir)
{
key?;
count += 1;
}
Ok(count)
}
}
impl<'l, S: BulkScan + StrictRows + 'l, const N: usize> Scope<'l, S, N> {
/// Keep only rows accepted by `f` and return the number deleted.
///
/// The predicate receives owned snapshots and runs without a collection lock.
/// This operation is weakly consistent and non-atomic: a concurrent replacement
/// or removal between the scan and conditional delete is left intact and does not
/// contribute to the count. Scan or decode errors abort the operation after any
/// earlier successful deletions.
///
/// A source configured with a change stream or unique claim targets is
/// refused before the scan begins with [`DbError::ChangeStreamRequiresTx`] or
/// [`DbError::UniqueIndexRequiresTx`], respectively; v1 has no transactional
/// retention alternative. TypedTree, TypedMap, and LsmTree adapters also need
/// values that implement `PartialEq`: the owned strict seam uses their native
/// compare-delete operation. The predicate owns a row snapshot; on skiplist
/// families its scan keeps a `seize` guard for the complete pass, delaying
/// reclamation of rows deleted by that pass until it returns.
pub fn retain(&self, mut f: impl FnMut(&S::Key, &S::Val) -> bool) -> DbResult<usize> {
self.scan.reject_nontx_bulk()?;
let mut deleted = 0;
for row in self
.scan
.try_rows(clone_bound(&self.start), clone_bound(&self.end), self.dir)
{
let (key, value) = row?;
if !f(&key, &value) && self.scan.delete_if_unchanged(&key, &value)? {
deleted += 1;
}
}
Ok(deleted)
}
}
fn scan_key_chunk<S: BulkScan>(
scan: S,
start: Bound<S::Key>,
end: Bound<S::Key>,
dir: Dir,
limit: usize,
) -> DbResult<(Vec<S::Key>, bool)> {
let take_limit = limit.min(CHUNK);
let take_plus_probe = take_limit + 1;
let mut keys = Vec::with_capacity(take_plus_probe);
let mut iter = scan.try_keys(start, end, dir);
while keys.len() < take_plus_probe {
match iter.next() {
Some(Ok(key)) => keys.push(key),
Some(Err(error)) => return Err(error),
None => break,
}
}
let exhausted = keys.len() < take_plus_probe;
drop(iter);
Ok((keys, exhausted))
}