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
//! Compacting a key range on request.
use tessarikv_types::KeyspaceId;
use super::Db;
use super::catalog::Command;
use super::compaction::manual::Request;
use crate::{Error, Result};
impl Db {
/// Compacts the user keys `[start, end)` of `keyspace` (unbounded above when `end` is
/// `None`) and returns once done: everything committed before the call is flushed, every
/// table holding keys in the range is merged down to the bottom level, and the bottom-level
/// tables the range touches are rewritten, so overwritten versions and deletes leave the disk
/// unless a snapshot still reads them. Writes, reads and automatic compactions go on
/// meanwhile.
///
/// It rewrites every byte the range holds, once per level it passes, so bound the range: the
/// whole keyspace needs free space on the order of its size. A universal keyspace merges all
/// its runs into the bottom level when any of them touches the range; a first-in-first-out
/// keyspace never merges, so it only flushes. In memory there are no tables and it returns at
/// once.
///
/// # Errors
///
/// [`crate::Error::UnknownKeyspace`] for a keyspace this database does not hold, and
/// [`crate::Error::FlushFailed`] or [`crate::Error::CompactionFailed`] when the background
/// work stopped.
pub async fn compact_range(
&self,
keyspace: KeyspaceId,
start: &[u8],
end: Option<&[u8]>,
) -> Result<()> {
self.check_keyspace(keyspace)?;
let Some(installer) = &self.installer else {
return Ok(());
};
self.flush().await?;
// The installer serves requests over a bounded channel and answers on the request's own
// reply channel; a closed channel means it stopped, for the reason it published.
let (request, outcome) = Request::new(keyspace.get(), start, end);
let stopped = || {
installer
.signals
.failure()
.unwrap_or(Error::CompactionFailed)
};
installer
.requests
.send(Command::Compact(request))
.await
.map_err(|_| stopped())?;
outcome.await.map_err(|_| stopped())?
}
}