#[cfg(test)]
#[allow(
clippy::unwrap_used,
clippy::indexing_slicing,
clippy::useless_vec,
clippy::unnecessary_map_or,
reason = "test code"
)]
mod test;
#[cfg(test)]
#[expect(clippy::unwrap_used, clippy::indexing_slicing, reason = "test code")]
mod picker_tests;
#[cfg(test)]
#[expect(clippy::expect_used, reason = "test code")]
mod primitive_tests;
mod primitives;
pub use primitives::{Granularity, Layout, LevelPolicy, Movement, Trigger};
use super::{Choice, CompactionStrategy, Input as CompactionInput};
use crate::{
HashSet, TableId,
compaction::state::{CompactionState, hidden_set::HiddenSet},
config::Config,
table::{Table, util::aggregate_run_key_range},
version::{Level, Version, run::Ranged},
};
#[cfg(not(feature = "std"))]
use alloc::vec::Vec;
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
struct MergeCost {
promoted: u64,
total: u64,
heat: u64,
}
impl MergeCost {
fn cheaper_than(self, other: Self) -> bool {
u128::from(self.total) * u128::from(other.promoted)
< u128::from(other.total) * u128::from(self.promoted)
}
fn better_than(self, other: Self, movement: Movement) -> bool {
match movement {
Movement::LeastRewrite { .. } => self.cheaper_than(other),
Movement::Hottest => {
let ours = u128::from(self.heat) * u128::from(other.total);
let theirs = u128::from(other.heat) * u128::from(self.total);
ours > theirs || (ours == theirs && self.cheaper_than(other))
}
}
}
}
struct MergeRanking<T> {
movement: Movement,
bound: u64,
offered: usize,
within: Option<(usize, MergeCost, T)>,
any: Option<(usize, MergeCost, T)>,
second: Option<MergeCost>,
}
impl<T: Copy> MergeRanking<T> {
fn new(overshoot: u64, slack: u64, movement: Movement) -> Self {
Self {
movement,
bound: overshoot.saturating_add(slack),
offered: 0,
within: None,
any: None,
second: None,
}
}
fn offer(&mut self, cost: MergeCost, candidate: T) {
let index = self.offered;
let movement = self.movement;
self.offered += 1;
if cost.promoted <= self.bound
&& self
.within
.as_ref()
.is_none_or(|(_, best, _)| cost.better_than(*best, movement))
{
self.within = Some((index, cost, candidate));
}
match &self.any {
Some((_, best, _)) if !cost.better_than(*best, movement) => {
if self
.second
.is_none_or(|second| cost.better_than(second, movement))
{
self.second = Some(cost);
}
}
previous => {
self.second = previous.as_ref().map(|(_, best, _)| *best);
self.any = Some((index, cost, candidate));
}
}
}
fn finish(self) -> Option<((usize, MergeCost, T), Option<MergeCost>)> {
match (self.within, self.any) {
(Some(within), Some(any)) if within.0 != any.0 => Some((within, Some(any.1))),
(Some(chosen), _) | (None, Some(chosen)) => Some((chosen, self.second)),
(None, None) => None,
}
}
fn offered(&self) -> usize {
self.offered
}
}
fn pick_minimal_compaction(
curr_level: &Level,
next_level: &Level,
hidden_set: &HiddenSet,
overshoot: u64,
table_base_size: u64,
movement: Movement,
cmp: &dyn crate::comparator::UserComparator,
) -> Option<(HashSet<TableId>, bool)> {
let promotion_slack = match movement {
Movement::LeastRewrite { slack } => slack.unwrap_or(table_base_size),
Movement::Hottest => table_base_size,
};
if let Some(ids) = largest_trivial_move(curr_level, next_level, hidden_set, cmp) {
return Some((ids, true));
}
if next_level.is_empty() {
return None;
}
let mut ranking = MergeRanking::new(overshoot, promotion_slack, movement);
let cap = 50 * table_base_size;
for (run_index, run) in next_level.iter().enumerate() {
let window_bytes = prefix_sums(run.iter().map(Table::file_size));
let window_hidden =
prefix_sums(run.iter().map(|t| u64::from(hidden_set.is_hidden(t.id()))));
let window_heat = prefix_sums(run.iter().map(Table::heat));
let pull_ins: Vec<PullIn> = curr_level
.iter()
.map(|curr_run| PullIn::new(curr_run, run, hidden_set, cmp))
.collect();
for width in 1..=run.len() {
let mut within_cap = false;
for start in 0..=(run.len() - width) {
let end = start + width;
#[expect(
clippy::indexing_slicing,
reason = "start < end <= run.len(), and the sums hold run.len() + 1 entries"
)]
let (next_level_size, next_hidden, next_heat) = (
window_bytes[end] - window_bytes[start],
window_hidden[end] - window_hidden[start],
window_heat[end] - window_heat[start],
);
if next_level_size > cap {
continue;
}
within_cap = true;
if next_hidden > 0 {
continue;
}
let mut promoted = 0;
let mut heat = next_heat;
let mut blocked = false;
for pull_in in &pull_ins {
let (bytes, hidden, reads) = pull_in.contained(start, end - 1);
promoted += bytes;
heat += reads;
blocked |= hidden > 0;
}
if promoted == 0 || blocked {
continue;
}
ranking.offer(
MergeCost {
promoted,
total: promoted + next_level_size,
heat,
},
(run_index, start, end),
);
}
if !within_cap {
break;
}
}
}
let offered = ranking.offered();
let ((chosen, cost, (run_index, start, end)), runner_up) = ranking.finish()?;
log::debug!(
"leveled: merge candidate {chosen} of {offered} chosen at {cost:?}, runner-up \
{runner_up:?} (overshoot {overshoot}, slack {promotion_slack})",
);
#[expect(
clippy::expect_used,
reason = "the ranking holds a window it was offered from these runs"
)]
let window = next_level
.iter()
.nth(run_index)
.and_then(|run| run.get(start..end))
.expect("the chosen window lies in its run");
let key_range = aggregate_run_key_range(window);
let contained: HashSet<_> = curr_level
.iter()
.flat_map(|run| run.get_contained_cmp(&key_range, cmp))
.map(Table::id)
.collect();
let pulled_in = contained.len();
let curr_ids = with_overlapping_runs(curr_level, contained, hidden_set, cmp)?;
let mut below: HashSet<_> = window.iter().map(Table::id).collect();
if curr_ids.len() > pulled_in || next_level.run_count() > 1 {
#[expect(clippy::expect_used, reason = "the window holds a table")]
let span = span_of(
curr_level
.iter()
.flat_map(|run| run.iter())
.filter(|table| curr_ids.contains(&table.id()))
.chain(window.iter()),
cmp,
)
.expect("a span over the window");
for table in next_level
.iter()
.flat_map(|run| run.get_overlapping_cmp(&span, cmp))
{
if below.insert(table.id()) && hidden_set.is_hidden(table.id()) {
return None;
}
}
below = with_overlapping_runs(next_level, below, hidden_set, cmp)?;
}
let mut ids = below;
ids.extend(curr_ids);
Some((ids, false))
}
fn span_of<'a>(
tables: impl Iterator<Item = &'a Table>,
cmp: &dyn crate::comparator::UserComparator,
) -> Option<crate::KeyRange> {
use core::cmp::Ordering;
let mut span: Option<(crate::UserKey, crate::UserKey)> = None;
for table in tables {
let range = table.key_range();
span = Some(match span {
None => (range.min().clone(), range.max().clone()),
Some((min, max)) => (
if cmp.compare(range.min(), &min) == Ordering::Less {
range.min().clone()
} else {
min
},
if cmp.compare(range.max(), &max) == Ordering::Greater {
range.max().clone()
} else {
max
},
),
});
}
span.map(crate::KeyRange::new)
}
fn with_overlapping_runs(
level: &Level,
mut ids: HashSet<TableId>,
hidden_set: &HiddenSet,
cmp: &dyn crate::comparator::UserComparator,
) -> Option<HashSet<TableId>> {
if level.run_count() <= 1 {
return Some(ids);
}
loop {
let Some(span) = span_of(
level
.iter()
.flat_map(|run| run.iter())
.filter(|table| ids.contains(&table.id())),
cmp,
) else {
return Some(ids);
};
let mut grew = false;
for table in level
.iter()
.flat_map(|run| run.get_overlapping_cmp(&span, cmp))
{
if ids.insert(table.id()) {
if hidden_set.is_hidden(table.id()) {
return None;
}
grew = true;
}
}
if !grew {
return Some(ids);
}
}
}
fn largest_trivial_move(
curr_level: &Level,
next_level: &Level,
hidden_set: &HiddenSet,
cmp: &dyn crate::comparator::UserComparator,
) -> Option<HashSet<TableId>> {
use core::ops::Bound;
let free_of = |runs: &[alloc::sync::Arc<crate::version::run::Run<Table>>],
skip: Option<usize>,
bounds: (Bound<&crate::Slice>, Bound<&crate::Slice>)| {
runs.iter().enumerate().all(|(index, run)| {
Some(index) == skip
|| run
.range_overlap_indexes_cmp::<crate::Slice, _, _>(&bounds, cmp)
.is_none()
})
};
let clear = |bounds| free_of(next_level, None, bounds);
for (run_index, curr_run) in curr_level.iter().enumerate() {
let mut best: Option<(usize, usize)> = None;
let (mut stretch_start, mut stretch_len) = (0, 0);
let mut previous: Option<&Table> = None;
for (index, table) in curr_run.iter().enumerate() {
let range = table.key_range();
let bounds = (Bound::Included(range.min()), Bound::Included(range.max()));
let movable = !hidden_set.is_hidden(table.id())
&& clear(bounds)
&& free_of(curr_level, Some(run_index), bounds);
if !movable {
stretch_len = 0;
previous = None;
continue;
}
let joins = previous.is_some_and(|previous| {
clear((
Bound::Excluded(previous.key_range().max()),
Bound::Excluded(range.min()),
))
});
if joins {
stretch_len += 1;
} else {
(stretch_start, stretch_len) = (index, 1);
}
previous = Some(table);
if best.is_none_or(|(_, len)| stretch_len > len) {
best = Some((stretch_start, stretch_len));
}
}
if let Some((start, len)) = best {
return curr_run
.get(start..start + len)
.map(|window| window.iter().map(Table::id).collect());
}
}
None
}
fn window_over(
level: &Level,
hidden_set: &HiddenSet,
overshoot: u64,
movement: Movement,
table_base_size: u64,
) -> Option<HashSet<TableId>> {
let (run, start, end) = match movement {
Movement::LeastRewrite { .. } => level.iter().find_map(|run| {
let start = run
.iter()
.position(|table| !hidden_set.is_hidden(table.id()))?;
let mut bytes = 0u64;
let mut end = start;
for table in run.iter().skip(start) {
if hidden_set.is_hidden(table.id()) {
break;
}
end += 1;
bytes += table.file_size();
if bytes >= overshoot {
break;
}
}
Some((run, start, end))
})?,
Movement::Hottest => {
let mut ranking = MergeRanking::new(overshoot, table_base_size, movement);
let cap = 50 * table_base_size;
for (run_index, run) in level.iter().enumerate() {
let bytes = prefix_sums(run.iter().map(Table::file_size));
let hidden =
prefix_sums(run.iter().map(|t| u64::from(hidden_set.is_hidden(t.id()))));
let heat = prefix_sums(run.iter().map(Table::heat));
for width in 1..=run.len() {
let mut within_cap = false;
for start in 0..=(run.len() - width) {
let end = start + width;
#[expect(
clippy::indexing_slicing,
reason = "start < end <= run.len(), and the sums hold run.len() + 1 entries"
)]
let (size, hidden, heat) = (
bytes[end] - bytes[start],
hidden[end] - hidden[start],
heat[end] - heat[start],
);
if size > cap {
continue;
}
within_cap = true;
if hidden > 0 {
continue;
}
ranking.offer(
MergeCost {
promoted: size,
total: size,
heat,
},
(run_index, start, end),
);
}
if !within_cap {
break;
}
}
}
let ((_, _, (run_index, start, end)), _) = ranking.finish()?;
(level.iter().nth(run_index)?, start, end)
}
};
let ids: HashSet<_> = run.get(start..end)?.iter().map(Table::id).collect();
(!ids.is_empty()).then_some(ids)
}
fn prefix_sums(items: impl Iterator<Item = u64>) -> Vec<u64> {
let mut sums = Vec::with_capacity(items.size_hint().0 + 1);
sums.push(0);
let mut total = 0;
for item in items {
total += item;
sums.push(total);
}
sums
}
struct PullIn {
first_from: Vec<usize>,
end_through: Vec<usize>,
bytes: Vec<u64>,
hidden: Vec<u64>,
reads: Vec<u64>,
}
impl PullIn {
fn new(
run: &[Table],
window_run: &[Table],
hidden_set: &HiddenSet,
cmp: &dyn crate::comparator::UserComparator,
) -> Self {
use core::cmp::Ordering;
let mut first_from = Vec::with_capacity(window_run.len());
let mut at = 0;
for table in window_run {
let min = table.key_range().min();
while run
.get(at)
.is_some_and(|t| cmp.compare(t.key_range().min(), min) == Ordering::Less)
{
at += 1;
}
first_from.push(at);
}
let mut end_through = Vec::with_capacity(window_run.len());
at = 0;
for table in window_run {
let max = table.key_range().max();
while run
.get(at)
.is_some_and(|t| cmp.compare(t.key_range().max(), max) != Ordering::Greater)
{
at += 1;
}
end_through.push(at);
}
Self {
first_from,
end_through,
bytes: prefix_sums(run.iter().map(Table::file_size)),
hidden: prefix_sums(run.iter().map(|t| u64::from(hidden_set.is_hidden(t.id())))),
reads: prefix_sums(run.iter().map(Table::heat)),
}
}
#[expect(
clippy::indexing_slicing,
reason = "first and last index the window run, and the stretch ends lie in 0..=run.len()"
)]
fn contained(&self, first: usize, last: usize) -> (u64, u64, u64) {
let (from, through) = (self.first_from[first], self.end_through[last]);
if from < through {
(
self.bytes[through] - self.bytes[from],
self.hidden[through] - self.hidden[from],
self.reads[through] - self.reads[from],
)
} else {
(0, 0, 0)
}
}
}
#[doc(hidden)]
pub const NAME: &str = "LeveledCompaction";
pub const DEFAULT_L0_FILE_THRESHOLD: u16 = 500;
#[derive(Clone)]
pub struct Strategy {
l0_threshold: u8,
l0_file_threshold: u16,
target_size: u64,
level_ratio_policy: Vec<f32>,
dynamic: bool,
multi_level: bool,
promotion_slack: Option<u64>,
policies: [Option<LevelPolicy>; 7],
}
impl Default for Strategy {
fn default() -> Self {
Self {
l0_threshold: 4,
l0_file_threshold: DEFAULT_L0_FILE_THRESHOLD,
target_size: 64 * 1_024 * 1_024,
level_ratio_policy: vec![10.0],
dynamic: false,
multi_level: false,
promotion_slack: None,
policies: [None; 7],
}
}
}
impl Strategy {
#[must_use]
pub fn with_level_ratio_policy(mut self, policy: Vec<f32>) -> Self {
self.level_ratio_policy = policy;
self
}
#[must_use]
pub fn with_l0_threshold(mut self, threshold: u8) -> Self {
self.l0_threshold = threshold.max(1);
self
}
#[must_use]
pub fn with_l0_file_threshold(mut self, threshold: u16) -> Self {
self.l0_file_threshold = threshold.max(1);
self
}
fn l0_triggered(&self, l0: &Level) -> bool {
shape_triggered(self.policy(0, 0, usize::MAX).trigger, l0)
}
}
fn shape_triggered(trigger: Trigger, level: &Level) -> bool {
match trigger {
Trigger::Shape { runs, tables } => {
level.run_count() >= usize::from(runs.max(1))
|| level.table_count() >= usize::from(tables.max(1))
}
Trigger::Size | Trigger::Never => false,
}
}
fn shape_score((runs, tables): (u8, u16), level: &Level) -> f64 {
#[expect(
clippy::cast_precision_loss,
reason = "counts of runs and tables are far below 2^52"
)]
let (run_count, table_count) = (level.run_count() as f64, level.table_count() as f64);
(run_count / f64::from(runs.max(1))).max(table_count / f64::from(tables.max(1)))
}
fn shape_window_owed((runs, tables): (u8, u16), level: &Level) -> u64 {
fn smallest(mut sizes: Vec<u64>, count: usize) -> u64 {
sizes.sort_unstable();
sizes.iter().take(count).sum()
}
let excess = |held: usize, limit: usize| {
let limit = limit.max(1);
if held >= limit { held + 1 - limit } else { 0 }
};
let run_bytes = smallest(
level
.iter()
.map(|run| run.iter().map(Table::file_size).sum())
.collect(),
excess(level.run_count(), usize::from(runs)),
);
let table_bytes = smallest(
level
.iter()
.flat_map(|run| run.iter())
.map(Table::file_size)
.collect(),
excess(level.table_count(), usize::from(tables)),
);
run_bytes.max(table_bytes)
}
impl Strategy {
#[must_use]
pub fn with_table_target_size(mut self, bytes: u64) -> Self {
self.target_size = bytes;
self
}
#[must_use]
pub fn with_dynamic_level_bytes(mut self, enabled: bool) -> Self {
self.dynamic = enabled;
self
}
#[must_use]
pub fn with_multi_level(mut self, enabled: bool) -> Self {
self.multi_level = enabled;
self
}
#[must_use]
pub fn with_promotion_slack(mut self, bytes: u64) -> Self {
self.promotion_slack = Some(bytes);
self
}
#[must_use]
pub fn with_level_policy(mut self, level: u8, policy: LevelPolicy) -> Self {
if let Some(slot) = self.policies.get_mut(usize::from(level)) {
*slot = Some(policy);
}
self
}
fn canonical_l1(&self, version: &Version, state: &CompactionState) -> (usize, usize, usize) {
#[expect(clippy::map_unwrap_or)]
let first_non_empty_level = version
.iter_levels()
.enumerate()
.skip(1)
.find(|(_, lvl)| !lvl.is_empty())
.map(|(idx, _)| idx)
.unwrap_or_else(|| version.level_count() - 1);
let mut canonical_l1_idx = first_non_empty_level;
let mut level_shift = canonical_l1_idx - 1;
if canonical_l1_idx > 1 && version.iter_levels().skip(1).any(|lvl| !lvl.is_empty()) {
let need_new_l1 = version
.iter_levels()
.enumerate()
.skip(1)
.filter(|(_, lvl)| !lvl.is_empty())
.all(|(idx, level)| {
let level_size = level
.iter()
.flat_map(|x| x.iter())
.filter(|x| !state.hidden_set().is_hidden(x.id()))
.map(Table::file_size)
.sum::<u64>();
#[expect(
clippy::cast_possible_truncation,
reason = "level index is bounded by level count (7, technically 255)"
)]
let target_size = self.level_target_size((idx - level_shift) as u8);
level_size > target_size
});
if need_new_l1 {
canonical_l1_idx -= 1;
level_shift -= 1;
}
}
(first_non_empty_level, canonical_l1_idx, level_shift)
}
fn policy(&self, physical: usize, level_shift: usize, last: usize) -> LevelPolicy {
let level = if physical == 0 {
0
} else {
physical.saturating_sub(level_shift).max(1)
};
let policy = self
.policies
.get(level)
.copied()
.flatten()
.unwrap_or_else(|| self.leveled_policy(level));
if physical == last && physical > 0 {
return LevelPolicy {
trigger: Trigger::Never,
..policy
};
}
policy
}
fn leveled_policy(&self, level: usize) -> LevelPolicy {
LevelPolicy {
trigger: if level == 0 {
Trigger::Shape {
runs: self.l0_threshold,
tables: self.l0_file_threshold,
}
} else {
Trigger::Size
},
layout: if level == 0 {
Layout::Tiered
} else {
Layout::Leveled
},
granularity: if level == 0 {
Granularity::Level {
through_full_next: self.multi_level,
}
} else {
Granularity::Window
},
movement: Movement::LeastRewrite {
slack: self.promotion_slack,
},
}
}
fn level_base_size(&self) -> u64 {
self.target_size * u64::from(self.l0_threshold)
}
fn level_target_size(&self, canonical_level_idx: u8) -> u64 {
assert!(
canonical_level_idx >= 1,
"level_target_size does not apply to L0",
);
if canonical_level_idx == 1 {
self.level_base_size()
} else {
#[expect(
clippy::cast_precision_loss,
reason = "precision loss is acceptable for level size calculations"
)]
let mut size = self.level_base_size() as f32;
for idx in 0..=(canonical_level_idx - 2) {
let ratio = self
.level_ratio_policy
.get(usize::from(idx))
.copied()
.unwrap_or_else(|| self.level_ratio_policy.last().copied().unwrap_or(10.0));
size *= ratio;
}
#[expect(
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
reason = "size is always positive and will never even come close to u64::MAX"
)]
{
size as u64
}
}
}
fn compute_level_targets(
&self,
version: &Version,
level_shift: usize,
state: &CompactionState,
) -> [u64; 7] {
let mut targets = [u64::MAX; 7];
targets[0] = self.l0_size_target();
if self.dynamic {
let mut lmax_idx = None;
for (idx, lvl) in version.iter_levels().enumerate().skip(1) {
if !lvl.is_empty() {
lmax_idx = Some(idx);
}
}
if let Some(lmax_idx) = lmax_idx {
#[expect(
clippy::expect_used,
reason = "lmax_idx was found by iterating levels, so it must exist"
)]
let lmax_level = version.level(lmax_idx).expect("level should exist");
let lmax_size: u64 = lmax_level
.iter()
.flat_map(|run| run.iter())
.filter(|table| !state.hidden_set().is_hidden(table.id()))
.map(Table::file_size)
.sum();
if let Some(slot) = targets.get_mut(lmax_idx) {
*slot = lmax_size;
}
#[expect(
clippy::cast_precision_loss,
reason = "precision loss is acceptable for level size calculations"
)]
let mut current_target = lmax_size as f64;
let dynamic_l1_idx = level_shift + 1;
for idx in (dynamic_l1_idx..lmax_idx).rev() {
let canonical = idx - level_shift;
let ratio_idx = canonical.saturating_sub(1);
let ratio = f64::from(
self.level_ratio_policy
.get(ratio_idx)
.copied()
.unwrap_or_else(|| {
self.level_ratio_policy.last().copied().unwrap_or(10.0)
}),
);
if !ratio.is_finite() || ratio <= 0.0 {
return self.compute_static_targets(level_shift);
}
current_target /= ratio;
#[expect(
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
reason = "target is always positive"
)]
if let Some(slot) = targets.get_mut(idx) {
*slot = current_target as u64;
}
}
let static_l1 = self.level_base_size();
if targets.get(dynamic_l1_idx).copied().unwrap_or(0) < static_l1 {
return self.compute_static_targets(level_shift);
}
return targets;
}
}
self.compute_static_targets(level_shift)
}
fn l0_size_target(&self) -> u64 {
if self.policy(0, 0, usize::MAX).trigger == Trigger::Size {
self.level_base_size()
} else {
u64::MAX
}
}
fn compute_static_targets(&self, level_shift: usize) -> [u64; 7] {
let mut targets = [u64::MAX; 7];
targets[0] = self.l0_size_target();
for (idx, slot) in targets.iter_mut().enumerate().skip(1) {
if idx <= level_shift {
continue; }
#[expect(
clippy::cast_possible_truncation,
reason = "level index is bounded by level count (7)"
)]
{
*slot = self.level_target_size((idx - level_shift) as u8);
}
}
targets
}
}
impl CompactionStrategy for Strategy {
fn get_name(&self) -> &'static str {
NAME
}
fn get_config(&self) -> Vec<crate::KvPair> {
vec![
(
crate::UserKey::from("leveled_l0_threshold"),
crate::UserValue::from(self.l0_threshold.to_le_bytes()),
),
(
crate::UserKey::from("leveled_l0_file_threshold"),
crate::UserValue::from(self.l0_file_threshold.to_le_bytes()),
),
(
crate::UserKey::from("leveled_target_size"),
crate::UserValue::from(self.target_size.to_le_bytes()),
),
(
crate::UserKey::from("leveled_level_ratio_policy"),
crate::UserValue::from({
use crate::io::{LittleEndian, WriteBytesExt};
let mut v = vec![];
#[expect(
clippy::expect_used,
clippy::cast_possible_truncation,
reason = "writing into Vec should not fail; policies have length of 255 max"
)]
v.write_u8(self.level_ratio_policy.len() as u8)
.expect("cannot fail");
for &f in &self.level_ratio_policy {
#[expect(clippy::expect_used, reason = "writing into Vec should not fail")]
v.write_f32::<LittleEndian>(f).expect("cannot fail");
}
v
}),
),
(
crate::UserKey::from("leveled_dynamic"),
crate::UserValue::from([u8::from(self.dynamic)]),
),
(
crate::UserKey::from("leveled_multi_level"),
crate::UserValue::from([u8::from(self.multi_level)]),
),
]
}
fn pending_compaction_bytes(&self, version: &Version) -> u64 {
let state = CompactionState::default();
let last = version.level_count() - 1;
let (_, _, level_shift) = self.canonical_l1(version, &state);
let targets = self.compute_level_targets(version, level_shift, &state);
let mut debt = 0u64;
for (idx, level) in version.iter_levels().enumerate() {
let policy = self.policy(idx, level_shift, last);
match policy.trigger {
trigger @ Trigger::Shape { runs, tables } => {
if shape_triggered(trigger, level) {
debt += match policy.granularity {
Granularity::Level { .. } => level.size(),
Granularity::Window => shape_window_owed((runs, tables), level),
};
}
}
Trigger::Never if idx != last => {}
Trigger::Size | Trigger::Never => {
let target = targets.get(idx).copied().unwrap_or(u64::MAX);
debt += level.size().saturating_sub(target);
}
}
}
debt
}
#[expect(clippy::too_many_lines)]
fn choose(&self, version: &Version, config: &Config, state: &CompactionState) -> Choice {
assert!(
version.level_count() == usize::from(config.level_count),
"leveled compaction requires version.level_count() == config.level_count (got version={}, config={})",
version.level_count(),
config.level_count,
);
let cmp = config.comparator.as_ref();
let last = version.level_count() - 1;
'trivial_lmax: {
if self.policy(0, 0, last).trigger == Trigger::Never {
break 'trivial_lmax;
}
#[expect(
clippy::expect_used,
reason = "level 0 is guaranteed to exist in a valid version"
)]
let l0 = version.level(0).expect("first level should exist");
if !l0.is_empty() && l0.is_disjoint() {
let lmax_index = version.level_count() - 1;
if (1..lmax_index).any(|idx| {
#[expect(
clippy::expect_used,
reason = "levels within level_count are guaranteed to exist"
)]
let level = version.level(idx).expect("level should exist");
!level.is_empty()
}) {
break 'trivial_lmax;
}
#[expect(
clippy::expect_used,
reason = "lmax_index is derived from level_count so level is guaranteed to exist"
)]
let lmax = version.level(lmax_index).expect("last level should exist");
if !lmax
.aggregate_key_range_cmp(cmp)
.overlaps_with_key_range_cmp(&l0.aggregate_key_range_cmp(cmp), cmp)
{
return Choice::Move(CompactionInput {
table_ids: l0.list_ids(),
#[expect(
clippy::cast_possible_truncation,
reason = "level count is at most 7, fits in u8"
)]
dest_level: lmax_index as u8,
canonical_level: 1,
target_size: self.target_size,
});
}
}
}
let (first_non_empty_level, canonical_l1_idx, level_shift) =
self.canonical_l1(version, state);
'trivial: {
let first_level = version.l0();
let target_level_idx = first_non_empty_level.min(canonical_l1_idx);
if self.policy(0, level_shift, last).trigger == Trigger::Never {
break 'trivial;
}
if first_level.run_count() == 1 {
if version.level_is_busy(0, state.hidden_set())
|| version.level_is_busy(target_level_idx, state.hidden_set())
{
break 'trivial;
}
let Some(target_level) = &version.level(target_level_idx) else {
break 'trivial;
};
if target_level.run_count() != 1 {
break 'trivial;
}
let key_range = first_level.aggregate_key_range_cmp(cmp);
let get_overlapping = target_level
.iter()
.flat_map(|run| run.get_overlapping_cmp(&key_range, cmp))
.map(Table::id)
.next();
if get_overlapping.is_none() && first_level.is_disjoint() {
#[expect(
clippy::cast_possible_truncation,
reason = "level index is bounded by level count (7)"
)]
return Choice::Move(CompactionInput {
table_ids: first_level.list_ids(),
dest_level: target_level_idx as u8,
canonical_level: 1,
target_size: self.target_size,
});
}
}
}
if let Trigger::Shape { runs, .. } = self.policy(0, level_shift, last).trigger {
let first_level = version.l0();
if first_level.run_count() > 1
&& first_level.table_count() < usize::from(runs.max(1))
&& !self.l0_triggered(first_level)
&& !version.level_is_busy(0, state.hidden_set())
{
return Choice::Merge(CompactionInput {
table_ids: first_level.list_ids(),
dest_level: 0,
canonical_level: 0,
target_size: self.target_size,
});
}
}
let level_targets = self.compute_level_targets(version, level_shift, state);
let mut scores = [( 0.0, 0u64); 7];
{
for (idx, level) in version.iter_levels().enumerate() {
let trigger = self.policy(idx, level_shift, last).trigger;
if let Trigger::Shape { runs, tables } = trigger {
if shape_triggered(trigger, level)
&& let Some(slot) = scores.get_mut(idx)
{
*slot = (shape_score((runs, tables), level), 0);
}
continue;
}
if trigger == Trigger::Never || level.is_empty() {
continue;
}
let level_size = level
.iter()
.flat_map(|x| x.iter())
.filter(|x| !state.hidden_set().is_hidden(x.id()))
.map(Table::file_size)
.sum::<u64>();
#[expect(clippy::indexing_slicing)]
let target_size = level_targets[idx];
#[expect(
clippy::indexing_slicing,
reason = "idx is from iter_levels().enumerate() so always < 7 = scores.len()"
)]
if level_size > target_size {
#[expect(
clippy::cast_precision_loss,
reason = "precision loss is acceptable for scoring calculations"
)]
let score = level_size as f64 / target_size as f64;
scores[idx] = (score, level_size - target_size);
if version
.level(idx + 1)
.is_some_and(|next_level| next_level.is_empty())
{
scores[idx] = (99.99, 999);
}
}
}
}
#[expect(clippy::expect_used, reason = "highest score is expected to exist")]
let (level_idx_with_highest_score, (score, overshoot_bytes)) = scores
.into_iter()
.enumerate()
.max_by(|(_, (score_a, _)), (_, (score_b, _))| {
score_a
.partial_cmp(score_b)
.unwrap_or(core::cmp::Ordering::Equal)
})
.expect("should have highest score somewhere");
if score < 1.0 {
return Choice::DoNothing;
}
let l0_policy = self.policy(0, level_shift, last);
if let (0, Granularity::Level { through_full_next }) =
(level_idx_with_highest_score, l0_policy.granularity)
{
let Some(first_level) = version.level(0) else {
return Choice::DoNothing;
};
if version.level_is_busy(0, state.hidden_set())
|| version.level_is_busy(canonical_l1_idx, state.hidden_set())
{
return Choice::DoNothing;
}
let Some(target_level) = &version.level(canonical_l1_idx) else {
return Choice::DoNothing;
};
let mut table_ids = first_level.list_ids();
let key_range = first_level.aggregate_key_range_cmp(cmp);
let target_level_overlapping_table_ids: Vec<_> = target_level
.iter()
.flat_map(|run| run.get_overlapping_cmp(&key_range, cmp))
.map(Table::id)
.collect();
let l1_tiered =
self.policy(canonical_l1_idx, level_shift, last).layout == Layout::Tiered;
if !l1_tiered {
table_ids.extend(&target_level_overlapping_table_ids);
}
if through_full_next && !l1_tiered {
let l1_score = scores.get(canonical_l1_idx).map_or(0.0, |(s, _)| *s);
let l2_idx = canonical_l1_idx + 1;
if l1_score > 1.0
&& l2_idx < version.level_count()
&& !version.level_is_busy(l2_idx, state.hidden_set())
&& let Some(l2) = version.level(l2_idx)
{
table_ids.extend(target_level.list_ids());
{
let mut input_ranges: Vec<_> = target_level
.iter()
.chain(first_level.iter())
.flat_map(|run| run.iter())
.map(|t| t.key_range().clone())
.collect();
input_ranges.sort_by(|a, b| cmp.compare(a.min(), b.min()));
let merged = crate::KeyRange::merge_sorted_cmp(input_ranges, cmp);
for run in l2.iter() {
for interval in &merged {
for l2t in run.get_overlapping_cmp(interval, cmp) {
table_ids.insert(Table::id(l2t));
}
}
}
}
#[expect(
clippy::cast_possible_truncation,
reason = "level index is bounded by level count (7)"
)]
return Choice::Merge(CompactionInput {
table_ids,
dest_level: l2_idx as u8,
canonical_level: 2,
target_size: self.target_size,
});
}
}
#[expect(
clippy::cast_possible_truncation,
reason = "level index is bounded by level count (7, technically 255)"
)]
let choice = CompactionInput {
table_ids,
dest_level: canonical_l1_idx as u8,
canonical_level: 1,
target_size: self.target_size,
};
if target_level_overlapping_table_ids.is_empty() && first_level.is_disjoint() {
return Choice::Move(choice);
}
return Choice::Merge(choice);
}
let next_level_idx = if level_idx_with_highest_score == 0 {
canonical_l1_idx
} else {
level_idx_with_highest_score + 1
};
#[expect(clippy::cast_possible_truncation)]
let (next_level_index, canonical_level) =
(next_level_idx as u8, (next_level_idx - level_shift) as u8);
let Some(level) = version.level(level_idx_with_highest_score) else {
return Choice::DoNothing;
};
let Some(next_level) = version.level(next_level_idx) else {
return Choice::DoNothing;
};
let policy = self.policy(level_idx_with_highest_score, level_shift, last);
let next_tiered = self.policy(next_level_idx, level_shift, last).layout == Layout::Tiered;
let tiered_window = if policy.granularity == Granularity::Window && next_tiered {
match window_over(
level,
state.hidden_set(),
overshoot_bytes,
policy.movement,
self.target_size,
) {
Some(ids) => {
let Some(ids) = with_overlapping_runs(level, ids, state.hidden_set(), cmp)
else {
return Choice::DoNothing;
};
Some(ids)
}
None => None,
}
} else {
None
};
let (table_ids, can_trivial_move) = match (tiered_window, policy.granularity) {
(Some(ids), _) => {
if version.level_is_busy(next_level_idx, state.hidden_set()) {
return Choice::DoNothing;
}
let overlaps = level
.iter()
.flat_map(|run| run.iter())
.filter(|table| ids.contains(&table.id()))
.any(|table| {
next_level
.iter()
.any(|run| !run.get_overlapping_cmp(table.key_range(), cmp).is_empty())
});
(ids, !overlaps)
}
(None, Granularity::Window) if !next_tiered => {
let Some(picked) = pick_minimal_compaction(
level,
next_level,
state.hidden_set(),
overshoot_bytes,
self.target_size,
policy.movement,
cmp,
) else {
return Choice::DoNothing;
};
picked
}
(None, Granularity::Window | Granularity::Level { .. }) => {
if version.level_is_busy(level_idx_with_highest_score, state.hidden_set())
|| version.level_is_busy(next_level_idx, state.hidden_set())
{
return Choice::DoNothing;
}
if policy.granularity
== (Granularity::Level {
through_full_next: true,
})
&& !next_tiered
&& scores
.get(next_level_idx)
.is_some_and(|(score, _)| *score > 1.0)
&& let Some(below) = version.level(next_level_idx + 1)
&& !version.level_is_busy(next_level_idx + 1, state.hidden_set())
&& !below.is_empty()
{
let mut ids = level.list_ids();
ids.extend(next_level.list_ids());
let mut ranges: Vec<_> = level
.iter()
.chain(next_level.iter())
.flat_map(|run| run.iter())
.map(|table| table.key_range().clone())
.collect();
ranges.sort_by(|a, b| cmp.compare(a.min(), b.min()));
for interval in &crate::KeyRange::merge_sorted_cmp(ranges, cmp) {
for run in below.iter() {
ids.extend(
run.get_overlapping_cmp(interval, cmp).iter().map(Table::id),
);
}
}
#[expect(
clippy::cast_possible_truncation,
reason = "level index is bounded by level count (7, technically 255)"
)]
return Choice::Merge(CompactionInput {
table_ids: ids,
dest_level: (next_level_idx + 1) as u8,
canonical_level: canonical_level + 1,
target_size: self.target_size,
});
}
let key_range = level.aggregate_key_range_cmp(cmp);
let overlapping: Vec<TableId> = next_level
.iter()
.flat_map(|run| run.get_overlapping_cmp(&key_range, cmp))
.map(Table::id)
.collect();
let mut ids = level.list_ids();
if !next_tiered {
ids.extend(&overlapping);
}
(ids, overlapping.is_empty())
}
};
let choice = CompactionInput {
table_ids,
dest_level: next_level_index,
canonical_level,
target_size: self.target_size,
};
if can_trivial_move && level.is_disjoint() {
return Choice::Move(choice);
}
Choice::Merge(choice)
}
}