coordinode-lsm-tree 5.12.3

// SPDX-License-Identifier: Apache-2.0
// Copyright (c) 2024-present, fjall-rs
// Copyright (c) 2026-present, Dmitry Prudnikov

#[cfg(test)]
#[allow(
    clippy::unwrap_used,
    clippy::indexing_slicing,
    clippy::useless_vec,
    clippy::unnecessary_map_or,
    reason = "test code"
)]
mod test;

use super::{Choice, CompactionStrategy, Input as CompactionInput};
use crate::{
    HashSet, TableId,
    compaction::state::{CompactionState, hidden_set::HiddenSet},
    config::Config,
    slice_windows::{GrowingWindowsExt, ShrinkingWindowsExt},
    table::{Table, util::aggregate_run_key_range},
    version::{Level, Version, run::Ranged},
};
#[cfg(not(feature = "std"))]
use alloc::vec::Vec;

/// A merge candidate's cost: the bytes it moves down a level (`promoted`) and
/// every byte it reads and rewrites to do so (`total`, the promoted bytes
/// plus those it pulls in from the level below).
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
struct MergeCost {
    promoted: u64,
    total: u64,
}

impl MergeCost {
    /// Whether `self` rewrites fewer bytes per promoted byte than `other`.
    ///
    /// Cross-multiplied in `u128`: two `u64` byte counts multiply into at
    /// most 128 bits, so the comparison is exact for every size a level can
    /// hold, where dividing first would lose the low bits.
    fn cheaper_than(self, other: Self) -> bool {
        u128::from(self.total) * u128::from(other.promoted)
            < u128::from(other.total) * u128::from(self.promoted)
    }
}

/// Streams merge candidates and keeps the one that pays down a level's debt at
/// the least rewriting per promoted byte.
///
/// Ranking by the total input alone prefers a small merge that promotes little
/// into a dense level below, which has to be repeated many times to retire the
/// same debt; the ratio ranks how much each promoted byte costs. Among the
/// candidates that promote no more than `overshoot + slack`, the cheapest one
/// wins, so a level is not asked to push down far more than it owes; when none
/// is that small, the cheapest overall still makes progress. Ties go to the
/// earliest candidate, so the choice is the same on every call.
///
/// Only the two leaders are held, whatever the number of candidates offered.
struct MergeRanking<T> {
    bound: u64,
    offered: usize,
    /// The cheapest candidate within the bound, with its offer index.
    within: Option<(usize, MergeCost, T)>,
    /// The cheapest candidate overall, with its offer index.
    any: Option<(usize, MergeCost, T)>,
    /// The cost of the cheapest candidate after `any`.
    second: Option<MergeCost>,
}

impl<T: Copy> MergeRanking<T> {
    fn new(overshoot: u64, slack: u64) -> Self {
        Self {
            // A bound past `u64::MAX` admits every candidate, which is what
            // clamping to `u64::MAX` means: no promoted byte count exceeds it.
            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;
        self.offered += 1;
        if cost.promoted <= self.bound
            && self
                .within
                .as_ref()
                .is_none_or(|(_, best, _)| cost.cheaper_than(*best))
        {
            self.within = Some((index, cost, candidate));
        }
        match &self.any {
            Some((_, best, _)) if !cost.cheaper_than(*best) => {
                if self.second.is_none_or(|second| cost.cheaper_than(second)) {
                    self.second = Some(cost);
                }
            }
            previous => {
                self.second = previous.as_ref().map(|(_, best, _)| *best);
                self.any = Some((index, cost, candidate));
            }
        }
    }

    /// The chosen candidate with its offer index and cost, and the cost of the
    /// cheapest candidate not chosen.
    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
    }
}

/// Tries to find the most optimal compaction set from one level into the other.
///
/// Scans all runs in both levels to handle transient multi-run states from
/// multi-level compaction (#108). See #122 Part 3.
fn pick_minimal_compaction(
    curr_level: &Level,
    next_level: &Level,
    hidden_set: &HiddenSet,
    overshoot: u64,
    table_base_size: u64,
    promotion_slack: u64,
    cmp: &dyn crate::comparator::UserComparator,
) -> Option<(HashSet<TableId>, bool)> {
    // NOTE: Find largest trivial move (if it exists)
    // Check all runs in curr_level for a window that doesn't overlap ANY run
    // in next_level.
    for curr_run in curr_level.iter() {
        if let Some(window) = curr_run.shrinking_windows().find(|window| {
            if hidden_set.is_blocked(window.iter().map(Table::id)) {
                return false;
            }

            if next_level.is_empty() {
                return true;
            }

            let key_range = aggregate_run_key_range(window);

            // Must not overlap ANY run in the next level
            next_level
                .iter()
                .all(|run| run.get_overlapping_cmp(&key_range, cmp).is_empty())
        }) {
            let ids = window.iter().map(Table::id).collect();
            return Some((ids, true));
        }
    }

    // NOTE: Look for merges
    // Iterate windows across all runs in next_level, pull in from all runs
    // in curr_level.
    if next_level.is_empty() {
        return None;
    }

    // The tables of `curr_level` a merge of `window` promotes, from all its runs.
    let pull_in = |window: &[Table]| {
        let key_range = aggregate_run_key_range(window);
        curr_level
            .iter()
            .flat_map(|run| run.get_contained_cmp(&key_range, cmp))
            .collect::<Vec<&Table>>()
    };

    // Candidates are ranked as they are enumerated, holding only a window and
    // its cost for the leaders; the winner's pull-in is rebuilt at the end.
    let mut ranking = MergeRanking::new(overshoot, promotion_slack);
    let windows = next_level.iter().flat_map(|run| {
        // Cap per-run windows at 50x table_base_size. take_while is safe
        // here because growing_windows within a single run are monotonically
        // increasing in size — once one exceeds the cap, all subsequent will too.
        run.growing_windows().take_while(|window| {
            let size = window.iter().map(Table::file_size).sum::<u64>();
            size <= (50 * table_base_size)
        })
    });
    for window in windows {
        if hidden_set.is_blocked(window.iter().map(Table::id)) {
            continue;
        }
        let curr_level_pull_in = pull_in(window);
        let curr_level_size = curr_level_pull_in
            .iter()
            .map(|t| Table::file_size(t))
            .sum::<u64>();
        if curr_level_size == 0
            || hidden_set.is_blocked(curr_level_pull_in.iter().map(|t| Table::id(t)))
        {
            continue;
        }
        let next_level_size = window.iter().map(Table::file_size).sum::<u64>();
        ranking.offer(
            MergeCost {
                promoted: curr_level_size,
                total: curr_level_size + next_level_size,
            },
            window,
        );
    }

    let offered = ranking.offered();
    let ((chosen, cost, window), runner_up) = ranking.finish()?;
    log::debug!(
        "leveled: merge candidate {chosen} of {offered} chosen at {cost:?}, runner-up \
         {runner_up:?} (overshoot {overshoot}, slack {promotion_slack})",
    );
    let mut ids: HashSet<_> = window.iter().map(Table::id).collect();
    ids.extend(pull_in(window).iter().map(|t| Table::id(t)));
    Some((ids, false))
}

#[doc(hidden)]
pub const NAME: &str = "LeveledCompaction";

/// Leveled compaction strategy (LCS)
///
/// When a level reaches some threshold size, parts of it are merged into overlapping tables in the next level.
///
/// Each level Ln for n >= 2 can have up to `level_base_size * ratio^(n - 1)` tables.
///
/// LCS suffers from comparatively high write amplification, but has decent read amplification and great space amplification (~1.1x).
///
/// LCS is the recommended compaction strategy to use.
///
/// More info here: <https://fjall-rs.github.io/post/lsm-leveling/>
#[derive(Clone)]
pub struct Strategy {
    l0_threshold: u8,

    /// The target table size as disk (possibly compressed).
    target_size: u64,

    /// Size ratio between levels of the LSM tree (a.k.a fanout, growth rate)
    level_ratio_policy: Vec<f32>,

    /// When true, dynamically sizes levels based on the actual data in
    /// the last non-empty level, reducing space amplification to ~1.1x.
    ///
    /// Same as `level_compaction_dynamic_level_bytes` in `RocksDB`.
    ///
    /// Default = false (static leveling).
    dynamic: bool,

    /// When true, enables multi-level compaction: if L0→L1 is chosen but
    /// L1 is already oversized, compacts L0+L1→L2 directly in one pass
    /// to avoid a write-then-rewrite cycle.
    ///
    /// Default = false.
    multi_level: bool,
}

impl Default for Strategy {
    fn default() -> Self {
        Self {
            l0_threshold: 4,
            target_size:/* 64 MiB */ 64 * 1_024 * 1_024,
            level_ratio_policy: vec![10.0],
            dynamic: false,
            multi_level: false,
        }
    }
}

impl Strategy {
    /// Sets the growth ratio between levels.
    ///
    /// Same as `set_max_bytes_for_level_multiplier` in `RocksDB`.
    ///
    /// Default = [10.0]
    #[must_use]
    pub fn with_level_ratio_policy(mut self, policy: Vec<f32>) -> Self {
        self.level_ratio_policy = policy;
        self
    }

    /// Sets the L0 threshold.
    ///
    /// When the number of tables in L0 reaches this threshold,
    /// they are merged into L1.
    ///
    /// Same as `level0_file_num_compaction_trigger` in `RocksDB`.
    ///
    /// Default = 4
    #[must_use]
    pub fn with_l0_threshold(mut self, threshold: u8) -> Self {
        self.l0_threshold = threshold;
        self
    }

    /// Sets the table target size on disk (possibly compressed).
    ///
    /// Same as `target_file_size_base` in `RocksDB`.
    ///
    /// Default = 64 MiB
    #[must_use]
    pub fn with_table_target_size(mut self, bytes: u64) -> Self {
        self.target_size = bytes;
        self
    }

    /// Enables dynamic level sizing based on actual data in the last level.
    ///
    /// When enabled, level target sizes are computed top-down from the actual
    /// size of the last non-empty level, divided by the ratio at each step.
    /// This reduces space amplification to ~1.1x while keeping write
    /// amplification comparable to static leveling.
    ///
    /// Same as `level_compaction_dynamic_level_bytes` in `RocksDB`.
    ///
    /// Default = false
    #[must_use]
    pub fn with_dynamic_level_bytes(mut self, enabled: bool) -> Self {
        self.dynamic = enabled;
        self
    }

    /// Enables multi-level compaction optimization.
    ///
    /// When L0→L1 compaction is selected but L1 already exceeds its target
    /// size, this option allows compacting L0+L1 directly into L2 in one
    /// pass, avoiding the write-then-rewrite cycle that would otherwise
    /// occur.
    ///
    /// Default = false
    #[must_use]
    pub fn with_multi_level(mut self, enabled: bool) -> Self {
        self.multi_level = enabled;
        self
    }

    /// Calculates the size of L1.
    fn level_base_size(&self) -> u64 {
        self.target_size * u64::from(self.l0_threshold)
    }

    /// Calculates the level target size.
    ///
    /// L1 = `level_base_size`
    ///
    /// L2 = `level_base_size * ratio`
    ///
    /// L3 = `level_base_size * ratio * ratio`
    ///
    /// ...
    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 {
            // u64::from(self.target_size)
            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;

            // NOTE: Minus 2 because |{L0, L1}|
            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
            }
        }
    }

    /// Computes level target sizes for all 7 levels.
    ///
    /// In static mode, uses the standard exponential formula.
    /// In dynamic mode, derives targets from the actual size of the last
    /// non-empty level, dividing backwards by the ratio at each step.
    /// Falls back to static mode when the tree is small (dynamic L1 target
    /// would be less than `level_base_size`).
    fn compute_level_targets(
        &self,
        version: &Version,
        level_shift: usize,
        state: &CompactionState,
    ) -> [u64; 7] {
        let mut targets = [u64::MAX; 7];

        // L0 target is not size-based (it's count-based), so leave at MAX
        targets[0] = u64::MAX;

        if self.dynamic {
            // Find the last non-empty level (Lmax) and its actual size.
            // Iterate forward and keep track of the last non-empty level
            // since iter_levels() does not support DoubleEndedIterator.
            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();

                // Work backwards from Lmax
                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;

                // Only backfill down to the effective L1 (accounting for
                // level_shift), not to physical level 1, so we don't
                // overwrite slots below the shifted canonical L1.
                let dynamic_l1_idx = level_shift + 1;

                for idx in (dynamic_l1_idx..lmax_idx).rev() {
                    let canonical = idx - level_shift;
                    // In the forward formula, target(k+1)/target(k) = ratio[k-1],
                    // so backwards: target(k) = target(k+1) / ratio[k-1]
                    // Clamp-to-zero: the ratio for the first dynamic level reuses
                    // index 0 rather than underflowing to a phantom level.
                    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)
                            }),
                    );

                    // Guard against invalid ratios (zero, negative, NaN, infinite).
                    // Fall back to static targets instead of leaving partial
                    // dynamic targets with u64::MAX in lower-level slots.
                    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;
                    }
                }

                // Fallback: if dynamic L1 target is too small, use static.
                // Compare the shifted L1 slot, not physical slot 1.
                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)
    }

    /// Computes static (exponential) level targets.
    fn compute_static_targets(&self, level_shift: usize) -> [u64; 7] {
        let mut targets = [u64::MAX; 7];

        for (idx, slot) in targets.iter_mut().enumerate().skip(1) {
            if idx <= level_shift {
                continue; // stays at u64::MAX
            }
            #[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_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 {
        // Read-only snapshot: an empty compaction state (nothing hidden) so the
        // targets reflect the full current footprint; no level shift.
        let targets = self.compute_level_targets(version, 0, &CompactionState::default());
        let mut debt = 0u64;
        for (idx, level) in version.iter_levels().enumerate() {
            // Each term is at most one level's on-disk size, so the running sum is
            // bounded by the tree's total footprint (disk capacity) and cannot
            // overflow u64 — plain addition.
            if idx == 0 {
                // L0 is count-triggered: once it reaches the L0 file threshold its
                // whole size is pending a merge into L1. Use the table (file)
                // count, matching the `choose` trigger; `iter().count()` would
                // count runs, undercounting a multi-table L0 run.
                if level.table_count() >= usize::from(self.l0_threshold) {
                    debt += level.size();
                }
            } else if let Some(&target) = targets.get(idx) {
                // max(0, size - target): the bytes above this level's target are
                // the work that must be rewritten downward.
                debt += level.size().saturating_sub(target);
            }
        }
        debt
    }

    #[expect(clippy::too_many_lines)]
    fn choose(&self, version: &Version, config: &Config, state: &CompactionState) -> Choice {
        // Tie the invariant to the caller-supplied `Config::level_count`
        // rather than a literal so a tree explicitly configured with a
        // non-default level count does not panic here. The assertion still
        // fires when a `Version` carries a different count than the
        // `Config` driving this compactor invocation — that mismatch is a
        // real bug (Version and Config disagree).
        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();

        // Trivial move into Lmax
        '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()
                }) {
                    // There are intermediary levels with data, cannot trivially move to Lmax
                    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,
                    });
                }
            }
        }

        // Find the level that corresponds to L1
        #[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;

        // Number of levels we have to shift to get from the actual level idx to the canonical
        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())
                        // NOTE: Take bytes that are already being compacted into account,
                        // otherwise we may be overcompensating
                        .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
                });

            // Move up L1 one level if all current levels are at capacity
            if need_new_l1 {
                canonical_l1_idx -= 1;
                level_shift -= 1;
            }
        }

        // Trivial move into L1
        'trivial: {
            let first_level = version.l0();
            let target_level_idx = first_non_empty_level.min(canonical_l1_idx);

            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);

                // Get overlapping tables in next level
                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,
                    });
                }
            }
        }

        // Intra-L0 compaction: merge multiple L0 runs into a single run within L0
        // when table count is below the L0→L1 threshold
        {
            let first_level = version.l0();

            if first_level.run_count() > 1
                && first_level.table_count() < usize::from(self.l0_threshold)
                && !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,
                });
            }
        }

        // Compute level targets (supports both static and dynamic modes)
        let level_targets = self.compute_level_targets(version, level_shift, state);

        // Scoring
        let mut scores = [(/* score */ 0.0, /* overshoot */ 0u64); 7];

        {
            // TODO(weak-tombstone-rewrite): incorporate `Table::weak_tombstone_count` and
            // `Table::weak_tombstone_reclaimable` when computing level scores so rewrite
            // decisions can prioritize tables that would free the most reclaimable values.

            // Score first level
            let first_level = version.l0();

            if first_level.table_count() >= usize::from(self.l0_threshold) {
                #[expect(
                    clippy::cast_precision_loss,
                    reason = "precision loss is acceptable for scoring calculations"
                )]
                let ratio = (first_level.table_count() as f64) / f64::from(self.l0_threshold);
                scores[0] = (ratio, 0);
            }

            // Score L1+
            for (idx, level) in version.iter_levels().enumerate().skip(1) {
                if level.is_empty() {
                    continue;
                }

                let level_size = level
                    .iter()
                    .flat_map(|x| x.iter())
                    // NOTE: Take bytes that are already being compacted into account,
                    // otherwise we may be overcompensating
                    .filter(|x| !state.hidden_set().is_hidden(x.id()))
                    .map(Table::file_size)
                    .sum::<u64>();

                // NOTE: We check for level length above
                #[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);

                    // NOTE: Force a trivial move
                    if version
                        .level(idx + 1)
                        .is_some_and(|next_level| next_level.is_empty())
                    {
                        scores[idx] = (99.99, 999);
                    }
                }
            }

            // NOTE: Never score Lmax
            {
                scores[6] = (0.0, 0);
            }
        }

        // Choose compaction
        #[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;
        }

        // We choose L0->L1 compaction
        if level_idx_with_highest_score == 0 {
            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);

            // Get overlapping tables in next level
            let target_level_overlapping_table_ids: Vec<_> = target_level
                .iter()
                .flat_map(|run| run.get_overlapping_cmp(&key_range, cmp))
                .map(Table::id)
                .collect();

            table_ids.extend(&target_level_overlapping_table_ids);

            // Multi-level compaction: if L1 is already oversized, skip it
            // and compact L0+L1 directly into L2 in one pass.
            // NOTE: Currently triggers on pre-compaction L1 score. A future
            // improvement could use projected post-compaction bytes to also
            // catch cases where L1 is close to its target and this batch
            // would push it over.
            if self.multi_level {
                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)
                {
                    // Include ALL L1 tables (we're emptying L1 into L2)
                    table_ids.extend(target_level.list_ids());

                    // Include overlapping L2 tables — query per merged
                    // interval instead of one coarse aggregate (#72).
                    // An aggregate across disjoint tables (e.g. [a,d] and
                    // [x,z] → [a,z]) covers gaps and pulls in L2 tables
                    // that don't actually overlap any input table.
                    //
                    // Merge input key ranges into disjoint intervals first
                    // to reduce redundant queries when L0 tables overlap
                    // (#122 Part 2). Sort by comparator-min, then coalesce.
                    {
                        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);
        }

        // We choose L1+ compaction instead

        // NOTE: Level count is 255 max
        #[expect(clippy::cast_possible_truncation)]
        let curr_level_index = level_idx_with_highest_score as u8;

        let next_level_index = curr_level_index + 1;

        let Some(level) = version.level(level_idx_with_highest_score) else {
            return Choice::DoNothing;
        };

        let Some(next_level) = version.level(next_level_index as usize) else {
            return Choice::DoNothing;
        };

        // pick_minimal_compaction scans all runs in both levels, handling
        // transient multi-run states from multi-level compaction (#108, #122).
        let Some((table_ids, can_trivial_move)) = pick_minimal_compaction(
            level,
            next_level,
            state.hidden_set(),
            overshoot_bytes,
            self.target_size,
            // A merge may promote up to one target table past the overshoot,
            // the smallest step a merge can take, and still be preferred.
            self.target_size,
            cmp,
        ) else {
            return Choice::DoNothing;
        };

        #[expect(
            clippy::cast_possible_truncation,
            reason = "level shift is bounded by level count (7, technically 255)"
        )]
        let choice = CompactionInput {
            table_ids,
            dest_level: next_level_index,
            canonical_level: next_level_index - (level_shift as u8),
            target_size: self.target_size,
        };

        if can_trivial_move && level.is_disjoint() {
            return Choice::Move(choice);
        }
        Choice::Merge(choice)
    }
}