coordinode-lsm-tree 7.4.0

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

use super::stream::RangeTombstoneUse;
use super::{CompactionAction, CompactionResult, CompactionStrategy, Input as CompactionPayload};
use crate::time::Instant;
use crate::tree::inner::{CompactionGuard, VersionsReadGuard};
use crate::{
    BlobFile, Config, HashSet, InternalValue, SeqNo, SequenceNumberCounter,
    SharedSequenceNumberGenerator, Table, TableId, UserKey,
    compaction::{
        Choice,
        filter::{Context, FilterBlobOutput, StreamFilterAdapter, TransformCounters},
        flavour::{FreshValues, RelocatingCompaction, StandardCompaction},
        state::CompactionState,
        stream::CompactionStream,
    },
    file::BLOBS_FOLDER,
    merge::Merger,
    run_scanner::RunScanner,
    stop_signal::StopSignal,
    tree::inner::TreeId,
    version::{Run, SuperVersions, Version},
    vlog::{BlobFileMergeScanner, BlobFileScanner, BlobFileWriter},
};
// `BlobFileId` only appears in the std-only parallel sub-compaction + tight-space
// relocation paths; gate the import so the no_std build does not see it unused.
#[cfg(feature = "std")]
use crate::vlog::BlobFileId;
use alloc::sync::Arc;
#[cfg(not(feature = "std"))]
use alloc::{boxed::Box, vec::Vec};
// no-std: spin mirrors parking_lot's Mutex/RwLock API without an allocator.
#[cfg(feature = "std")]
use parking_lot::{Mutex, RwLock};
#[cfg(not(feature = "std"))]
use spin::{Mutex, RwLock};

#[cfg(feature = "metrics")]
use crate::metrics::Metrics;

pub type CompactionReader<'a> = Box<dyn Iterator<Item = crate::Result<InternalValue>> + 'a>;

/// Minimum total input size for a compaction to be split into parallel
/// sub-compactions. Below this the per-thread setup + the extra output tables
/// (one per sub-range) cost more than the parallelism buys, so the compaction
/// stays single-threaded (which also keeps small/test compactions producing a
/// single merged table). Default for [`Config::subcompaction_min_bytes`].
#[cfg(feature = "std")]
pub const SUBCOMPACTION_MIN_INPUT_BYTES: u64 = 8 * 1024 * 1024;

/// Compaction options
#[derive(Clone)]
pub struct Options {
    pub tree_id: TreeId,

    pub global_seqno: SharedSequenceNumberGenerator,

    pub visible_seqno: SharedSequenceNumberGenerator,

    /// The tree's pooled table files, which the outputs are written into.
    pub(crate) table_files: Arc<crate::table::file_pool::TableFilePools>,

    pub blob_file_id_generator: SequenceNumberCounter,

    /// Configuration of tree.
    pub config: Arc<Config>,

    pub version_history: Arc<RwLock<SuperVersions>>,

    /// Compaction strategy to use.
    pub strategy: Arc<dyn CompactionStrategy>,

    /// Stop signal to interrupt a compaction worker in case
    /// the tree is dropped.
    pub stop_signal: StopSignal,

    /// The GC watermark: a version only snapshots below it could see may be
    /// collected, and a run that collects one raises the retention floor to
    /// just below it (never lowering a higher floor). See
    /// [`AbstractTree::major_compact`](crate::AbstractTree::major_compact).
    pub gc_watermark: crate::SeqNo,

    pub compaction_state: Arc<Mutex<CompactionState>>,

    /// The tree-wide deletion pause, installed on every compaction-produced table
    /// before it becomes visible. A flush registers it via `register_tables`, but
    /// compaction installs its outputs directly (`install_merge` / tight-space),
    /// so it must be threaded here — without it a Page-ECC output's in-place heal
    /// skips the checkpoint mutation window and can race a checkpoint hard-link.
    pub deletion_pause: Arc<crate::deletion_pause::DeletionPause>,

    /// The tree-wide ECC heal-hint sink, installed on every
    /// compaction-produced table alongside the deletion pause and for the
    /// same reason: a flush registers it via `register_tables`, but
    /// compaction installs its outputs directly. Without it a
    /// confirmed-persistent ECC correction while reading such an output
    /// corrects the block in memory yet can never queue the SST for a
    /// healing rewrite, so the bitrot stays on disk indefinitely.
    pub heal_hints: Arc<crate::heal_hints::HealHints>,

    /// The tree's background file deleter, installed on every compaction
    /// output as a flush installs it on its own.
    #[cfg(feature = "std")]
    pub background_deleter: Arc<crate::BackgroundDeleter>,

    /// Whether the tree's filters exceed their budget, which a compaction's
    /// filter plan reads and updates.
    pub filter_budget: Arc<crate::filter_budget::FilterBudget>,

    /// Shared handle to the live runtime config. Compaction loads
    /// a fresh snapshot via [`crate::runtime_config::handle::RuntimeConfigHandle::load_full`]
    /// each time it writes the manifest, so toggles applied via
    /// [`crate::Tree::update_runtime_config`] take effect on the
    /// next compaction cycle without restart.
    pub runtime_config: Arc<crate::runtime_config::handle::RuntimeConfigHandle>,

    /// Optional per-tree encryption provider, threaded into manifest
    /// writes so compaction-driven version commits inherit the same
    /// AEAD pipeline the data blocks use.
    pub encryption: Option<Arc<dyn crate::encryption::EncryptionProvider>>,

    /// The tree's compaction I/O rate limiter, shared by every compaction it
    /// runs. Built once at open from [`Config::compaction_rate_limit`]; a
    /// limit of `0` makes every request immediate (no throttling). Only the
    /// compaction merge loop calls it, so flush and user reads are never
    /// throttled.
    pub rate_limiter: Arc<crate::rate_limiter::RateLimiter>,

    /// The pool the tree's compactions reserve the room for their outputs
    /// in, which other trees may share.
    pub(crate) capacity: Arc<crate::CapacityPool>,

    /// The files this run has created that no installed version names yet,
    /// removed by [`do_compaction`] when the run returns. [`do_compaction`]
    /// gives each run its own; the clones a run hands its sub-compactions
    /// share it.
    pub(crate) outputs: super::output_ledger::OutputLedger,

    /// The work this run did, against the estimate it was planned against.
    /// [`do_compaction`] gives each run its own, as it does `outputs`.
    pub(crate) work: Arc<super::work::WorkLedger>,

    #[cfg(feature = "metrics")]
    pub metrics: Arc<Metrics>,
}

impl Options {
    pub fn from_tree(tree: &crate::Tree, strategy: Arc<dyn CompactionStrategy>) -> Self {
        Self {
            global_seqno: tree.config.seqno.clone(),
            visible_seqno: tree.config.visible_seqno.clone(),
            tree_id: tree.id,
            table_files: Arc::clone(&tree.table_files),
            blob_file_id_generator: tree.blob_file_id_counter.clone(),
            config: tree.config.clone(),
            version_history: tree.version_history.clone(),
            stop_signal: tree.stop_signal.clone(),
            strategy,
            gc_watermark: 0,

            compaction_state: tree.compaction_state.clone(),
            deletion_pause: tree.deletion_pause.clone(),
            heal_hints: tree.heal_hints.clone(),
            #[cfg(feature = "std")]
            background_deleter: Arc::clone(&tree.background_deleter),
            filter_budget: tree.filter_budget.clone(),
            runtime_config: tree.runtime_config.clone(),
            encryption: tree.config.encryption.clone(),
            // The tree's one budget, not a fresh one: see
            // `TreeInner::compaction_rate_limiter`.
            rate_limiter: Arc::clone(&tree.compaction_rate_limiter),
            capacity: Arc::clone(&tree.compaction_capacity),
            outputs: super::output_ledger::OutputLedger::default(),
            work: Arc::default(),

            #[cfg(feature = "metrics")]
            metrics: tree.metrics.clone(),
        }
    }
}

/// Runs compaction task.
///
/// This will block until the compactor is fully finished.
pub fn do_compaction(opts: &Options) -> crate::Result<CompactionResult> {
    // Ledgers of this run's own: the caller may share its options with runs
    // that overlap this one, and their outputs and work are theirs.
    let opts = &Options {
        outputs: super::output_ledger::OutputLedger::default(),
        work: Arc::default(),
        ..opts.clone()
    };
    // Outputs an earlier run kept go first, once superseded, so this run is
    // not checked against quota they no longer need.
    remove_superseded_outputs(opts);
    let result = run_compaction(opts);
    // A run its room stopped declined, as one the gate skips does: its
    // inputs are as they were, and only its outputs go, below.
    let result = match result {
        Err(e) if opts.outputs.out_of_room() => {
            log::info!("Compaction stopped: its outputs outgrew the room left for them ({e})");
            Ok(CompactionResult::nothing())
        }
        result => result,
    };
    // Every file the run created that it did not install goes now, whichever
    // writer made it and wherever the run stopped: the handles on them
    // dropped as the run returned. The run's room goes back once they are
    // gone. An install whose outcome is unknown may still name them after a
    // power loss, so they stay until a later install supersedes it.
    if matches!(result, Err(crate::Error::InstallOutcomeUnknown(_))) {
        let kept = opts.outputs.keep_uninstalled();
        opts.version_history.write().keep_outputs(kept);
    } else {
        opts.outputs
            .remove_uninstalled(opts.tree_id, opts.config.descriptor_table.as_deref());
    }
    // This run's install may be the rotation that superseded kept outputs,
    // and another install may have rotated since this run's failed.
    remove_superseded_outputs(opts);
    // Charged whether or not the run succeeded: a tight-space pass that fails
    // after installing slices leaves them installed, and what they read and
    // wrote was done.
    let work = opts.work.report();
    #[cfg(feature = "metrics")]
    opts.metrics.charge_compaction_work(&work);
    let mut result = result?;
    result.work = work;
    Ok(result)
}

/// Removes the outputs kept after an install of unknown outcome that a
/// rotation has superseded since, and gives their quota back.
fn remove_superseded_outputs(opts: &Options) {
    if !opts.version_history.read().holds_kept_outputs() {
        return;
    }
    let superseded = opts.version_history.write().take_superseded_outputs();
    for kept in superseded {
        kept.remove(opts.tree_id, opts.config.descriptor_table.as_deref());
    }
}

fn run_compaction(opts: &Options) -> crate::Result<CompactionResult> {
    let compaction_state = opts.compaction_state.lock();
    let reclaimed = drop_covered_tables(&compaction_state, opts)?;
    // The strategy's action, with every table the run took in.
    run_strategy(compaction_state, opts).map(|result| match reclaimed {
        Some(reclaimed) if result.action == CompactionAction::Nothing => reclaimed,
        Some(reclaimed) => CompactionResult {
            tables_in: result.tables_in + reclaimed.tables_in,
            ..result
        },
        None => result,
    })
}

/// Drops, before the strategy chooses, every table a range tombstone deletes
/// for every reader at the watermark. Unlinking reads nothing, while a merge
/// that picked those tables first would read and rewrite them only to discard
/// every row.
fn drop_covered_tables(
    compaction_state: &CompactionGuard<'_>,
    opts: &Options,
) -> crate::Result<Option<CompactionResult>> {
    // A clone of the version, so the scan runs without the history lock.
    let version = opts
        .version_history
        .read()
        .latest_version_ref()
        .version
        .clone();
    let covered = super::covered::covered_tables(
        &version,
        opts.gc_watermark,
        compaction_state.hidden_set(),
        opts.config.comparator.as_ref(),
    );
    let Some((ids, effect)) = covered else {
        return Ok(None);
    };
    drop_tables(compaction_state, opts, &ids, effect).map(Some)
}

fn run_strategy(
    mut compaction_state: CompactionGuard<'_>,
    opts: &Options,
) -> crate::Result<CompactionResult> {
    let version_history_lock = opts.version_history.read();

    // A strategy that drops tables counts what the drop frees, and a blob file
    // a memtable row references outlives its last table: the choice sees the
    // memtables as of now. A tree without blob files reads none.
    let latest = version_history_lock.latest_version_ref();
    compaction_state.set_memtable_blob_files(if latest.version.blob_files.len() > 0 {
        latest.memtable_blob_files()
    } else {
        crate::HashSet::default()
    });

    let start = Instant::now();
    log::trace!(
        "Consulting compaction strategy {:?}",
        opts.strategy.get_name(),
    );
    // Structural decision plus the runtime-config-driven density-rewrite
    // fallback (applied here, where the live runtime config lives, so the
    // structural strategies stay free of it). The snapshot is read for the
    // choice only; the writers take their own, which is the one that must stay
    // held until their outputs are installed.
    let choice = super::choose_with_density_rewrite(
        &*opts.strategy,
        &version_history_lock.latest_version_ref().version,
        &opts.config,
        &opts.runtime_config.load(),
        &compaction_state,
    );

    log::debug!("Compaction choice: {choice:?} in {:?}", start.elapsed());

    match choice {
        Choice::Merge(payload) => {
            gated_merge(compaction_state, version_history_lock, opts, &payload)
        }
        Choice::Move(payload) => {
            // Checked before the cross-folder conversion below, not after: that
            // conversion rewrites only the selected inputs, so it lands them on
            // the same level under the same overlapping table and inherits the
            // placement rather than fixing it. The placement is what is wrong,
            // so neither form of this task may run. Refusing matches what this
            // path already does for a strategy that hands over hidden tables.
            if !nothing_the_move_passes_under_overlaps(
                &version_history_lock.latest_version_ref().version,
                &payload,
                opts.config.comparator.as_ref(),
            ) {
                log::warn!(
                    "Compaction task created by {:?} would move tables beneath an overlapping table, hiding the versions it moved - declining to run it",
                    opts.strategy.get_name(),
                );
                return Ok(CompactionResult::nothing());
            }

            // Cross-folder trivial moves are not possible — the file must be
            // rewritten to end up in the correct storage tier directory.
            // This applies even when both folders are on the same filesystem,
            // because rename() across tiered paths would break the routing
            // invariant (table path = f(level)).
            if opts.config.level_routes.is_some() {
                let (dst_folder, _) = opts.config.tables_folder_for_level(payload.dest_level);
                let version = &version_history_lock.latest_version_ref().version;
                // Check actual on-disk table paths (not configured routing) to
                // handle tables that may have been recovered from a different
                // tier after route reconfiguration.
                let cross_folder = version
                    .iter_levels()
                    .flat_map(|level| level.iter())
                    .flat_map(|run| run.iter())
                    .filter(|t| payload.table_ids.contains(&t.id()))
                    .any(|t| t.path.parent() != Some(dst_folder.as_path()));
                if cross_folder {
                    log::debug!("Converting trivial move to merge: cross-folder level routing");
                    return gated_merge(compaction_state, version_history_lock, opts, &payload);
                }
            }

            // A move carries its tables' blob layout down unchanged, and a
            // table at the bottom level may never be merged again. When locality
            // relocation would rewrite some of their blob files, the same inputs
            // are merged instead, within the same budget and space check.
            if moves_into_locality_relocation(
                &version_history_lock.latest_version_ref().version,
                opts,
                &payload,
            )? {
                log::debug!("Converting trivial move to merge: blob locality relocation");
                return gated_merge(compaction_state, version_history_lock, opts, &payload);
            }

            drop(version_history_lock);

            move_tables(&compaction_state, opts, &payload)
        }
        Choice::Drop(payload) => {
            drop(version_history_lock);

            let ids = payload.into_iter().collect::<Vec<_>>();
            // Whole tables go (FIFO / TTL eviction): the rows every older
            // snapshot saw in them are gone, so no snapshot up to this install
            // is servable.
            drop_tables(
                &compaction_state,
                opts,
                &ids,
                crate::version::RetentionEffect::DropsData,
            )
        }
        Choice::DoNothing => {
            log::trace!("Compactor chose to do nothing");
            Ok(CompactionResult::nothing())
        }
    }
}

/// Where a merge stands once it asked for the room its outputs take.
enum SpaceGate {
    /// Run the merge as chosen, in the room reserved for it, which is the
    /// bytes it writes into its level's folder and into the blob folder.
    Run(super::capacity::Reservation, (u64, u64)),
    /// Run a smaller fitting subset (a minimal reclaiming merge) instead of the
    /// chosen one, which did not fit, in the room reserved for it.
    Narrowed(CompactionPayload, super::capacity::Reservation, (u64, u64)),
    /// No fitting subset exists; skip this cycle. The truly-too-big single merge
    /// is the opt-in tight-space mode's domain; meanwhile `Drop` / `Move` cycles
    /// still reclaim and reorganize. Carries the bytes the merge as chosen
    /// writes into its level's folder and into the blob folder, which bound
    /// what the tight-space pass rewrites.
    Skip((u64, u64)),
}

/// Runs the merge of `payload` in the room its outputs take, a smaller merge
/// that fits, or the tight-space pass, or skips it. A merge a move turned into,
/// to rewrite its tables elsewhere, asks for its room here as any other does.
fn gated_merge(
    compaction_state: CompactionGuard<'_>,
    version_history_lock: VersionsReadGuard<'_>,
    opts: &Options,
    payload: &CompactionPayload,
) -> crate::Result<CompactionResult> {
    // Read once: the budget is asked for the pass that then runs, whatever a
    // runtime update sets in between.
    let tight_space = tight_space_enabled(opts);
    let decision = {
        let super_version = version_history_lock.latest_version();
        // A strategy's choice naming a table the version does not hold is
        // declined before its room is reckoned, as the merge would decline it.
        if payload
            .table_ids
            .iter()
            .any(|&id| super_version.version.get_table(id).is_none())
        {
            log::warn!(
                "Compaction task created by {:?} contained tables not referenced in the level manifest",
                opts.strategy.get_name(),
            );
            return Ok(CompactionResult::nothing());
        }
        let decision = space_gate_for_merge(&super_version.version, opts, payload)?;
        // The merge that runs asks the write budgets of the media it writes
        // to; one that waits for credit drops the room it reserved.
        let budget = match &decision {
            SpaceGate::Run(_, bytes) => budget_gate(
                &super_version.version,
                opts,
                payload.dest_level,
                *bytes,
                Narrowing::None,
            ),
            SpaceGate::Narrowed(narrowed, _, bytes) => budget_gate(
                &super_version.version,
                opts,
                narrowed.dest_level,
                *bytes,
                Narrowing::ForSpace,
            ),
            // The tight-space pass reclaims room on a full volume, as the
            // narrowed merge does: it reserves what it rewrites, at most the
            // merge as chosen, and runs past an empty budget as a space
            // override. Without the pass nothing runs, and nothing is asked.
            SpaceGate::Skip(bytes) if tight_space => budget_gate(
                &super_version.version,
                opts,
                payload.dest_level,
                *bytes,
                Narrowing::ForSpace,
            ),
            SpaceGate::Skip(_) => BudgetGate::Run(Vec::new()),
        };
        let BudgetGate::Run(holds) = budget else {
            log::debug!(
                "Compaction write budget: {}-table merge waits for credit",
                payload.table_ids.len(),
            );
            return Ok(CompactionResult::nothing());
        };
        opts.outputs.hold_budget(holds);
        decision
    };
    match decision {
        SpaceGate::Run(room, _) => {
            opts.outputs.reserve(room);
            merge_tables(compaction_state, version_history_lock, opts, payload)
        }
        SpaceGate::Narrowed(narrowed, room, _) => {
            log::debug!(
                "Compaction space gate: narrowed merge from {} to {} tables to fit free space",
                payload.table_ids.len(),
                narrowed.table_ids.len(),
            );
            opts.outputs.reserve(room);
            merge_tables(compaction_state, version_history_lock, opts, &narrowed)
        }
        SpaceGate::Skip(_) => {
            #[cfg(feature = "std")]
            if tight_space {
                return run_tight_space_compaction(
                    compaction_state,
                    version_history_lock,
                    opts,
                    payload,
                );
            }
            log::info!(
                "Compaction space gate: skipping {}-table merge — free space cannot cover the transient output and no fitting subset exists (opt-in tight-space reclaim handles this)",
                payload.table_ids.len(),
            );
            Ok(CompactionResult::nothing())
        }
    }
}

/// Whether a merge the volume has no room for runs as the tight-space pass,
/// which only a `std` build has.
fn tight_space_enabled(opts: &Options) -> bool {
    cfg!(feature = "std") && opts.runtime_config.load().tight_space_compaction
}

/// Where a merge stands once it asked the write budgets of the media it
/// writes to.
enum BudgetGate {
    /// Run it, holding these reservations, one per budget.
    Run(Vec<crate::write_budget::BudgetHold>),
    /// A budget has no credit for it: it waits.
    Wait,
}

/// Whether the merge asking the budget is the one chosen, or the smaller one
/// the space gate ran instead because the volume had no room for it.
#[derive(Clone, Copy, PartialEq, Eq)]
enum Narrowing {
    None,
    ForSpace,
}

/// Reserves, in the write budget of each medium a merge into `dest_level`
/// writes to, the `sst_bytes` of tables it writes into the level's folder and
/// the `blob_bytes` of blob files it writes into the blob folder: the two
/// added up where both folders' backends name one budget.
///
/// A budget without credit for its part lets the merge run anyway when the
/// caller forces discretionary work on it, when the merge is the reclaiming
/// one the space gate narrowed to on a full volume or the tight-space pass
/// that reclaims room where no merge fits, or when the tree has
/// reached a stop threshold on an axis compaction drains (its L0 runs, its
/// pending compaction bytes): waiting for credit would then leave the volume
/// full or stall the writers. Each is reserved as an override, charged and
/// reported. Otherwise the merge waits, and the reservations it took on other
/// budgets go back. A merge on media without a budget reserves nothing.
fn budget_gate(
    version: &Version,
    opts: &Options,
    dest_level: u8,
    (sst_bytes, blob_bytes): (u64, u64),
    narrowing: Narrowing,
) -> BudgetGate {
    use crate::write_budget::Override;

    let (_, sst_fs) = opts.config.tables_folder_for_level(dest_level);
    // A merge of a tree that separates values may write new blob files though
    // it relocates none: a merge operator's result, a compaction filter's
    // replacement and a locality relocation picked as the merge runs all land
    // in the blob folder. Their medium is asked for every such merge, for
    // nothing reserved, so an overrun one holds the merge back and what lands
    // there is charged through its ledger.
    let may_write_blobs = blob_bytes > 0 || opts.config.kv_separation_opts.is_some();
    let mut demands: Vec<(&Arc<crate::write_budget::WriteBudget>, u64)> = Vec::new();
    for (fs, bytes, writes) in [
        (&*sst_fs, sst_bytes, sst_bytes > 0),
        (&*opts.config.fs, blob_bytes, may_write_blobs),
    ] {
        let Some(budget) = fs.write_budget() else {
            continue;
        };
        if !writes {
            continue;
        }
        if let Some(demand) = demands.iter_mut().find(|(b, _)| Arc::ptr_eq(b, budget)) {
            demand.1 += bytes;
        } else {
            demands.push((budget, bytes));
        }
    }
    let mut holds = Vec::with_capacity(demands.len());
    // Worked out once, by the first budget short of credit.
    let mut stopped = None;
    for (budget, bytes) in demands {
        let hold = budget.reserve_or_override(bytes, || {
            if budget.forced() {
                Some(Override::Forced)
            } else if narrowing == Narrowing::ForSpace {
                Some(Override::Space)
            } else if *stopped.get_or_insert_with(|| at_compaction_stop(version, opts)) {
                Some(Override::Backpressure)
            } else {
                None
            }
        });
        let Some(hold) = hold else {
            return BudgetGate::Wait;
        };
        holds.push(hold);
    }
    BudgetGate::Run(holds)
}

/// Whether the tree is at a stop threshold on an axis compaction drains: its
/// L0 runs or its pending compaction bytes. The memtable axis is drained by
/// flushes, which no budget holds back.
fn at_compaction_stop(version: &Version, opts: &Options) -> bool {
    let thresholds = crate::BackpressureThresholds {
        memtable_slowdown: None,
        memtable_stop: None,
        ..opts.runtime_config.load().backpressure
    };
    if thresholds.is_off() {
        return false;
    }
    let signals = crate::BackpressureSignals {
        l0_runs: version.l0().run_count(),
        pending_compaction_bytes: opts.strategy.pending_compaction_bytes(version),
        unflushed_memtable_bytes: 0,
    };
    crate::Backpressure::compute(&signals, &thresholds) == crate::Backpressure::Stop
}

/// Whether two paths sit on PROVABLY separate free-space pools: both backends
/// report a volume id AND the ids differ.
///
/// The unproven cases (a backend that cannot answer, a route to the same mount)
/// deliberately answer `false` — shared. Every caller uses this to decide
/// whether space freed on one path relieves the other, and guessing
/// "independent" there admits a transient peak straight into `ENOSPC` or a
/// reclaim that lands where nothing needed it.
fn separate_volumes(
    a_fs: &dyn crate::fs::Fs,
    a_path: &crate::path::Path,
    b_fs: &dyn crate::fs::Fs,
    b_path: &crate::path::Path,
) -> bool {
    match (a_fs.volume_id(a_path), b_fs.volume_id(b_path)) {
        (Some(a), Some(b)) => a != b,
        _ => false,
    }
}

/// The bytes a merge into `dest_level` writing `sst_bytes` of tables and
/// `blob_bytes` of relocated blob files takes, folder by folder: the tables in
/// the level's folder, the blob files in the tree's blob folder.
///
/// Each folder belongs to the volume its backend reports. Two folders are
/// counted apart only on volumes proven apart; otherwise their demands add up
/// on one volume, since each would fit its free space alone while together
/// they exhaust it. A level route names a folder, not a mount, so routing
/// alone proves nothing. A tree that relocates no blob files writes nothing
/// to the blob folder, and the pool leaves that folder out of the check.
fn with_merge_demands<T>(
    config: &Config,
    dest_level: u8,
    (sst_bytes, blob_bytes): (u64, u64),
    f: impl FnOnce(&[super::capacity::Demand<'_>]) -> T,
) -> T {
    let (sst_path, sst_fs) = config.tables_folder_for_level(dest_level);
    let blob_dir = config.path.join(BLOBS_FOLDER);
    f(&[
        super::capacity::Demand {
            fs: &*sst_fs,
            path: &sst_path,
            bytes: sst_bytes,
        },
        super::capacity::Demand {
            fs: &*config.fs,
            path: &blob_dir,
            bytes: blob_bytes,
        },
    ])
}

/// Whether a merge into `dest_level` writing `sst_bytes` of tables and
/// `blob_bytes` of relocated blob files fits under `quota_headroom` and the
/// free space of the volumes it writes to, past what the compactions sharing
/// `pool` have still to write, taking the flush floor if it must. The storage
/// statistics ask it of a full compaction, so that what they report is what
/// the gate admits.
pub fn merge_would_fit(
    pool: &crate::CapacityPool,
    tree: TreeId,
    config: &Config,
    quota_headroom: u64,
    (sst_bytes, blob_bytes): (u64, u64),
    dest_level: u8,
) -> bool {
    with_merge_demands(config, dest_level, (sst_bytes, blob_bytes), |demands| {
        pool.would_fit(
            tree,
            demands,
            super::capacity::Check {
                quota_headroom,
                floor: super::capacity::Floor::MayConsume,
            },
        )
    })
}

/// Reserves the room the merge of `payload` takes, narrowing it to a minimal
/// reclaiming subset when it does not fit.
///
/// The room is reserved in the tree's pool whether or not storage admission
/// is on, so that the compactions sharing the pool count it; only admission
/// refuses a merge. A merge is checked against the logical quota for the
/// bytes it writes in all, and against the free space of each volume it
/// writes to past what the other compactions in the pool have still to
/// write, consuming the flush floor if it must, which breaks the deadlock of
/// a merge that frees space and has none to start in.
///
/// The tables are bounded by the bytes of the inputs' files (a tight-space
/// restricted input by its live suffix: the punched prefix is superseded and
/// never rewritten), the blob files by the physical size of the stale ones
/// selected for relocation. A compaction filter or merge operator may write
/// more: the merge takes the excess on as its outputs land, or stops (see
/// [`super::capacity::Reservation::landed`]).
///
/// - It fits (or admission is off) → [`SpaceGate::Run`].
/// - It does not fit, but a run-adjacent pair does → [`SpaceGate::Narrowed`].
///   Candidates are tried in ascending SST size; a larger pair with fewer stale
///   blob rewrites can fit where the smallest does not, so each is reserved
///   for in full (its own relocation set included).
/// - No fitting subset exists → [`SpaceGate::Skip`].
fn space_gate_for_merge(
    version: &Version,
    opts: &Options,
    payload: &CompactionPayload,
) -> crate::Result<SpaceGate> {
    let rc = opts.runtime_config.load_full();
    let check = if rc.storage_admission_check {
        // `max(0, limit - used)`: an operator quota set below the live
        // footprint leaves zero headroom — the clamp-to-zero is the intended
        // min-semantics (not masking). The footprint stat is propagated, not
        // swallowed (an undercounted `used` would overstate the headroom).
        let quota_headroom = match rc.storage_limit_bytes {
            Some(limit) => limit.saturating_sub(crate::storage_stats::compute_used_bytes(version)?),
            None => u64::MAX,
        };
        Some(super::capacity::Check {
            quota_headroom,
            floor: super::capacity::Floor::MayConsume,
        })
    } else {
        None
    };

    // Per-payload demand. Recomputed per payload because narrowing changes both
    // the SST input set AND (via `pick_blob_files_to_rewrite` on the actual
    // payload) the stale blob relocation set, so a narrowed merge can fit where
    // the full one did not. Stat failures propagate (a silent 0 would undercount
    // and admit the very rewrite this gate must block).
    let demand = |p: &CompactionPayload| -> crate::Result<(u64, u64)> {
        let sst_sigma: u64 = p
            .table_ids
            .iter()
            .filter_map(|&id| version.get_table(id))
            .try_fold(0u64, |acc, t| t.live_file_size().map(|size| acc + size))?;
        let blob_sigma: u64 = match &opts.config.kv_separation_opts {
            Some(blob_opts) => pick_blob_files_to_rewrite(&p.table_ids, version, blob_opts)?
                .iter()
                .try_fold(0u64, |acc, bf| bf.physical_size().map(|size| acc + size))?,
            None => 0,
        };
        Ok((sst_sigma, blob_sigma))
    };
    let reserve = |bytes: (u64, u64), dest_level: u8| {
        with_merge_demands(&opts.config, dest_level, bytes, |demands| {
            opts.capacity.reserve(opts.tree_id, demands, check)
        })
    };

    let chosen = demand(payload)?;
    if let Some(room) = reserve(chosen, payload.dest_level) {
        return Ok(SpaceGate::Run(room, chosen));
    }

    // Constrained: try the run-adjacent narrowing candidates in ascending SST
    // size and run the first that fits. Skip only when none fit (a truly-too-big
    // single merge is the opt-in tight-space domain).
    for narrowed in narrow_merge_candidates(version, payload)? {
        let bytes = demand(&narrowed)?;
        if let Some(room) = reserve(bytes, narrowed.dest_level) {
            return Ok(SpaceGate::Narrowed(narrowed, room, bytes));
        }
    }
    Ok(SpaceGate::Skip(chosen))
}

/// The run-adjacent table-pair narrowing candidates for a too-large merge,
/// sorted by combined SST `live_file_size` ascending (the same measure the gate
/// charges, so a restricted input's superseded prefix does not distort the
/// order), each preserving the merge's destination.
///
/// Narrowing is only safe within a single run (key-sorted, non-overlapping
/// tables): a run-adjacent pair is always a valid, reclaiming merge, and the
/// merge stream culls exactly that pair. A cross-run / cross-level payload (e.g.
/// a leveled `Lₙ → Lₙ₊₁` overlap set) cannot be narrowed here without risking an
/// invalid partial merge, so it yields no candidates (the chosen merge is
/// skipped and left to the opt-in tight-space mode). The pair must be adjacent
/// IN THE RUN — a gap would make the merge stream pull the in-between tables
/// too, breaking the size bound.
///
/// The caller tries the candidates in order and runs the first that fits the
/// full space gate: the smallest-SST pair can still fail on its stale blob
/// relocation set or the combined single-volume budget, while a slightly larger
/// pair with fewer blob rewrites fits — ranking by SST size alone and stopping at
/// the first would wrongly skip the merge.
fn narrow_merge_candidates(
    version: &Version,
    payload: &CompactionPayload,
) -> crate::Result<Vec<CompactionPayload>> {
    // The single run that holds every payload table (if any).
    let Some(run) = version
        .iter_levels()
        .flat_map(|level| level.iter())
        .find(|run| {
            payload
                .table_ids
                .iter()
                .all(|id| run.iter().any(|t| t.id() == *id))
        })
    else {
        return Ok(Vec::new());
    };

    let run_tables: Vec<&Table> = run.iter().collect();
    let mut candidates: Vec<(u64, CompactionPayload)> = Vec::new();
    for pair in run_tables.windows(2) {
        let [a, b] = pair else { continue };
        if !payload.table_ids.contains(&a.id()) || !payload.table_ids.contains(&b.id()) {
            continue;
        }
        // Two on-disk file sizes; the sum is bounded by filesystem capacity and
        // cannot overflow u64.
        let combined = a.live_file_size()? + b.live_file_size()?;
        candidates.push((
            combined,
            CompactionPayload {
                table_ids: [a.id(), b.id()].into_iter().collect(),
                dest_level: payload.dest_level,
                canonical_level: payload.canonical_level,
                target_size: payload.target_size,
            },
        ));
    }
    candidates.sort_by_key(|(combined, _)| *combined);
    Ok(candidates.into_iter().map(|(_, p)| p).collect())
}

fn pick_run_indexes(run: &Run<Table>, to_compact: &[TableId]) -> Option<(usize, usize)> {
    let lo = run
        .iter()
        .position(|table| to_compact.contains(&table.id()))?;

    let hi = run
        .iter()
        .rposition(|table| to_compact.contains(&table.id()))?;

    Some((lo, hi))
}

/// The pacer of the run's input reads: it tells the rate limiter how long
/// they take, for a latency backoff on the limiter to steer by, and counts
/// what they return to the run's work. A limiter shared with a verification
/// scan would otherwise keep a rate the scan lowered after the scan ends,
/// since only reads report latency.
fn input_pacer(opts: &Options) -> crate::table::util::Pacer {
    Arc::new(super::work::InputPacer::new(&opts.rate_limiter, &opts.work))
}

/// The work a merge of `tables` into `payload`'s level is expected to cost,
/// relocating `blob_files` and applying the range tombstones of `rt_plan`.
fn merge_estimate(
    opts: &Options,
    version: &Version,
    payload: &CompactionPayload,
    tables: &[Table],
    blob_files: &[BlobFile],
    rt_plan: Option<&(
        Vec<crate::range_tombstone::RangeTombstone>,
        RangeTombstoneUse,
    )>,
) -> super::work::WorkEstimate {
    // The groups and blocks are recorded as the merge records them: a merge
    // relocating blob files writes every row itself.
    let carry = if blob_files.is_empty() {
        super::carry::plan(
            &opts.config,
            &opts.runtime_config.load(),
            payload.canonical_level.into(),
            tables,
        )
    } else {
        None
    };
    // Only a deleting tombstone changes the rows it covers; a boundary one
    // ends an operand chain, which a merge operator's rows answer for.
    let deleting = match rt_plan {
        Some((tombstones, RangeTombstoneUse::Delete)) => tombstones.as_slice(),
        _ => &[],
    };
    super::work::estimate_merge(
        &super::work::MergePlan {
            inputs: tables,
            blob_files,
            gc_stats: version.gc_stats(),
            user_code: opts.config.compaction_filter_factory.is_some()
                || opts.config.merge_operator.is_some(),
            range_tombstones: deleting,
            comparator: opts.config.comparator.as_ref(),
        },
        |table| carry.as_ref().is_some_and(|carry| carry.takes(table)),
    )
}

fn create_compaction_stream<'a>(
    version: &Version,
    to_compact: &[TableId],
    gc_watermark: SeqNo,
    merge_operator: Option<Arc<dyn crate::merge_operator::MergeOperator>>,
    comparator: crate::comparator::SharedComparator,
    pace: Option<&crate::table::util::Pacer>,
    carry: Option<&crate::table::group_carry::CarryTarget>,
) -> crate::Result<Option<CompactionStream<'a, Merger<CompactionReader<'a>>>>> {
    let mut readers: Vec<CompactionReader<'_>> = vec![];
    let mut found = 0;

    for run in version.iter_levels().flat_map(|lvl| lvl.iter()) {
        if run.len() > 1 {
            let Some((lo, hi)) = pick_run_indexes(run, to_compact) else {
                continue;
            };

            let scanner = RunScanner::culled(run.clone(), (Some(lo), Some(hi)), pace.cloned())?;
            let scanner = match carry {
                Some(target) => scanner.with_carry(target.clone()),
                None => scanner,
            };
            readers.push(Box::new(scanner));

            found += hi - lo + 1;
        } else {
            for table in run.iter().filter(|x| to_compact.contains(&x.metadata.id)) {
                found += 1;
                let scanner = table.scan_paced(pace)?;
                let scanner = match carry {
                    Some(target) if target.takes(table) => {
                        scanner.with_carry(target.carry_for(table))
                    }
                    _ => scanner,
                };
                readers.push(Box::new(scanner));
            }
        }
    }

    Ok(if found == to_compact.len() {
        Some(
            CompactionStream::new(Merger::new(readers, comparator), gc_watermark)
                .with_merge_operator(merge_operator),
        )
    } else {
        None
    })
}

/// Like [`create_compaction_stream`] but restricts every input table to the key
/// range `bounds`, used by parallel sub-compactions where each thread owns a
/// disjoint slice of the key space. Each input table is read via
/// [`crate::Table::range`] (a raw, all-seqno, key-bounded scan that seeks within
/// the table), so a sub-compaction only touches the blocks overlapping its
/// slice. The bounds must partition the key space across sub-compactions
/// (`Excluded(lo)..=Included(hi)`) so every entry lands in exactly one.
#[cfg(feature = "std")]
#[expect(
    clippy::too_many_arguments,
    reason = "the serial stream's inputs plus the range and the carry plan"
)]
fn create_bounded_compaction_stream<'a>(
    version: &'a Version,
    to_compact: &HashSet<TableId>,
    bounds: (core::ops::Bound<UserKey>, core::ops::Bound<UserKey>),
    gc_watermark: SeqNo,
    merge_operator: Option<Arc<dyn crate::merge_operator::MergeOperator>>,
    comparator: crate::comparator::SharedComparator,
    pace: Option<&crate::table::util::Pacer>,
    carry: Option<&crate::table::group_carry::CarryTarget>,
) -> crate::Result<Option<CompactionStream<'a, Merger<CompactionReader<'a>>>>> {
    let mut readers: Vec<CompactionReader<'_>> = vec![];
    let mut found = 0;

    for run in version.iter_levels().flat_map(|lvl| lvl.iter()) {
        for table in run.iter().filter(|x| to_compact.contains(&x.metadata.id)) {
            found += 1;
            // A table whose groups or blocks can be carried is scanned one by
            // one, recording those wholly within this range.
            if let Some(target) = carry
                && target.takes(table)
            {
                let scanner = table.scan_carrying(
                    crate::table::group_carry::KeyBounds {
                        lo: bounds.0.clone(),
                        hi: bounds.1.clone(),
                        comparator: comparator.clone(),
                    },
                    Arc::clone(&target.queue),
                    pace,
                )?;
                match scanner {
                    Some(scanner) => readers.push(Box::new(scanner)),
                    None => readers.push(Box::new(core::iter::empty())),
                }
                continue;
            }
            // Compaction input is maintenance, not a read a caller made, so
            // it stays out of the read counters like the serial scanner's.
            let reader = table.range_iter(bounds.clone()).for_maintenance();
            let reader = match pace {
                Some(pace) => reader.with_pace(Arc::clone(pace)),
                None => reader,
            };
            readers.push(Box::new(reader));
        }
    }

    Ok((found == to_compact.len()).then(|| {
        CompactionStream::new(Merger::new(readers, comparator), gc_watermark)
            .with_merge_operator(merge_operator)
    }))
}

/// Sorts the destination-level max-keys, drops comparator-equal duplicates, and
/// removes the global maximum (splitting after it yields an empty trailing
/// range), returning the candidate interior boundary keys. Split out of
/// [`subcompaction_boundaries`] so the comparator-equality dedup is unit
/// testable without constructing a full [`Version`].
/// Gathers every range tombstone in the version (all levels), to gate
/// bottommost seqno-zeroing: a key covered by any tombstone keeps its real
/// seqno. Includes tombstones from tables NOT in the current compaction, so
/// "beyond output level" coverage is respected.
fn collect_version_tombstones(version: &Version) -> Vec<crate::range_tombstone::RangeTombstone> {
    version
        .iter_levels()
        .flat_map(|level| level.iter())
        .flat_map(|run| run.iter())
        .flat_map(|t| t.range_tombstones().iter().cloned())
        .collect()
}

/// Which range tombstones a compaction is given, and what it may do with them.
///
/// The bottommost one deletes what they cover. A composing merge operator needs
/// them anywhere, as chain boundaries only: a range tombstone between two
/// operands ends the chain, and a fold across it would produce one operand
/// newer than the still-propagating tombstone. Gathering them costs a pass over
/// the version, so nothing is gathered when neither applies.
fn range_tombstone_plan(
    version: &Version,
    is_last_level: bool,
    config: &crate::config::Config,
) -> Option<(
    Vec<crate::range_tombstone::RangeTombstone>,
    RangeTombstoneUse,
)> {
    let use_ = if is_last_level {
        RangeTombstoneUse::Delete
    } else if composes_operands(config) {
        RangeTombstoneUse::Boundary
    } else {
        return None;
    };

    Some((collect_version_tombstones(version), use_))
}

/// Whether the configured merge operator opts its operands into composition.
fn composes_operands(config: &crate::config::Config) -> bool {
    config
        .merge_operator
        .as_ref()
        .is_some_and(|op| op.composes_operands())
}

/// Key ranges of every table this compaction does NOT read.
///
/// A key inside none of them has all of its versions in the inputs, which is
/// what [`CompactionStream::with_input_completeness`] needs to let a chain be
/// composed. It is the same question [`holds_every_surviving_version`] asks,
/// asked per key instead of over the whole compacted range, and without the
/// bottommost requirement that one adds for proving a base absent.
fn outside_tables<'a>(version: &'a Version, table_ids: &HashSet<TableId>) -> Vec<&'a Table> {
    version
        .iter_tables()
        .filter(|table| !table_ids.contains(&table.id()))
        .collect()
}

/// Builds the per-key completeness predicate for
/// [`CompactionStream::with_input_completeness`] over the tables this
/// compaction did not read.
///
/// Three tests, cheapest first. The seqno range rules out the tables that
/// cannot hold a version BETWEEN the chain's ends: an older one is the base
/// the composed operand will meet later, and a newer one applies after it, so
/// neither breaks the chain. The key range then rules out tables that do not
/// span the key at all, and the table's filter answers what the key range
/// cannot: whether it holds THIS key rather than merely spanning it.
///
/// All three are conservative in the safe direction. A table whose seqno range
/// is unknown, a filter false positive, or an error reading one all read as
/// "may hold it", and the fold is declined.
/// The completeness predicate this compaction should declare, or `None` when
/// its operator does not compose and the question never arises.
fn input_completeness_for<'a>(
    version: &'a Version,
    table_ids: &HashSet<TableId>,
    config: &crate::config::Config,
) -> Option<impl Fn(&[u8], SeqNo, SeqNo) -> bool + 'a> {
    composes_operands(config).then(|| {
        input_completeness_predicate(
            outside_tables(version, table_ids),
            config.comparator.clone(),
        )
    })
}

fn input_completeness_predicate(
    outside: Vec<&Table>,
    comparator: crate::comparator::SharedComparator,
) -> impl Fn(&[u8], SeqNo, SeqNo) -> bool + '_ {
    use crate::version::run::Ranged as _;

    move |key: &[u8], oldest: SeqNo, newest: SeqNo| {
        let cmp = comparator.as_ref();
        let key_hash = crate::hash::hash64(key);

        !outside.iter().any(|table| {
            may_break_chain(
                table.seqno_range(),
                table.key_range(),
                cmp,
                key,
                (oldest, newest),
                || table.bloom_may_contain_key(key, key_hash).unwrap_or(true),
            )
        })
    }
}

/// Whether one table outside the inputs could hold a version of `key` inside
/// the `(oldest, newest)` sequence-number interval, and so break a chain being
/// composed across it. See [`input_completeness_predicate`].
///
/// Split out from the table walk so the three tests are exercisable without
/// building tables: `may_contain` is only consulted once the cheap ranges have
/// failed to rule the table out, since it can read a filter block.
fn may_break_chain(
    seqno_range: Option<(SeqNo, SeqNo)>,
    key_range: &crate::KeyRange,
    comparator: &dyn crate::comparator::UserComparator,
    key: &[u8],
    (oldest, newest): (SeqNo, SeqNo),
    may_contain: impl FnOnce() -> bool,
) -> bool {
    // An unknown seqno range proves nothing, so it has to be assumed to
    // overlap; the key range and the filter can still rule the table out.
    let within_seqnos = seqno_range.is_none_or(|(low, high)| high >= oldest && low <= newest);
    if !within_seqnos {
        return false;
    }

    let within_keys = comparator.compare(key_range.min(), key) != core::cmp::Ordering::Greater
        && comparator.compare(key_range.max(), key) != core::cmp::Ordering::Less;

    within_keys && may_contain()
}

/// The list the bottommost seqno-zeroer should see: it only ever acts at the
/// bottom, so boundary-only tombstones are none of its business.
fn zeroing_tombstones_from(
    plan: Option<&(
        Vec<crate::range_tombstone::RangeTombstone>,
        RangeTombstoneUse,
    )>,
) -> Vec<crate::range_tombstone::RangeTombstone> {
    match plan {
        Some((tombstones, RangeTombstoneUse::Delete)) => tombstones.clone(),
        _ => Vec::new(),
    }
}

/// Whether relocating `payload.table_ids` to `payload.dest_level` leaves the
/// point read reaching the same version of every key.
///
/// A move rewrites nothing: it just relabels which level the tables sit on. The
/// read path walks levels from the top and stops at the first one holding the
/// key, taking "higher" to mean "newer". A move that drops tables underneath an
/// overlapping table breaks that reading: the older copy above now answers the
/// read. It does not show up as corruption, since both tables are intact on
/// disk. Within the destination itself the moved runs go ahead of the ones
/// already there, which is correct for tables newer than them; an overlap there
/// is still refused, as the move cannot prove the moved tables are the newer.
///
/// This condition is per table rather than over the moved set as a whole, so
/// that a move spanning a gap is not refused for an overlap that falls inside
/// it. "Newly" is what matters: only the levels BETWEEN the source and the
/// destination change precedence, in whichever direction the move goes. Levels
/// outside that span keep winning exactly as before, so an overlap there is not
/// this move's problem and refusing over it would decline a safe move forever.
///
/// One overlap is survivable, and only one: a table leaving L0 may pass under a
/// table that stays there, provided everything in the mover is older than
/// everything in the table it passes. L0 is read by taking the best answer
/// across all of its runs and only then looking lower, so the newer peer wins
/// while it is visible and a snapshot that predates it falls through to the
/// moved table. No other level reads that way: from L1 down the lookup stops at
/// the first run whose range covers the key, hit or miss, so a table hidden
/// under an overlapping one is not reached at all, however old it is.
fn nothing_the_move_passes_under_overlaps(
    version: &Version,
    payload: &CompactionPayload,
    comparator: &dyn crate::comparator::UserComparator,
) -> bool {
    let dest = usize::from(payload.dest_level);

    // Grouped by the level they leave, because that is what decides which
    // levels each one crosses. There are at most as many groups as the tree has
    // levels, and a move almost always takes one or two.
    let mut by_source: Vec<(usize, Vec<&Table>)> = Vec::new();
    for (source, table) in moved_tables(version, payload) {
        match by_source.iter_mut().find(|(level, _)| *level == source) {
            Some((_, tables)) => tables.push(table),
            None => by_source.push((source, vec![table])),
        }
    }

    for (source, mut tables) in by_source {
        sort_by_min_key(&mut tables, comparator);
        let span = source.min(dest)..=source.max(dest);

        for (idx, level) in version.iter_levels().enumerate() {
            if !span.contains(&idx) {
                continue;
            }
            let mut others: Vec<&Table> = level
                .iter()
                .flat_map(|run| run.iter())
                .filter(|other| !payload.table_ids.contains(&other.id()))
                .collect();
            sort_by_min_key(&mut others, comparator);

            if idx == 0 {
                // L0 is the one level where an overlap can still be allowed,
                // and the test is per pair, so every pair has to be examined
                // rather than just the first one found. Pairwise is affordable
                // precisely here: L0 is kept small by the stall / stop
                // thresholds, which is not true of the levels below it.
                if tables.iter().any(|table| {
                    others.iter().any(|other| {
                        overlaps(table, other, comparator) && !proven_older(table, other)
                    })
                }) {
                    return false;
                }
            } else if any_overlap(&tables, &others, comparator) {
                // Below L0 no overlap is allowed, so the question collapses to
                // whether one exists, which two sorted walks answer in one
                // pass. The level can hold thousands of tables and this runs
                // under the version read lock, so rescanning it per moved
                // table is what this avoids.
                return false;
            }
        }
    }

    true
}

/// Sorts tables by the minimum key of their range, which is the order both
/// sides of [`any_overlap`] are required to be in.
fn sort_by_min_key(tables: &mut [&Table], comparator: &dyn crate::comparator::UserComparator) {
    use crate::version::run::Ranged as _;

    tables.sort_by(|a, b| comparator.compare(a.key_range().min(), b.key_range().min()));
}

/// Whether any table on one side overlaps any table on the other.
///
/// Linear in the two lists, where the obvious nested loop is the product of
/// their lengths. Both MUST be sorted by minimum key ([`sort_by_min_key`]).
///
/// A pointer advances only past a table that cannot overlap anything left on
/// the other side, since every table remaining there starts at or after the one
/// it was just compared against. So no overlap is stepped over.
///
/// It answers EXISTENCE only. Enumerating every overlapping pair needs more
/// than two pointers, because the tables within one list may overlap each other
/// (a level holds several runs, and only a run is internally disjoint).
fn any_overlap(
    left: &[&Table],
    right: &[&Table],
    comparator: &dyn crate::comparator::UserComparator,
) -> bool {
    use crate::version::run::Ranged as _;

    let mut l = 0;
    let mut r = 0;

    while let (Some(a), Some(b)) = (left.get(l), right.get(r)) {
        if comparator
            .compare(a.key_range().max(), b.key_range().min())
            .is_lt()
        {
            l += 1;
        } else if comparator
            .compare(b.key_range().max(), a.key_range().min())
            .is_lt()
        {
            r += 1;
        } else {
            return true;
        }
    }

    false
}

/// The tables this task moves, each with the level it currently sits on.
fn moved_tables<'a>(version: &'a Version, payload: &CompactionPayload) -> Vec<(usize, &'a Table)> {
    version
        .iter_levels()
        .enumerate()
        .flat_map(|(idx, level)| {
            level
                .iter()
                .flat_map(|run| run.iter())
                .map(move |table| (idx, table))
        })
        .filter(|(_, table)| payload.table_ids.contains(&table.id()))
        .collect()
}

/// Whether everything in `table` is older than everything in `other`, which is
/// what lets a table leave L0 under a peer that overlaps it.
///
/// A table whose bounds do not translate into the tree's seqno space proves
/// nothing, so it answers `false`: the exemption exists to let a PROVABLY safe
/// move through, and no proof means the move is refused like any other overlap.
fn proven_older(table: &Table, other: &Table) -> bool {
    match (table.seqno_range(), other.seqno_range()) {
        (Some((_, newest_moved)), Some((oldest_peer, _))) => newest_moved < oldest_peer,
        _ => false,
    }
}

fn overlaps(a: &Table, b: &Table, comparator: &dyn crate::comparator::UserComparator) -> bool {
    use crate::version::run::Ranged as _;

    a.key_range()
        .overlaps_with_key_range_cmp(b.key_range(), comparator)
}

/// Whether this compaction holds every surviving version of the keys it reads,
/// so that a key missing from its inputs is genuinely absent from the tree.
///
/// Neither the destination level nor the level layout proves that. A level can
/// hold several overlapping runs, so rewriting part of one leaves older
/// versions in a run that was never read; and a move can drop a NEWER entry
/// into an empty last level while an OLDER one stays on a level above, so
/// "nothing below me" is not "nothing older than me" either. What does prove it
/// is that no table outside the inputs can hold another version of these keys:
/// every table this compaction does not read is disjoint from the range it
/// covers. The three decisions that turn "not in my inputs" into "not in the
/// tree" hang off this: dropping a tombstone, zeroing a seqno, and folding
/// merge operands onto an absent base.
fn holds_every_surviving_version(
    version: &Version,
    table_ids: &HashSet<TableId>,
    is_last_level: bool,
    comparator: &dyn crate::comparator::UserComparator,
) -> bool {
    use crate::version::run::Ranged as _;

    if !is_last_level {
        return false;
    }

    let covered = crate::KeyRange::aggregate_cmp(
        version
            .iter_tables()
            .filter(|table| table_ids.contains(&table.id()))
            .map(Table::key_range),
        comparator,
    );

    version
        .iter_tables()
        .filter(|table| !table_ids.contains(&table.id()))
        .all(|table| {
            !table
                .key_range()
                .overlaps_with_key_range_cmp(&covered, comparator)
        })
}

/// Garbage-collects range tombstones for a bottommost compaction's output.
///
/// A tombstone at or below the watermark has been fully applied (every live
/// snapshot sees it, and this compaction physically dropped the keys it covers),
/// so it can be dropped — UNLESS a table outside this compaction overlaps its
/// range and might still hold a covered key, in which case dropping it would
/// resurrect that key. Tombstones above the watermark are always kept (read-time
/// application still needs them). Non-bottommost compactions keep everything.
#[cfg(feature = "std")]
fn range_tombstones_after_gc(
    input_rts: &[crate::range_tombstone::RangeTombstone],
    outside: &OutsideSpans,
    watermark: SeqNo,
    is_last_level: bool,
    comparator: &crate::comparator::SharedComparator,
) -> Vec<crate::range_tombstone::RangeTombstone> {
    if !is_last_level {
        return input_rts.to_vec();
    }
    let cmp = comparator.as_ref();
    input_rts
        .iter()
        .filter(|rt| {
            // Strict visibility: a tombstone at or above the watermark is still
            // needed by the oldest live snapshot (which reads at the watermark
            // and does not see `RT@watermark`), so keep it. Only strictly-below
            // tombstones are candidates for GC.
            !rt.visible_at(watermark) || outside.overlaps(rt, cmp)
        })
        .cloned()
        .collect()
}

/// The keys the tables outside a compaction hold, as disjoint inclusive spans
/// in key order: whether a table outside overlaps a range tombstone is then a
/// binary search, not a walk of the whole version per tombstone.
pub struct OutsideSpans(Vec<(UserKey, UserKey)>);

impl OutsideSpans {
    /// The spans of `version`'s tables other than `input_ids`.
    pub fn new(
        version: &Version,
        input_ids: &HashSet<TableId>,
        cmp: &dyn crate::comparator::UserComparator,
    ) -> Self {
        Self::from_ranges(
            version
                .iter_levels()
                .flat_map(|level| level.iter())
                .flat_map(|run| run.iter())
                .filter(|t| !input_ids.contains(&t.id()))
                .map(|t| {
                    let kr = &t.metadata.key_range;
                    (kr.min().clone(), kr.max().clone())
                })
                .collect(),
            cmp,
        )
    }

    /// Merges inclusive `[min, max]` ranges, in any order, into spans.
    fn from_ranges(
        mut ranges: Vec<(UserKey, UserKey)>,
        cmp: &dyn crate::comparator::UserComparator,
    ) -> Self {
        use core::cmp::Ordering::{Greater, Less};

        ranges.sort_by(|a, b| cmp.compare(&a.0, &b.0));
        let mut spans: Vec<(UserKey, UserKey)> = Vec::with_capacity(ranges.len());
        for (min, max) in ranges {
            match spans.last_mut() {
                Some((_, end)) if cmp.compare(&min, end) != Greater => {
                    if cmp.compare(end, &max) == Less {
                        *end = max;
                    }
                }
                _ => spans.push((min, max)),
            }
        }
        Self(spans)
    }

    /// Whether a table outside the compaction holds keys in `rt`'s span, which
    /// only `rt` keeps deleted once the compaction's inputs are gone.
    fn overlaps(
        &self,
        rt: &crate::range_tombstone::RangeTombstone,
        cmp: &dyn crate::comparator::UserComparator,
    ) -> bool {
        use core::cmp::Ordering::Less;

        // [rt.start, rt.end) overlaps [min, max]: the first span ending at or
        // past the start has the least min of those that can.
        let first = self
            .0
            .partition_point(|(_, max)| cmp.compare(max, &rt.start) == Less);
        self.0
            .get(first)
            .is_some_and(|(min, _)| cmp.compare(min, &rt.end) == Less)
    }
}

/// The keys a writer answering for `bounds` covers, as `[lower, upper)` with
/// `None` for an end of the key space.
#[cfg(feature = "std")]
fn answered_span(
    bounds: &(core::ops::Bound<UserKey>, core::ops::Bound<UserKey>),
) -> (Option<UserKey>, Option<UserKey>) {
    use core::ops::Bound;

    (
        // An excluded lower bound is the previous range's included upper one:
        // a piece starting there is written by both, which deletes no more.
        match &bounds.0 {
            Bound::Included(key) | Bound::Excluded(key) => Some(key.clone()),
            Bound::Unbounded => None,
        },
        match &bounds.1 {
            Bound::Excluded(key) => Some(key.clone()),
            Bound::Included(key) => crate::range_tombstone::upper_bound_exclusive(key),
            Bound::Unbounded => None,
        },
    )
}

/// The tombstones of `rts` a writer answering for `[lower, upper)` keeps
/// past its outputs' keys: those reaching into that span that a table outside
/// the compaction overlaps.
fn tombstones_outside(
    rts: &[crate::range_tombstone::RangeTombstone],
    outside: &OutsideSpans,
    (lower, upper): (Option<UserKey>, Option<UserKey>),
    comparator: &crate::comparator::SharedComparator,
) -> crate::table::multi_writer::TombstonesOutside {
    use core::cmp::Ordering::Less;

    let cmp = comparator.as_ref();
    crate::table::multi_writer::TombstonesOutside {
        tombstones: rts
            .iter()
            .filter(|rt| {
                lower
                    .as_ref()
                    .is_none_or(|lower| cmp.compare(lower, &rt.end) == Less)
                    && upper
                        .as_ref()
                        .is_none_or(|upper| cmp.compare(&rt.start, upper) == Less)
                    && outside.overlaps(rt, cmp)
            })
            .cloned()
            .collect(),
        lower,
        upper,
    }
}

#[cfg_attr(
    not(feature = "std"),
    allow(
        dead_code,
        reason = "parallel sub-compaction boundary picker; the threaded parallel path is std-gated, so unused under no_std"
    )
)]
fn boundary_candidates(
    mut keys: Vec<UserKey>,
    comparator: &crate::comparator::SharedComparator,
) -> Vec<UserKey> {
    if keys.len() < 2 {
        return Vec::new();
    }
    keys.sort_by(|a, b| comparator.compare(a, b));
    // Dedup under the configured comparator, not raw bytes: a custom comparator
    // can rank two byte-distinct keys as equal, and a leftover equal cut point
    // would make adjacent sub-compaction ranges overlap or gap.
    keys.dedup_by(|a, b| comparator.compare(a, b).is_eq());
    keys.pop();
    keys
}

/// Interior split-boundary keys for a parallel compaction, derived from the
/// destination level's existing table boundaries (`RocksDB`'s approach: aligning
/// sub-compaction cuts to target-level files keeps outputs structured). Returns
/// at most `max_ranges - 1` keys (evenly sampled), sorted by `comparator`.
/// Empty → the compaction stays single-threaded (no usable cut points).
#[cfg(feature = "std")]
fn subcompaction_boundaries(
    version: &Version,
    dest_level: usize,
    max_ranges: usize,
    comparator: &crate::comparator::SharedComparator,
) -> Vec<UserKey> {
    if max_ranges < 2 {
        return Vec::new();
    }
    let Some(level) = version.level(dest_level) else {
        return Vec::new();
    };
    let keys: Vec<UserKey> = level
        .iter()
        .flat_map(|run| run.iter())
        .map(|t| t.metadata.key_range.max().clone())
        .collect();
    let keys = boundary_candidates(keys, comparator);
    if keys.is_empty() {
        return Vec::new();
    }
    let want = (max_ranges - 1).min(keys.len());
    if want == keys.len() {
        return keys;
    }
    // Evenly sample `want` boundaries across the candidates.
    let mut out = Vec::with_capacity(want);
    for i in 1..=want {
        let idx = ((i * keys.len()) / (want + 1)).min(keys.len() - 1);
        if let Some(key) = keys.get(idx) {
            out.push(key.clone());
        }
    }
    out.dedup();
    out
}

/// Turns interior boundary keys into `boundaries.len() + 1` disjoint key ranges
/// that partition the whole key space: `(Unbounded, Included(b0)]`,
/// `(Excluded(b_i), Included(b_{i+1})]`, …, `(Excluded(b_last), Unbounded)`.
/// Every entry falls in exactly one range, so the sub-compaction outputs union
/// to the same set the serial compaction would produce.
///
/// A boundary is the last key of a table of the level written into, so it
/// closes its range: the table's rows, its last row group's included, lie in
/// one range, which can then copy that group whole.
#[cfg(feature = "std")]
fn ranges_from_boundaries(
    boundaries: &[UserKey],
) -> Vec<(core::ops::Bound<UserKey>, core::ops::Bound<UserKey>)> {
    use core::ops::Bound::{Excluded, Included, Unbounded};
    let mut ranges = Vec::with_capacity(boundaries.len() + 1);
    let mut lo = Unbounded;
    for b in boundaries {
        ranges.push((lo.clone(), Included(b.clone())));
        lo = Excluded(b.clone());
    }
    ranges.push((lo, Unbounded));
    ranges
}

/// Error returned when a merge is interrupted by the stop signal, so the caller
/// re-shows the inputs and skips the install instead of committing a truncated
/// output. The install swaps every input table for what was written, so a
/// truncated commit drops the unread tail out of the tree; both the serial and
/// the parallel path refuse for that reason.
pub(super) fn cancelled_compaction() -> crate::Error {
    crate::Error::from(crate::io::Error::new(
        crate::io::ErrorKind::Interrupted,
        "compaction cancelled by stop signal",
    ))
}

/// One slice of a tight-space pass: the key it ends before (`None` for the
/// last, which runs to the end of the inputs), and the bytes it writes to
/// the tables and to the blob files.
#[cfg(feature = "std")]
#[derive(Clone, Debug, PartialEq, Eq)]
struct TightSlice {
    end: Option<UserKey>,
    tables: u64,
    blobs: u64,
}

/// The most bytes a slice may write: to the tables, to the blob files, and
/// to both together (a volume they share, and the tree's quota).
#[cfg(feature = "std")]
#[derive(Clone, Copy, Debug)]
struct SliceCaps {
    tables: u64,
    blobs: u64,
    total: u64,
}

/// The tables a slice writes: one is opened for every `target_size` bytes,
/// and one more on every key of `cut_keys` (in key order) the slice passes.
#[cfg(feature = "std")]
struct SliceOutputs<'a> {
    target_size: u64,
    cut_keys: &'a [UserKey],
}

/// What a relocating pass moves: the frames its inputs point at in the
/// `stale` blob files, written into files of `target_size` bytes, a file for
/// each of `file_groups` codecs and lifetime classes at once.
#[cfg(feature = "std")]
struct Relocated<'a> {
    stale: &'a HashSet<BlobFileId>,
    target_size: u64,
    file_groups: u64,
}

/// A blob file's metadata past its frames, short of its two keys: the block
/// that holds its counts, codec, version and the like.
#[cfg(feature = "std")]
const BLOB_META_BYTES: u64 = 1_024;

/// The bytes a relocated frame of `value` under `key` takes in a blob file.
#[cfg(feature = "std")]
fn frame_bytes(key: &[u8], on_disk_size: u32) -> u64 {
    crate::vlog::blob_file::writer::BLOB_HEADER_LEN as u64
        + key.len() as u64
        + u64::from(on_disk_size)
}

/// The bytes of the blob frames `item` points at in the `stale` files, which
/// a relocation copies: its own value, or the references among its cells.
#[cfg(feature = "std")]
fn stale_frame_bytes(item: &InternalValue, stale: &HashSet<BlobFileId>) -> crate::Result<u64> {
    use crate::blob_tree::field_row::{RowCell, decode_row};
    use crate::blob_tree::handle::BlobIndirection;
    use crate::coding::Decode;

    let key = &item.key.user_key;
    if item.key.value_type.is_indirection() {
        let indirection = BlobIndirection::decode_from(&mut &item.value[..])?;
        return Ok(if stale.contains(&indirection.vhandle.blob_file_id) {
            frame_bytes(key, indirection.vhandle.on_disk_size)
        } else {
            0
        });
    }
    if item.key.value_type.is_cell_row() {
        let mut bytes = 0;
        for field in decode_row(&item.value)? {
            if let RowCell::Ref { indirection, .. } = field.cell
                && stale.contains(&indirection.vhandle.blob_file_id)
            {
                bytes += frame_bytes(key, indirection.vhandle.on_disk_size);
            }
        }
        return Ok(bytes);
    }
    Ok(0)
}

/// The slices a tight-space pass of `inputs` runs in, each within `caps`
/// as far as its blocks allow, and what each writes.
///
/// A block weighs its share of its table's file, index and filter included,
/// and, in a relocating pass, the frames its entries point at in the stale
/// blob files, which the slice relocates: values are not spread evenly over
/// the keys, so a share of the table's bytes says nothing of a share of
/// theirs. A slice's tables weigh their blocks and the bytes each table it
/// opens holds past them, as `outputs` counts them. A slice ends before the
/// block that would take it past a cap, on the last key of the block
/// before. The key it ends on goes to the next slice, so that one counts the
/// entries of that key again, read from the blocks ending on it. A block of
/// a restricted input below its restriction is punched and writes nothing.
///
/// The index and block reads go through `pace`, the run's input pacer: they
/// are the run's reads as much as the merge's are.
#[cfg(feature = "std")]
fn tight_slices(
    inputs: &[Table],
    caps: SliceCaps,
    outputs: &SliceOutputs<'_>,
    relocated: Option<&Relocated<'_>>,
    comparator: &crate::comparator::SharedComparator,
    pace: &crate::table::util::Pacer,
) -> crate::Result<Vec<TightSlice>> {
    /// The blocks ending on one key: what they write, and where they lie as
    /// the input, the block and the input's bytes per data byte.
    struct Block<'a> {
        end_key: UserKey,
        tables: u64,
        blobs: u64,
        sources: Vec<(&'a Table, crate::table::BlockHandle, (u64, u64))>,
    }

    /// The bytes a table writes for an entry, at the most: its key and value
    /// and what frames them in a block.
    const ENTRY_FRAMING: u64 = 32;

    let cmp = comparator.as_ref();
    let mut blocks: Vec<Block> = Vec::new();
    let mut longest_key = 0u64;
    // What a table holds past its blocks and the sections that grow with
    // them, at the least the inputs show: the bytes every table takes
    // however few blocks it holds. The rest of a table's bytes go with its
    // blocks, in each block's share of the file.
    let mut fixed: Option<u64> = None;
    for input in inputs {
        let handles = input
            .maintenance_index_walk()
            .with_pace(Arc::clone(pace))
            .collect::<crate::Result<Vec<_>>>()?;
        // Bounded by the file's length.
        let data_bytes: u64 = handles.iter().map(|h| u64::from(h.size())).sum();
        // A file shorter than its sections says nothing of the rest.
        if let Some(past_blocks) = input
            .file_len()
            .checked_sub(data_bytes)
            .and_then(|past| past.checked_sub(input.data_scaled_section_bytes()))
        {
            fixed = Some(fixed.map_or(past_blocks, |least| least.min(past_blocks)));
        }
        for handle in handles {
            let end_key = handle.end_key().clone();
            if input
                .restrict_lower_bound()
                .is_some_and(|bound| cmp.compare(&end_key, bound) == core::cmp::Ordering::Less)
            {
                continue;
            }
            // The block's share of the file: at most the file's length.
            let tables = u64::try_from(
                u128::from(handle.size()) * u128::from(input.file_len())
                    / u128::from(data_bytes.max(1)),
            )
            .unwrap_or(u64::MAX);
            let mut blobs = 0u64;
            if let Some(relocated) = relocated
                && let Some(block) =
                    input.load_data_block_for_maintenance(handle.as_ref(), &**pace)?
            {
                block.try_for_each_entry(comparator.clone(), |item| {
                    longest_key = longest_key.max(item.key.user_key.len() as u64);
                    // Bounded by the stale files' bytes.
                    blobs += stale_frame_bytes(&item, relocated.stale)?;
                    Ok(())
                })?;
            }
            blocks.push(Block {
                end_key,
                tables,
                blobs,
                sources: alloc::vec![(
                    input,
                    *handle.as_ref(),
                    (input.file_len(), data_bytes.max(1)),
                )],
            });
        }
    }
    blocks.sort_by(|a, b| cmp.compare(&a.end_key, &b.end_key));
    // Blocks ending on one key go into one slice together: a slice ends on a
    // key, not between the inputs that hold it.
    let mut grouped: Vec<Block> = Vec::with_capacity(blocks.len());
    for block in blocks {
        match grouped.last_mut() {
            Some(group)
                if cmp.compare(&group.end_key, &block.end_key) == core::cmp::Ordering::Equal =>
            {
                group.tables += block.tables;
                group.blobs += block.blobs;
                group.sources.extend(block.sources);
            }
            _ => grouped.push(block),
        }
    }
    let mut blocks = grouped;
    let fixed = fixed.unwrap_or(0);
    // A table opened on a cut key weighs on the block that passes it.
    let mut cut_keys = outputs.cut_keys.iter().peekable();
    for block in &mut blocks {
        while cut_keys
            .next_if(|key| cmp.compare(key, &block.end_key) != core::cmp::Ordering::Greater)
            .is_some()
        {
            block.tables += fixed;
        }
    }

    // Every sum is bounded by the inputs' and the stale files' bytes. The
    // first table of a slice and every one past a target's worth of blocks
    // weigh their fixed bytes too. The destination level encodes the blocks
    // its own way (codec, filter, where blocks end), which an input's share
    // of its file does not bound: a sixteenth more is kept for it. The blob
    // frames go across as they lie and need none.
    let with_tables = |tables: u64| {
        let tables = tables + tables / 16;
        tables + fixed * (1 + tables / outputs.target_size.max(1))
    };
    // Each output blob file closes with its metadata, its first and last keys
    // among it: a file per target's worth of frames, and one more for each
    // codec and lifetime class the slice's frames may go to.
    let with_blobs = |blobs: u64| match relocated {
        Some(relocated) if blobs > 0 => {
            let files = blobs.div_ceil(relocated.target_size.max(1)) + relocated.file_groups;
            blobs + files * (BLOB_META_BYTES + 2 * longest_key)
        }
        _ => blobs,
    };
    let past = |acc: (u64, u64), block: &Block| {
        let tables = with_tables(acc.0 + block.tables);
        let blobs = with_blobs(acc.1 + block.blobs);
        tables > caps.tables || blobs > caps.blobs || tables + blobs > caps.total
    };
    // What the entries of the key the blocks of `group` end on write: the
    // next slice's, once a slice ends on that key.
    let key_weight = |group: &Block| -> crate::Result<(u64, u64)> {
        let mut weight = (0u64, 0u64);
        for &(input, handle, (file_len, data_bytes)) in &group.sources {
            let Some(block) = input.load_data_block_for_maintenance(&handle, &**pace)? else {
                continue;
            };
            block.try_for_each_entry(comparator.clone(), |item| {
                if cmp.compare(&item.key.user_key, &group.end_key) != core::cmp::Ordering::Equal {
                    return Ok(());
                }
                let raw = item.key.user_key.len() as u64 + item.value.len() as u64 + ENTRY_FRAMING;
                // The entry's share of the file, as a block's is: bounded by
                // the file's length.
                weight.0 +=
                    u64::try_from(u128::from(raw) * u128::from(file_len) / u128::from(data_bytes))
                        .unwrap_or(u64::MAX);
                if let Some(relocated) = relocated {
                    weight.1 += stale_frame_bytes(&item, relocated.stale)?;
                }
                Ok(())
            })?;
        }
        Ok(weight)
    };
    let mut slices = Vec::new();
    let mut acc = (0u64, 0u64);
    // The blocks ending on the key the slice so far ends on.
    let mut last: Option<&Block> = None;
    for block in &blocks {
        if let Some(previous) = last
            && past(acc, block)
        {
            slices.push(TightSlice {
                end: Some(previous.end_key.clone()),
                tables: with_tables(acc.0),
                blobs: with_blobs(acc.1),
            });
            acc = key_weight(previous)?;
        }
        acc = (acc.0 + block.tables, acc.1 + block.blobs);
        last = Some(block);
    }
    slices.push(TightSlice {
        end: None,
        tables: with_tables(acc.0),
        blobs: with_blobs(acc.1),
    });
    Ok(slices)
}

/// Builds a per-slice compaction payload over the still-surviving input ids,
/// inheriting the original payload's level / target fields.
#[cfg(feature = "std")]
fn slice_payload_for(views: &[Table], payload: &CompactionPayload) -> CompactionPayload {
    CompactionPayload {
        table_ids: views.iter().map(Table::id).collect(),
        dest_level: payload.dest_level,
        canonical_level: payload.canonical_level,
        target_size: payload.target_size,
    }
}

/// Opt-in tight-space compaction: rewrites the merge's inputs in key-range
/// slices, installing each slice as one durable version edit and reclaiming each
/// consumed input prefix via hole punching once the prior view drains, so the
/// peak transient footprint is one slice rather than the whole rewrite. Engaged
/// from [`do_compaction`] when the space gate finds no fitting merge and the tree
/// opts in.
///
/// Handles single- and multi-input merges, including KV-separated trees. Each
/// slice merges the surviving inputs over `[lower, boundary)`; an input whose
/// data extends past the boundary is re-opened as a restricted view (clamped +
/// prefix punched), while one fully consumed by the slice is dropped outright.
/// Blob fragmentation each slice produces is folded into the running GC stats so
/// dead blob files are dropped at the final removal (a blob may still be
/// referenced by an unprocessed slice, so per-slice dropping would be unsafe).
#[cfg(feature = "std")]
fn run_tight_space_compaction(
    mut compaction_state: CompactionGuard<'_>,
    version_history_lock: VersionsReadGuard<'_>,
    opts: &Options,
    payload: &CompactionPayload,
) -> crate::Result<CompactionResult> {
    use core::ops::Bound;

    // Capability gate: the destination volume must support hole punching.
    let (dest_path, dest_fs) = opts.config.tables_folder_for_level(payload.dest_level);
    if !dest_fs.capabilities(&dest_path).punch_hole {
        log::info!("Tight-space compaction unavailable: backend lacks punch_hole");
        return Ok(CompactionResult::nothing());
    }

    // A reclaim a checkpoint's window forced the deletion pause to retain is
    // space this very compaction is short of, so collect it before deciding
    // anything: the checkpoint that blocked it has usually released by now.
    opts.deletion_pause.retry_pending_reclaims();

    let latest = version_history_lock.latest_version();
    let Some(inputs) = payload
        .table_ids
        .iter()
        .map(|&id| latest.version.get_table(id).cloned())
        .collect::<Option<Vec<_>>>()
    else {
        return Ok(CompactionResult::nothing());
    };

    // KV-separated trees: identify the stale (fragmented) blob files this merge
    // would relocate. When non-empty the slice loop runs a RELOCATING tight
    // compaction (blob defrag) — moving live entries into a fresh compact file
    // and punching each stale file's consumed prefix per slice. We keep only the
    // IDs, not the `BlobFile` handles: retaining an Arc would pin the stale
    // `Inner` and block its punch-on-drop. Each slice re-fetches the current
    // handle from the version.
    let stale_files = match &opts.config.kv_separation_opts {
        Some(blob_opts) => {
            let picked: HashSet<TableId> = payload.table_ids.iter().copied().collect();
            pick_blob_files_to_rewrite(&picked, &latest.version, blob_opts)?
        }
        None => Vec::new(),
    };
    // Each stale file is punched through its OWN backend, which a separately
    // backed `blobs/` can refuse.
    if let Some(bf) = stale_files
        .iter()
        .find(|bf| !bf.0.fs.capabilities(&bf.0.path).punch_hole)
    {
        log::info!(
            "Tight-space compaction unavailable: the backend of blob file {} \
             lacks punch_hole, so its relocated prefix would stay allocated",
            bf.0.path.display(),
        );
        return Ok(CompactionResult::nothing());
    }
    // Sum of on-disk blob-file sizes; bounded by filesystem capacity, so it
    // cannot overflow u64 (matches the gate's `acc + size`).
    let mut stale_total_bytes = 0u64;
    for bf in &stale_files {
        stale_total_bytes += bf.physical_size()?;
    }
    // Slices apply no range tombstone, so none is planned for. Recorded once
    // the pass is sure to run, below.
    let estimate = merge_estimate(opts, &latest.version, payload, &inputs, &stale_files, None);
    let stale_ids: Vec<BlobFileId> = stale_files.iter().map(BlobFile::id).collect();
    // A relocation writes a file for each codec of its sources.
    let mut codecs: Vec<crate::CompressionType> = Vec::new();
    for bf in &stale_files {
        if !codecs.contains(&bf.compression()) {
            codecs.push(bf.compression());
        }
    }
    let stale_codecs = codecs.len() as u64;
    drop(stale_files);
    let relocating = !stale_ids.is_empty();

    // The reclaim runs through each INPUT's own backend, not the destination's:
    // under level routing an input can sit on a volume that refuses the punch.
    // Admitting the slice anyway commits its restriction and writes its output
    // while the consumed prefix stays allocated, so the pass ends using MORE
    // space on the volume it was meant to relieve.
    if let Some(t) = inputs
        .iter()
        .find(|t| !t.fs.capabilities(&t.path).punch_hole)
    {
        log::info!(
            "Tight-space compaction unavailable: the backend of input table {} lacks \
             punch_hole, so its consumed prefix would stay allocated",
            t.path.display(),
        );
        return Ok(CompactionResult::nothing());
    }

    // Being able to punch is not the same as punching where it helps. Each
    // slice adds its WHOLE output to the destination volume but frees only the
    // share of its inputs that lives THERE, so a destination under pressure
    // with inputs on another volume grows monotonically across slices: the pass
    // commits several, hits ENOSPC, and never relieves the pool that engaged
    // it. When the destination can hold the whole merge it is not the
    // constrained pool (the gate skipped for the blob volume instead) and the
    // reclaim landing elsewhere is fine.
    let inputs_total: u64 = inputs.iter().map(Table::file_len).sum();
    // What the compactions sharing the pool have still to write is not free.
    let dest_free = opts.capacity.available(&*dest_fs, &dest_path);
    // What they held then: the slices are sized to the room left past it.
    let dest_held = opts.capacity.held_on(&*dest_fs, &dest_path);
    if dest_free < inputs_total {
        // Only a PROVEN different volume forfeits the relief; an unproven pair
        // counts as shared, leaving the single-volume default untouched.
        //
        // Some inputs elsewhere is not the same as no relief. The output is
        // bounded by the inputs, and each slice frees the DESTINATION-local
        // share as it goes, so the volume only has to absorb the remote share:
        // 90 MB local plus 10 MB remote completes with 20 MB free. Requiring
        // every input to be colocated would leave a constrained tree read-only
        // with the reclaimable bytes sitting on the very volume that is full.
        let remote_total: u64 = inputs
            .iter()
            .filter(|t| separate_volumes(&*t.fs, &t.path, &*dest_fs, &dest_path))
            .map(Table::file_len)
            .sum();
        if dest_free < remote_total {
            log::info!(
                "Tight-space compaction unavailable: {remote_total} bytes of input live on a \
                 different volume than the constrained destination and only {dest_free} bytes \
                 are free there, so the slices would grow it without relief",
            );
            return Ok(CompactionResult::nothing());
        }
    }

    // The tree-wide logical quota is the OTHER constraint the space gate
    // enforces, and it is the one that engaged this pass whenever the backend
    // cannot report free space: an unanswerable `available_space` reads as
    // physically unbounded, so nothing else is left to reject a merge. The
    // clamp-to-zero matches the gate's min-semantics (a quota set below the
    // live footprint leaves no headroom).
    let quota_limit = opts.runtime_config.load().storage_limit_bytes;
    let used_at_start = crate::storage_stats::compute_used_bytes(&latest.version)?;
    let quota_headroom = match quota_limit {
        Some(limit) => limit
            .saturating_sub(used_at_start)
            .saturating_sub(opts.capacity.quota_held(opts.tree_id)),
        None => u64::MAX,
    };
    // What the pass reads and writes, and the footprint of all else when it
    // starts: a slice is checked against the quota less what all else has
    // grown by since. The pass's own growth is what its slices were cut for.
    let mut own_tables: HashSet<TableId> = payload.table_ids.iter().copied().collect();
    let mut own_blobs: HashSet<BlobFileId> = stale_ids.iter().copied().collect();
    let others_at_start =
        crate::storage_stats::compute_used_bytes_but(&latest.version, &own_tables, &own_blobs)?;
    let cut_keys = super::flavour::output_cut_keys(&latest.version, opts, payload);

    drop(latest);
    drop(version_history_lock);
    if inputs.is_empty() {
        return Ok(CompactionResult::nothing());
    }
    let tables_in = inputs.len();

    let comparator = opts.config.comparator.clone();
    // A relocating merge writes the RELOCATED blob payload as well as the
    // (handle-only, small) tables, and that payload lands on the BLOB volume,
    // which level routing can separate from the SST destination: each is
    // capped by the room on its own volume, or both together by the room on
    // one they share. Whatever the volumes allow, a slice has to fit the
    // quota too: a tree-wide logical limit, not a property of one volume.
    // Without it a quota-only deployment slices nothing (an unbounded cap
    // cuts no slice) and never leaves read-only, which is the state that
    // engaged the pass.
    let blob_dir = opts.config.path.join(BLOBS_FOLDER);
    let relocating_blobs = relocating && stale_total_bytes > 0;
    let blob_free = if relocating_blobs {
        opts.capacity.available(&*opts.config.fs, &blob_dir)
    } else {
        0
    };
    let blob_held = opts.capacity.held_on(&*opts.config.fs, &blob_dir);
    // The tables and the relocated blob files take their room from one volume.
    let shared_volume =
        relocating_blobs && !separate_volumes(&*dest_fs, &dest_path, &*opts.config.fs, &blob_dir);
    let caps = if !relocating_blobs {
        SliceCaps {
            tables: dest_free.max(1),
            blobs: u64::MAX,
            total: quota_headroom.max(1),
        }
    } else if !shared_volume {
        SliceCaps {
            tables: dest_free.max(1),
            blobs: blob_free.max(1),
            total: quota_headroom.max(1),
        }
    } else {
        SliceCaps {
            tables: u64::MAX,
            blobs: u64::MAX,
            total: dest_free.min(blob_free).min(quota_headroom).max(1),
        }
    };
    let stale_set: HashSet<BlobFileId> = stale_ids.iter().copied().collect();
    let relocated = relocating_blobs.then(|| Relocated {
        stale: &stale_set,
        target_size: opts
            .config
            .kv_separation_opts
            .as_ref()
            .map_or(u64::MAX, |blob_opts| blob_opts.file_target_size),
        file_groups: stale_codecs
            * opts
                .config
                .kv_separation_opts
                .as_ref()
                .map_or(1, |blob_opts| u64::from(blob_opts.lifetime_groups.get())),
    });
    let outputs = SliceOutputs {
        target_size: payload.target_size,
        cut_keys: &cut_keys,
    };
    let slices = tight_slices(
        &inputs,
        caps,
        &outputs,
        relocated.as_ref(),
        &comparator,
        &input_pacer(opts),
    )?;
    let Some((tail, slices)) = slices.split_last() else {
        return Ok(CompactionResult::nothing());
    };
    // A slice past the caps (a block, or the blocks of one key, alone past
    // them) cannot be cut any finer: it would write past the room the pass
    // found. The pass runs the slices before it, which free room as they go,
    // and stops there; one that comes first leaves the pass nothing to run.
    // A single slice within them is the whole merge, in one go.
    let past_caps = |slice: &TightSlice| {
        slice.tables > caps.tables
            || slice.blobs > caps.blobs
            || slice.tables + slice.blobs > caps.total
    };
    if past_caps(slices.first().unwrap_or(tail)) {
        return Ok(CompactionResult::nothing());
    }

    opts.work.plan(estimate);

    // Each slice holds the room it takes, so the compactions sharing the pool
    // count it: the tables its blocks rewrite and the stale frames they
    // relocate, as the slices were cut.
    //
    // Checked as each slice starts, in one step with taking it: the caps were
    // what was free past what the other compactions sharing the pool held
    // when the pass measured it, and another may have taken that room since,
    // or between two slices. Only their holds are checked: the pass's own
    // slices write ahead of the punches that free their inputs, so its room
    // between slices is not what it measured and is not the question. A slice
    // the others left no room for ends the pass there, its installed slices
    // kept; the next compaction picks up the rest. What a slice writes past
    // its room it takes on as it lands, under the quota as it stands then and
    // short of the flush floor, or the pass stops. The quota a slice is
    // checked against is the one the pass was cut for, less what the files
    // it neither reads nor writes (`own_tables`, `own_blobs`) have grown by
    // since: a flush, another compaction.
    let reserve_slice = |slice: &TightSlice,
                         own_tables: &HashSet<TableId>,
                         own_blobs: &HashSet<BlobFileId>|
     -> crate::Result<bool> {
        if past_caps(slice) {
            return Ok(false);
        }
        let quota_headroom = match quota_limit {
            Some(limit) => {
                let others = crate::storage_stats::compute_used_bytes_but(
                    &opts.version_history.read().latest_version().version,
                    own_tables,
                    own_blobs,
                )?;
                // What all else grew by: none when it shrank.
                limit
                    .saturating_sub(used_at_start)
                    .saturating_sub(others.saturating_sub(others_at_start))
            }
            None => u64::MAX,
        };
        let demands = [
            super::capacity::Demand {
                fs: &*dest_fs,
                path: &dest_path,
                bytes: slice.tables,
            },
            super::capacity::Demand {
                fs: &*opts.config.fs,
                path: &blob_dir,
                bytes: slice.blobs,
            },
        ];
        #[cfg(all(test, feature = "std"))]
        {
            let squeeze = opts
                .config
                .tight_slice_squeeze
                .swap(0, core::sync::atomic::Ordering::SeqCst);
            if squeeze > 0 {
                // Another tree's hold, kept for the life of the pool.
                core::mem::forget(opts.capacity.hold(
                    TreeId::MAX,
                    &[super::capacity::Demand {
                        fs: &*dest_fs,
                        path: &dest_path,
                        bytes: squeeze,
                    }],
                ));
            }
        }
        // What the others may hold on each volume: what they held when the
        // pass measured it, and what room a slice left past it then, both of
        // its demands taken from a volume they share. The slack is never
        // below none.
        let (tables_room, blobs_room) = if shared_volume {
            let both = slice.tables + slice.blobs;
            (both, both)
        } else {
            (slice.tables, slice.blobs)
        };
        let [tables, blobs] = demands;
        let within = [
            (tables, dest_held + dest_free.saturating_sub(tables_room)),
            (blobs, blob_held + blob_free.saturating_sub(blobs_room)),
        ];
        // The slice before is installed: what it held is not the others'.
        opts.outputs.release_room();
        let check = super::capacity::Check {
            quota_headroom,
            floor: super::capacity::Floor::MayConsume,
        };
        Ok(
            match opts.capacity.reserve_within(opts.tree_id, &within, check) {
                Some(room) => {
                    opts.outputs.reserve(room);
                    true
                }
                None => false,
            },
        )
    };
    // Set when a slice no longer fits: the pass ends there.
    let mut stopped = false;
    // Set when a slice's consumed inputs are still held once it is installed:
    // the room the next slice was cut to is not back yet.
    let mut reclaim_deferred = false;

    // Hide the inputs so the strategy cannot re-select them mid-loop.
    compaction_state
        .hidden_set_mut()
        .hide(payload.table_ids.iter().copied());

    let dst_lvl: usize = payload.canonical_level.into();
    // Tight-space slice merges apply NO removal semantics: no bottommost GC
    // (tombstone drop, seqno zeroing, range-tombstone application) even when the
    // destination IS the last level, and no user compaction filter (passed as
    // `false` to each slice's sub-compaction below). A slice restricts a surviving
    // input and hides its consumed prefix; if a fully-consumed sibling carried the
    // tombstone for a key that also lives in that prefix, last-level GC would drop
    // the tombstone from the output — and a filter verdict would drop a record
    // that survives nowhere else. A crash window that leaves the survivor
    // unrestricted (sidecar not yet written, prefix not yet punched) would then
    // let manifest repair re-expose the removed rows. Retaining every record makes
    // the slice output a SUPERSET that shadows any surviving prefix in every crash
    // window; a later normal last-level compaction reclaims
    // the deferred GC. Space is still reclaimed here via the hole punch, which is
    // what tight-space is for.
    let bottommost_gc = false;
    let blobs_folder = opts.config.path.join(BLOBS_FOLDER);
    // Canonicalize range tombstones across all inputs (a boundary-spanning RT is
    // otherwise copied into every slice output).
    let mut rts: Vec<crate::range_tombstone::RangeTombstone> = inputs
        .iter()
        .flat_map(|t| t.range_tombstones().iter().cloned())
        .collect();
    rts.sort();
    rts.dedup();

    let mut current_views = inputs;
    let mut lower: Bound<UserKey> = Bound::Unbounded;
    // Cumulative per-stale-file consumed frontier (`blob_file_id -> frame_end`).
    // Doubles as the next slice's scan-resume map and the punch offset for each
    // stale file's prior view. Empty (and unused) on the non-relocating path.
    let mut resume_offsets: crate::HashMap<BlobFileId, u64> = crate::HashMap::default();
    // One runtime snapshot for every slice: the slices are one compaction.
    let rc = opts.runtime_config.load_full();

    let result = (|| -> crate::Result<usize> {
        let mut tables_out = 0usize;

        // Every slice but the tail ends before a key.
        for (slice, boundary) in slices
            .iter()
            .filter_map(|slice| slice.end.as_ref().map(|end| (slice, end)))
        {
            if reclaim_deferred || !reserve_slice(slice, &own_tables, &own_blobs)? {
                stopped = true;
                return Ok(tables_out);
            }
            // Merge [lower, boundary) over the surviving (restricted) input views.
            let slice_payload = slice_payload_for(&current_views, payload);
            let version = opts.version_history.read().latest_version();

            // Re-fetch the stale blob files' CURRENT handles (slice 0: originals;
            // later: the prior slice's re-opened views). Never retained across the
            // iteration — a lingering Arc would block the prior view's punch.
            let current_stale: Vec<BlobFile> = if relocating {
                stale_ids
                    .iter()
                    .filter_map(|id| version.version.blob_files.get(*id).cloned())
                    .collect()
            } else {
                Vec::new()
            };
            let reloc = if relocating {
                Some(RelocationSetup {
                    stale_files: current_stale.clone(),
                    resume_offsets: resume_offsets.clone(),
                })
            } else {
                None
            };

            let span = (lower.clone(), Bound::Excluded(boundary.clone()));
            // Planned per slice over its current views: a plan holds the views
            // it inherits from, and a view held past its slice would block the
            // punch of its consumed prefix. Held until the slice's install, so
            // the filter budget keeps the room its outputs take until then.
            let filter_sizing = super::flavour::plan_filters(
                &version.version,
                opts,
                &slice_payload,
                &[],
                Some(span.clone()),
                &rc,
            );
            let produced = run_subcompaction(
                opts,
                &slice_payload,
                &version.version,
                Vec::new(),
                &rts,
                None,
                span,
                dst_lvl,
                bottommost_gc,
                // Slices apply no removal semantics; see the parameter docs.
                false,
                &blobs_folder,
                reloc,
                &rc,
                filter_sizing.clone(),
            )?;
            drop(version);

            let outputs: Vec<Table> = produced.created_tables().to_vec();
            own_tables.extend(outputs.iter().map(Table::id));
            own_blobs.extend(produced.created_blob_files().iter().map(BlobFile::id));
            // Install the tree-wide sinks on each slice output before the
            // version edit makes it visible (mirrors `install_merge`): the
            // deletion pause, so a later in-place heal cannot race a
            // checkpoint hard-link, and the heal-hint sink, so a
            // confirmed-persistent ECC correction on a read can queue this
            // SST for a healing rewrite instead of leaving the bitrot on
            // disk.
            for table in &outputs {
                table.bind_to_tree(&crate::table::TableSinks {
                    deletion_pause: &opts.deletion_pause,
                    heal_hints: &opts.heal_hints,
                    read_budget: opts.config.columnar_read_budget,
                    #[cfg(feature = "std")]
                    background_deleter: &opts.background_deleter,
                    track_filter_probes: opts.config.filter_advisor.is_some(),
                });
            }
            // KV-separation: blob files this slice relocated live entries into,
            // plus the GC diff of entries it dropped.
            // The reopened stale views pushed onto it below share the path of a
            // blob file the current version names; no writer created them, so the
            // run never counts them among the files it may remove.
            let mut new_blobs: Vec<BlobFile> = produced.created_blob_files().to_vec();
            let frag = produced.blob_frag_map().clone();
            let gc_diff = if frag.is_empty() { None } else { Some(frag) };
            let released_objects = produced.released_objects().to_vec();

            // Advance the cumulative frontier from this slice's relocation, then
            // release `produced` (and its clones of the stale Inners) so the prior
            // views can punch once drained.
            if relocating {
                for (id, fe) in produced.consumed_through() {
                    let slot = resume_offsets.entry(*id).or_insert(0);
                    *slot = (*slot).max(*fe);
                }
            }
            // Charged once this slice is installed, below.
            let relocated_bytes = produced.relocated_bytes();
            let carried = produced.carried();
            let installed_work = produced.installed_work()?;
            let mut installed_files = crate::tree::InstalledCharge::default();
            installed_files.add(
                &outputs,
                produced.created_blob_files(),
                crate::fs::WriteOrigin::CompactionTable,
                crate::fs::WriteOrigin::CompactionBlob,
            )?;
            drop(produced);

            // Serialize each surviving input's suffix-digest capture (inside
            // `reopen_restricted` below) and this slice's manifest install
            // against a concurrent in-place heal on the same table identity. A
            // patrol that heals a suffix block BETWEEN the digest read and the
            // version edit would bind the restricted manifest to a pre-heal
            // suffix digest; the patrol then sees the restriction change and
            // skips its whole-file refresh, and its attestation binds whole-file
            // (not suffix) digests, so once the prefix is punched the mismatch is
            // permanently unattributable. Hold each restricted input's heal lock
            // across the reopen AND the `upgrade_version` below (released when
            // these guards drop at the end of this slice iteration). Only inputs
            // that extend past the boundary (`max >= boundary`) are re-opened, so
            // only those are locked. The patrol holds at most one table's heal
            // lock at a time, so acquiring several here cannot deadlock it.
            //
            // Lock-order note: this path takes `compaction_state` (held from
            // `do_compaction`) BEFORE these heal locks, while the patrol reconcile
            // takes a heal lock and then `compaction_state` inside
            // `refresh_table_checksum`. To keep that inversion from deadlocking,
            // the patrol side `try_lock`s `compaction_state` and skips its refresh
            // on contention rather than blocking — so it never waits on
            // `compaction_state` while holding a heal lock this loop wants.
            #[cfg(feature = "page_ecc")]
            let heal_locks: Vec<Arc<parking_lot::Mutex<()>>> = current_views
                .iter()
                .filter(|v| {
                    comparator.compare(v.metadata.key_range.max(), boundary)
                        != core::cmp::Ordering::Less
                })
                .map(Table::heal_lock_arc)
                .collect();
            #[cfg(feature = "page_ecc")]
            let _heal_guards: Vec<parking_lot::MutexGuard<'_, ()>> =
                heal_locks.iter().map(|l| l.lock()).collect();

            // Classify each input at this boundary: restrict (extends past it) or
            // remove (fully consumed by this slice). Re-open restricted views as
            // distinct Inners so a prior view drops and punches independently.
            //
            // `run_subcompaction` above finalized this slice's outputs, which only
            // the install below names: a failure before it (a probe-count transfer,
            // a restricted reopen) leaves them to the run, which removes what it
            // never installed as it returns. No `.restrict-bound` sidecar is written
            // before the install either: the sidecar write is STRICTLY POST-COMMIT
            // (the mark step below the install), so an aborted slice never publishes
            // one. That ordering is the crash-safety invariant: a sidecar on disk
            // always denotes a committed restriction (see the mark step and
            // `docs/manifest-recovery.md`).

            // The slice's outputs take the share of the probe counts the prefix
            // they rewrote holds; the restricted views keep the suffix's.
            crate::table::probe_stats::inherit_into(&outputs, &current_views)?;

            // Re-open each stale blob file as a distinct Inner: the re-opened view
            // replaces the original in the new version (same id), and the original
            // — held only by prior snapshots after this — punches its consumed
            // `[data_start, frontier)` prefix when it drains. Mirrors the SST
            // `reopen_restricted` swap. Files with no consumption yet are skipped
            // (nothing to punch, original stays installed).
            let mut prior_to_punch: Vec<(BlobFile, u64)> = Vec::new();
            if relocating {
                // Test-only failpoint: a restricted-blob reopen failure at this
                // exact point (outputs finalized, install not yet run).
                #[cfg(test)]
                if opts
                    .config
                    .fail_tight_blob_reopen
                    .swap(false, core::sync::atomic::Ordering::SeqCst)
                {
                    return Err(cancelled_compaction());
                }
                for sf in &current_stale {
                    if let Some(off) = resume_offsets.get(&sf.id()).copied() {
                        // The re-opened view serves only `[off, end)`: the prior
                        // view punches everything below once its readers drain.
                        // Capture the digest over that live suffix NOW, while the
                        // file is still whole, and record `off` as the view's
                        // frontier — a whole-file digest would never match the
                        // punched file, and integrity checks hash from the
                        // frontier for exactly this reason.
                        new_blobs.push(sf.reopen_restricted(off)?);
                        prior_to_punch.push((sf.clone(), off));
                    }
                }
            }

            let mut restricted_pairs: Vec<(TableId, Table)> = Vec::new();
            let mut removed_ids: Vec<TableId> = Vec::new();
            let mut next_views: Vec<Table> = Vec::new();
            for view in &current_views {
                if comparator.compare(view.metadata.key_range.max(), boundary)
                    == core::cmp::Ordering::Less
                {
                    removed_ids.push(view.id());
                } else {
                    // Open the restricted view over the same physical SST. This is an
                    // IN-MEMORY reopen (it reads the live suffix digest but writes
                    // nothing to disk), so a failure here commits nothing and just
                    // rolls back the finalized outputs. The `.restrict-bound` sidecar
                    // that records this view's exact bound is written only AFTER the
                    // install commits (the mark step below), never here.
                    //
                    // Later ordinary compactions read this view through
                    // `Table::scan()`, which starts at the restriction bound
                    // (blocks below it are punched and unreadable, and the
                    // prefix rows' authoritative copies live in the slice
                    // output) — so the restricted SST stays compactable and is
                    // eventually retired through the normal path.
                    // A restriction only ever RISES. The boundaries were built
                    // from the whole block index, punched prefix included, so an
                    // input a previous run already restricted can be handed a
                    // boundary BELOW its committed bound; reopening there would
                    // republish the SST as serving its own punched prefix, whose
                    // reads land in the zeros and whose rows the earlier slice
                    // output already owns.
                    let bound = match view.restrict_lower_bound() {
                        Some(committed)
                            if comparator.compare(committed, boundary)
                                == core::cmp::Ordering::Greater =>
                        {
                            committed.clone()
                        }
                        _ => boundary.clone(),
                    };
                    let restricted = view.reopen_restricted(bound)?;
                    restricted_pairs.push((view.id(), restricted.clone()));
                    next_views.push(restricted);
                }
            }

            // Bind the blob files this slice publishes, exactly as its SST
            // outputs were bound above — both the freshly written ones and the
            // re-opened restricted views, since the version edit below makes
            // all of them reachable at once. An unbound blob's `Drop` can
            // unlink it while a checkpoint is capturing, and its prefix punch
            // can zero bytes the checkpoint has already hard-linked.
            for blob_file in &new_blobs {
                blob_file.bind_to_tree(&crate::table::TableSinks {
                    deletion_pause: &opts.deletion_pause,
                    heal_hints: &opts.heal_hints,
                    read_budget: opts.config.columnar_read_budget,
                    #[cfg(feature = "std")]
                    background_deleter: &opts.background_deleter,
                    // A blob file has no filter.
                    track_filter_probes: false,
                });
            }

            #[cfg(test)]
            opts.config.fire_before_output_install();

            // The edit drops the fully consumed inputs' filters from the
            // published figure; the plan gives back its credit for them first.
            if let Some(sizing) = &filter_sizing {
                sizing.release_replaced();
            }
            // A stale file a memtable row still references keeps its whole
            // view and is not punched: the row borrows an object the slice
            // copied, and links the file from no table until it is flushed.
            // Decided under the install's version lock, which orders the
            // writes of such rows against this edit.
            let mut unpunched: crate::HashSet<BlobFileId> = crate::HashSet::default();
            // Install one atomic, durable version edit for the slice.
            // A failed install commits nothing, so no sidecar was written (the
            // mark step below is strictly post-commit) and the punches never arm;
            // the run removes the slice's outputs as it returns.
            let mut history = opts.version_history.write();
            // An input that left the version while the slice ran was removed
            // with its data (a `clear`): the slice's outputs hold that data.
            if !super::flavour::holds_every_input(
                &history.latest_version().version,
                current_views.iter().map(Table::id),
            ) {
                log::warn!(
                    "Compaction inputs left the version before a slice's install, declining it"
                );
                return Ok(tables_out);
            }
            history.upgrade_version(
                &opts.config.path,
                |sv| {
                    unpunched.extend(
                        prior_to_punch
                            .iter()
                            .map(|(bf, _)| bf.id())
                            .filter(|id| sv.memtables_reference_blob_file(*id)),
                    );
                    let new_blobs: Vec<BlobFile> = new_blobs
                        .into_iter()
                        .filter(|bf| !unpunched.contains(&bf.id()))
                        .collect();
                    let mut copy = sv.clone();
                    let ctx = crate::version::TransformContext::new(comparator.as_ref());
                    copy.version = copy.version.with_tight_slice(
                        &restricted_pairs,
                        &removed_ids,
                        &outputs,
                        new_blobs,
                        gc_diff,
                        payload.dest_level as usize,
                        &ctx,
                    );
                    Ok(copy)
                },
                &opts.global_seqno,
                &opts.visible_seqno,
                &*opts.config.fs,
                opts.runtime_config.load_full(),
                opts.encryption.clone(),
                // A slice is a merge with GC below the watermark, plus the
                // punched input prefix: older snapshots lose both.
                crate::version::RetentionEffect::GcBelow(opts.gc_watermark),
            )?;
            // Still under the write lock that published the slice: a reference
            // read from an older version names an object the slice let go of.
            if !released_objects.is_empty() {
                let published = history.latest_version().version.id();
                history.released().record(published, released_objects, []);
            }
            drop(history);
            // The version names this slice's outputs now: a later slice's failure
            // must not remove them.
            opts.outputs.installed();
            opts.work.installed(installed_work);
            installed_files.charge();
            #[cfg(feature = "metrics")]
            super::flavour::charge_installed(&opts.metrics, relocated_bytes, carried);
            #[cfg(not(feature = "metrics"))]
            let _ = (relocated_bytes, carried);
            // The published version counts the outputs' filters now, and the
            // plan's views must not outlive the slice.
            drop(filter_sizing);

            // Mark, then punch. The install committed, so now record each restricted
            // input's exact bound to its `.restrict-bound` sidecar — STRICTLY AFTER
            // the commit, so a sidecar on disk always denotes a committed restriction
            // (manifest repair honors it without a commit protocol). Then arm the
            // prefix punch. A fully-consumed input is deleted outright (its data is in
            // the outputs).
            //
            // A sidecar write that fails HERE is not fatal: the restriction is already
            // durable in the manifest, which is what the bound is recovered FROM —
            // the next open republishes any sidecar that never landed. Log it and
            // leave that input UNPUNCHED meanwhile, so a manifest-loss repair that
            // runs before that open sees no-sidecar + unpunched and keeps the whole
            // input rather than deriving a conservative bound and dropping up to one
            // live block. Keeping it whole is itself only tolerable transiently: the
            // repair would publish both the input and the slice output as L0 runs,
            // and while point reads are shadowed by the output, merge operands are
            // deliberately never deduplicated across sources, so an operand in the
            // consumed prefix WOULD be applied twice. Hence the republish on open —
            // never punch without the sidecar, and never leave it missing for good.
            //
            // The slice's sidecars are published together: each written and
            // synced, then renamed, then their folder synced once rather than
            // once per input. Each is durable when that sync returns, and each
            // input's punch waits for its own.
            let mut restricted: Vec<(&Table, &UserKey)> = Vec::new();
            for view in &current_views {
                if removed_ids.contains(&view.id()) {
                    view.mark_as_deleted();
                } else {
                    // The bound this slice COMMITTED for this input, which the
                    // reopen above may have clamped up to a restriction an
                    // earlier pass already committed. The sidecar is what a
                    // manifest-loss repair trusts, so recording the raw boundary
                    // here would republish the SST as serving an interval the
                    // earlier slice output owns and whose blocks are punched
                    // out. The punch offset follows the same bound (punching to
                    // the lower one would merely redo a subset of an existing
                    // hole, but the two must not disagree).
                    let bound = restricted_pairs
                        .iter()
                        .find(|(id, _)| *id == view.id())
                        .and_then(|(_, restricted)| restricted.restrict_lower_bound())
                        .unwrap_or(boundary);
                    restricted.push((view, bound));
                }
            }
            let writes: Vec<_> = restricted
                .iter()
                .map(|(view, bound)| view.restrict_sidecar_write(bound))
                .collect();
            let published = crate::restrict_bound::write_many(&writes, opts.config.sync_mode);
            drop(writes);
            for ((view, bound), result) in restricted.into_iter().zip(published) {
                // A failed sidecar write and a failed punch-offset lookup end the
                // same way. Both are post-commit, so neither is fatal: propagating
                // `?` here would abort an already-committed slice. The input is
                // left unpunched and a later compaction reclaims it.
                match result.and_then(|()| view.punch_offset_for(bound)) {
                    Ok(off) => view.mark_punch_on_drop(off),
                    // On a LATER slice of the same input, a sidecar failure leaves
                    // the PREVIOUS slice's sidecar in place — valid but
                    // stale (a lower bound). Deliberately left there: a
                    // manifest-loss repair before the next open then
                    // restricts to that lower bound and republishes only the
                    // newly consumed interval beside the slice output — the
                    // same already-tolerated duplication class as a FIRST
                    // slice's failure (whole unpunched input republished),
                    // and strictly less of it. REMOVING the stale sidecar
                    // gains nothing and can lose data: the punch geometry
                    // re-derives the SAME lower bound (the holes end where
                    // that slice punched), while a backend that cannot
                    // prove holes then classifies the zeroed prefix as
                    // damage and salvage drops live rows the slice output
                    // does not carry. With the manifest alive, the next
                    // open replaces a DISAGREEING sidecar from the
                    // manifest's bound, which closes this window.
                    Err(e) => log::error!(
                        "tight-space could not arm the prefix punch of table {} after \
                         committing its restriction: {e}; leaving the input unpunched, \
                         a later compaction rewrites it",
                        view.id(),
                    ),
                }
            }
            // Arm each re-opened stale blob file's prior view to punch its
            // relocated `[data_start, frontier)` prefix once it drains.
            for (bf, off) in &prior_to_punch {
                if !unpunched.contains(&bf.id()) {
                    bf.mark_punch_on_drop(*off);
                }
            }

            tables_out += outputs.len();
            // The views this slice let go of: their prefix punch, or their
            // deletion, runs once nothing holds them.
            let let_go_tables: Vec<_> = current_views.iter().map(Table::downgrade).collect();
            let let_go_blobs: Vec<_> = prior_to_punch
                .iter()
                .filter(|(bf, _)| !unpunched.contains(&bf.id()))
                .map(|(bf, _)| bf.downgrade())
                .collect();
            current_views = next_views;
            lower = Bound::Included(boundary.clone());

            // Drop this iteration's handles to the stale Inners: the install
            // above replaced the version that held them, so the originals drop
            // and punch NOW, reclaiming space before the next slice.
            drop(prior_to_punch);
            drop(current_stale);
            // The next slice is cut to the room this one gives back. A reader
            // still holding the version before it, or a checkpoint pausing
            // deletions, keeps that room taken for now: the pass stops here,
            // its slices installed, and the next compaction picks up the rest
            // once it is reclaimed.
            reclaim_deferred = let_go_tables.iter().any(|view| view.strong_count() > 0)
                || let_go_blobs.iter().any(|view| view.strong_count() > 0)
                || opts.deletion_pause.is_active();

            // Test-only crash point: abort right after the first slice is durably
            // installed and punched, exercising the reopen-with-restriction path.
            #[cfg(test)]
            if opts
                .config
                .fail_tight_after_first_slice
                .swap(false, core::sync::atomic::Ordering::SeqCst)
            {
                return Err(cancelled_compaction());
            }

            if current_views.is_empty() {
                // Every input was consumed before the last boundary.
                return Ok(tables_out);
            }
        }

        // Tail [last boundary, hi): merge the remainder, remove the inputs, and
        // drop the now fully-consumed stale blob files (install_merge deletes
        // them outright — the prior slices already punched their prefixes).
        if reclaim_deferred || !reserve_slice(tail, &own_tables, &own_blobs)? {
            stopped = true;
            return Ok(tables_out);
        }
        let slice_payload = slice_payload_for(&current_views, payload);
        let version = opts.version_history.read().latest_version();
        let tail_reloc = if relocating {
            let current_stale: Vec<BlobFile> = stale_ids
                .iter()
                .filter_map(|id| version.version.blob_files.get(*id).cloned())
                .collect();
            Some(RelocationSetup {
                stale_files: current_stale,
                resume_offsets: resume_offsets.clone(),
            })
        } else {
            None
        };
        let produced = run_subcompaction(
            opts,
            &slice_payload,
            &version.version,
            current_views.clone(),
            &rts,
            None,
            (lower.clone(), Bound::Unbounded),
            dst_lvl,
            bottommost_gc,
            // Slices apply no removal semantics; see the parameter docs.
            false,
            &blobs_folder,
            tail_reloc,
            &rc,
            super::flavour::plan_filters(
                &version.version,
                opts,
                &slice_payload,
                &[],
                Some((lower.clone(), Bound::Unbounded)),
                &rc,
            ),
        )?;
        drop(version);
        let tail_out = produced.created_tables().len();
        let installed = super::flavour::install_merge(
            &mut opts.version_history.write(),
            opts,
            &slice_payload,
            vec![produced],
        )?;
        Ok(tables_out + installed.map_or(0, |_| tail_out))
    })();
    // Held across every slice's install: the installed outputs name their
    // dictionaries from here on (see `WritePin`).
    drop(rc);

    compaction_state
        .hidden_set_mut()
        .show(payload.table_ids.iter().copied());

    let tables_out = result?;
    if stopped && tables_out == 0 {
        log::info!("Tight-space compaction stopped: its first slice no longer fits the room left");
        return Ok(CompactionResult::nothing());
    }

    Ok(CompactionResult {
        action: CompactionAction::Merged,
        dest_level: Some(payload.dest_level),
        tables_in,
        tables_out,
        work: super::WorkReport::default(),
    })
}

/// Per-slice relocation inputs for the tight-space blob-defrag path: the stale
/// blob files to rewrite and, per file, the absolute data-section offset to
/// resume its scan at (the prior slice's consumed frontier, with everything
/// before it already hole-punched).
#[cfg(feature = "std")]
struct RelocationSetup {
    stale_files: Vec<BlobFile>,
    resume_offsets: crate::HashMap<BlobFileId, u64>,
}

/// Runs one sub-compaction over the key range `bounds`: builds a bounded merge
/// stream of the inputs, applies the same transforms as the serial path
/// (tombstone eviction, KV-separation drop tracking, compaction filter), writes
/// its output SSTs, and returns a [`ProducedOutput`](super::flavour::ProducedOutput)
/// WITHOUT installing a version edit (the caller merges all outputs into one).
/// Only the sub-compaction that `owns_input_deletion` carries the input tables
/// to be dropped, so the shared inputs are marked deleted exactly once.
#[cfg(feature = "std")]
#[expect(
    clippy::too_many_arguments,
    reason = "a sub-compaction needs the full compaction context (opts, payload, \
              version, inputs, range, level info, blob folder) threaded in by value/ref \
              so it can run on its own thread; bundling into a struct would just move \
              the argument list"
)]
fn run_subcompaction(
    opts: &Options,
    payload: &CompactionPayload,
    version: &Version,
    tables_for_deletion: Vec<Table>,
    input_range_tombstones: &[crate::range_tombstone::RangeTombstone],
    // The keys the tables outside the compaction hold, when the caller built
    // them once for sub-ranges sharing `version`; `None` builds them here.
    outside_spans: Option<&OutsideSpans>,
    bounds: (core::ops::Bound<UserKey>, core::ops::Bound<UserKey>),
    dst_lvl: usize,
    is_last_level: bool,
    // Whether the user compaction filter runs in this merge. `false` on
    // tight-space slices, which apply NO removal semantics (bottommost GC is
    // deferred the same way): a slice's output must be a SUPERSET that shadows
    // any surviving input prefix in every crash window — a post-commit sidecar
    // write failure deliberately leaves the input unpunched, and a
    // manifest-loss repair then republishes that prefix whole, so a record the
    // filter removed here would survive nowhere else and resurrect. The
    // filtering is deferred, not lost: the next normal compaction applies it.
    apply_compaction_filter: bool,
    blobs_folder: &std::path::Path,
    // When `Some` (tight-space blob defrag only), this sub-compaction relocates
    // the live entries of `stale_files` into a fresh compact blob file, resuming
    // each stale file's scan at `resume_offsets` so an already-punched prefix is
    // never re-read. `None` is the pass-through path (no blob relocation).
    relocation: Option<RelocationSetup>,
    // The compaction's runtime snapshot, shared by all of its sub-compactions
    // (or slices): one configuration for everything the compaction writes, the
    // filter's blob files included, held until the output is installed.
    rc: &Arc<crate::runtime_config::RuntimeConfig>,
    // The compaction's filter plan, shared by all of its sub-compactions (or
    // slices) so they draw on one reservation of the filter budget.
    filter_sizing: Option<crate::filter_budget::FilterPlan>,
) -> crate::Result<super::flavour::ProducedOutput> {
    use super::flavour::CompactionFlavour;

    // Test-only failpoint: the first range to observe an armed flag fails (and
    // disarms it), so exactly one sub-compaction errors while its siblings
    // succeed — deterministically driving the rollback path that marks the
    // siblings' finalized files deleted and restores the hidden inputs.
    #[cfg(test)]
    if opts
        .config
        .fail_one_subcompaction
        .swap(false, std::sync::atomic::Ordering::SeqCst)
    {
        return Err(cancelled_compaction());
    }

    // The blob accounting of the dropped versions, shared by the stream that
    // reports them and the loop that settles a key's ownership before its
    // kept rows are written (see `ownership`).
    let mut ownership = super::ownership::OwnershipLedger::default();
    if let Some(reloc) = &relocation {
        ownership.relocating(reloc.stale_files.iter().map(BlobFile::id));
    }
    let ledger = core::cell::RefCell::new(ownership);
    let ledger_error = core::cell::RefCell::new(None);
    let mut ledger_hook = super::ownership::LedgerHook {
        ledger: &ledger,
        error: &ledger_error,
    };

    // Declared before the stream that borrows it, so it outlives it.
    let completeness = input_completeness_for(version, &payload.table_ids, &opts.config);

    // A relocating slice ends at an exclusive key: its frames below that key
    // that no written pointer claimed are drained into the frontier once the
    // slice is written (see `RelocatingCompaction::drain_unclaimed_below`).
    let drain_below = match (&relocation, &bounds.1) {
        (Some(_), core::ops::Bound::Excluded(upper)) => Some(upper.clone()),
        _ => None,
    };
    // The keys this range's writer answers for, past its outputs included.
    let answered = answered_span(&bounds);
    // This range's writer does not wait for a boundary past its upper bound,
    // which another range's writer reaches.
    let range_upper = match &bounds.1 {
        core::ops::Bound::Included(upper) | core::ops::Bound::Excluded(upper) => {
            Some(upper.clone())
        }
        core::ops::Bound::Unbounded => None,
    };

    // Row groups and data blocks of this range the merge leaves as they were
    // are copied into the output instead of being encoded again (see
    // `carry`). Each range records and matches its own. A relocating slice
    // writes every row itself, so its scans record nothing.
    let carry = if relocation.is_some() {
        None
    } else {
        super::carry::plan(
            &opts.config,
            rc,
            payload.canonical_level.into(),
            &version
                .iter_tables()
                .filter(|t| payload.table_ids.contains(&t.id()))
                .cloned()
                .collect::<Vec<_>>(),
        )
    };
    let pace = input_pacer(opts);

    let Some(mut merge_iter) = create_bounded_compaction_stream(
        version,
        &payload.table_ids,
        bounds,
        opts.gc_watermark,
        opts.config.merge_operator.clone(),
        opts.config.comparator.clone(),
        Some(&pace),
        carry.as_ref(),
    )?
    else {
        // The caller validated every input exists, so a missing table here is
        // unexpected. Fail closed: an empty output would let the install delete
        // the source tables (this range may own input deletion) while producing
        // no replacement SSTs. Returning an error makes the parallel caller
        // re-show the inputs and skip the install entirely.
        return Err(crate::Error::from(crate::io::Error::other(
            "sub-compaction input tables disappeared mid-flight",
        )));
    };

    // Whole-version range tombstones drive compaction-time RT application (drop
    // covered KVs in the merge, with blob-GC accounting), merge-chain
    // boundaries for a composing operator, and the bottommost seqno-zeroing
    // gate below. Gathered from every level so coverage outside this compaction
    // is respected.
    let rt_plan = range_tombstone_plan(version, is_last_level, &opts.config);

    // Dropping a tombstone, zeroing a seqno and folding operands onto an absent
    // base all read "not in my inputs" as "not in the tree", so they need the
    // stronger premise than the destination level alone.
    let holds_everything = holds_every_surviving_version(
        version,
        &payload.table_ids,
        is_last_level,
        opts.config.comparator.as_ref(),
    );

    merge_iter = merge_iter
        .evict_tombstones(holds_everything)
        .zero_seqnos(false);
    if let Some((tombstones, rt_use)) = &rt_plan {
        merge_iter = merge_iter.with_range_tombstones(
            tombstones.clone(),
            opts.config.comparator.clone(),
            *rt_use,
        );
    }
    if let Some(predicate) = &completeness {
        merge_iter = merge_iter.with_input_completeness(predicate);
    }
    let version_tombstones = zeroing_tombstones_from(rt_plan.as_ref());

    let filter_ctx = Context { is_last_level };
    let mut compaction_filter = if apply_compaction_filter {
        opts.config
            .compaction_filter_factory
            .as_ref()
            .map(|f| f.make_filter(&filter_ctx))
    } else {
        None
    };
    // Shared transform counter, created for EVERY sub-compaction: the filter
    // adapter ticks it on non-Keep verdicts, the merge stream on its own
    // visibility-changing drops (bottommost tombstone elision, RT
    // application, weak annihilation), and the table writer marks
    // transformed the lineage of any output whose window saw one (see
    // `prepare_table_writer`).
    let transform_marker = Some(Arc::new(portable_atomic::AtomicU64::new(0)));
    if let Some(marker) = &transform_marker {
        merge_iter = merge_iter.with_transform_marker(Arc::clone(marker));
    }

    // KV separation (no relocation on this path): track fragmentation from
    // dropped/GC'd entries so the merged install updates blob GC stats.
    if opts.config.kv_separation_opts.is_some() {
        merge_iter = merge_iter.with_drop_callback(&mut ledger_hook);
    }

    // Versions that go in and do not come out: read after `produce` to tell a
    // run that collected history from one that collected none.
    let gc_balance = merge_iter.gc_balance();

    let mut filter_blob_writer = None;
    // A relocating pass separates the filter's new values into its own blob
    // output, beside the frames it relocates (see `FilterBlobOutput`).
    let relocating = relocation.is_some() && opts.config.kv_separation_opts.is_some();
    // Filter-only transform counter (see `TransformCounters::filter`): read
    // after `produce` to decide the install's retention effect.
    let filter_marker = Arc::new(portable_atomic::AtomicU64::new(0));
    let merge_iter = merge_iter.with_filter(StreamFilterAdapter::new(
        compaction_filter.as_deref_mut(),
        opts,
        version,
        FilterBlobOutput {
            folder: blobs_folder,
            writer: (!relocating).then_some(&mut filter_blob_writer),
            runtime: rc,
        },
        &filter_ctx,
        TransformCounters {
            transform: transform_marker.clone(),
            filter: Arc::clone(&filter_marker),
        },
    ));

    // Bottommost seqno-zeroing: at the last level, entries below the GC
    // watermark and not covered by any range tombstone get seqno 0 (packs to
    // one byte). Tombstones are gathered from the whole version so coverage in
    // levels outside this compaction still blocks zeroing.
    let merge_iter = super::seqno_zeroer::BottommostSeqnoZeroer::new(
        merge_iter,
        holds_everything,
        version_tombstones,
        opts.gc_watermark,
        opts.config.comparator.clone(),
        Arc::clone(&gc_balance),
    );

    // block_parallel = false: this sub-compaction already runs on a pool thread,
    // so its block compression must stay serial (nested-pool deadlock otherwise).
    // whole_run = false: several sub-compactions (or tight-space slices) share
    // this payload's lineage, so no single writer may close the run.
    let mut table_writer = super::flavour::prepare_table_writer(
        version,
        opts,
        payload,
        filter_sizing,
        false,
        transform_marker,
        false,
        rc,
    )?;
    if let Some(upper) = range_upper {
        table_writer.limit_cut_alignment(upper);
    }
    let mut compactor: Box<dyn CompactionFlavour> = match relocation {
        // Tight-space blob defrag: relocate the stale files' live entries into a
        // fresh compact file, resuming each stale scan at its carried-over
        // frame boundary (the prior slice punched everything before it).
        Some(reloc) if opts.config.kv_separation_opts.is_some() => {
            #[expect(clippy::expect_used, reason = "guarded by is_some() above")]
            let blob_opts = opts
                .config
                .kv_separation_opts
                .as_ref()
                .expect("kv_separation_opts present");

            let scanner = BlobFileMergeScanner::new(
                reloc
                    .stale_files
                    .iter()
                    .map(|bf| {
                        let from = reloc.resume_offsets.get(&bf.id()).copied().unwrap_or(0);
                        open_blob_scanner_at_frontier(bf, from, opts)
                    })
                    .collect::<crate::Result<Vec<_>>>()?,
                opts.config.comparator.clone(),
            );

            let writer = BlobFileWriter::new_for(
                opts.blob_file_id_generator.clone(),
                blobs_folder,
                opts.tree_id,
                opts.config.descriptor_table.clone(),
                opts.config.fs.clone(),
                crate::fs::WriteOrigin::CompactionBlob,
            )?
            .use_output_ledger(opts.outputs.clone())
            .use_target_size(blob_opts.file_target_size)
            .use_passthrough_compression(rc.blob_compression)
            .use_sync_mode(opts.config.sync_mode)
            .use_writeback_bytes(opts.config.writeback_bytes)
            .use_lifetime_groups(blob_opts.lifetime_groups)
            .use_clock(opts.config.clock.clone())
            .use_key_domains(
                opts.config.key_domains.clone(),
                opts.config.comparator.clone(),
            );
            // The policy here is only the OPENING value. Relocation copies
            // frames verbatim out of files that may predate a policy change, so
            // `RelocatingCompaction` records each output file's codec from the
            // SOURCE its frames came from and rotates when that changes;
            // stamping the current policy on borrowed bytes is precisely the
            // mislabelling that rule exists to prevent. The set is what pins the
            // dictionary that recorded codec names, since this writer
            // compresses nothing of its own.
            let writer = with_fresh_encoding(writer, opts, rc);

            let inner = StandardCompaction::new(table_writer, tables_for_deletion);
            Box::new(
                RelocatingCompaction::new(
                    inner,
                    scanner.peekable(),
                    writer,
                    reloc.stale_files,
                    opts.rate_limiter.clone(),
                    opts.stop_signal.clone(),
                    opts.config.comparator.clone(),
                    opts.tree_id,
                )
                .with_drain_below(drain_below)
                .with_fresh_values(FreshValues {
                    options: blob_opts.clone(),
                    compression: rc.blob_compression,
                }),
            )
        }
        _ => Box::new(StandardCompaction::new(table_writer, tables_for_deletion)),
    };

    // Propagate range tombstones to this sub-range's output (GC'd at the last
    // level when fully applied); the writer clips them to each output table's
    // key range (a boundary-spanning RT is written clipped on both sides).
    if !input_range_tombstones.is_empty() {
        // The sub-ranges of one compaction share its spans; a slice reads a
        // version of its own and builds them here.
        let built;
        let outside_spans = if let Some(spans) = outside_spans {
            spans
        } else {
            built = OutsideSpans::new(version, &payload.table_ids, opts.config.comparator.as_ref());
            &built
        };
        let output_rts = range_tombstones_after_gc(
            input_range_tombstones,
            outside_spans,
            opts.gc_watermark,
            is_last_level,
            &opts.config.comparator,
        );
        if !output_rts.is_empty() {
            let outside = tombstones_outside(
                &output_rts,
                outside_spans,
                answered,
                &opts.config.comparator,
            );
            compactor.write_range_tombstones(&output_rts, outside);
        }
    }

    let mut matcher = compactor.carrying().and(carry.as_ref()).map(|carry| {
        super::carry::CarryMatcher::new(Arc::clone(&carry.queue), opts.config.comparator.clone())
    });
    let read_pace: Option<&dyn crate::table::util::ReadPacer> = Some(&*pace);
    // Chosen once, so a compaction that copies nothing pays no check per row.
    let mut carrying_emit;
    let mut plain_emit;
    let emit: &mut dyn FnMut(InternalValue) -> crate::Result<()> = if let Some(matcher) =
        matcher.as_mut()
        && let Some(flavour) = compactor.carrying()
    {
        carrying_emit = move |row: InternalValue| {
            matcher.write(
                row,
                &mut FlavourSink {
                    flavour: &mut *flavour,
                    pace: read_pace,
                },
            )
        };
        &mut carrying_emit
    } else {
        plain_emit = |row: InternalValue| compactor.write(row);
        &mut plain_emit
    };

    for (idx, item) in merge_iter.enumerate() {
        let item = item?;

        // One key + value length; the sum is far below u64::MAX → plain add.
        let io_bytes = item.key.user_key.len() as u64 + item.value.len() as u64;
        if opts
            .rate_limiter
            .request_abortable(io_bytes, || opts.stop_signal.is_stopped())
        {
            // Abort, do not produce: a truncated sub-range would be installed
            // atomically alongside its siblings, dropping the unwritten tail of
            // this range (a gap in the middle of the key space). The error makes
            // the caller re-show the inputs and skip the install.
            return Err(cancelled_compaction());
        }

        ledger.borrow_mut().admit(item, &mut *emit)?;
        if let Some(e) = ledger_error.borrow_mut().take() {
            return Err(e);
        }

        if idx % 1_000_000 == 0 && opts.stop_signal.is_stopped() {
            return Err(cancelled_compaction());
        }
    }

    if let Some(filter) = compaction_filter {
        filter.finish();
    }

    let extra_blob_files = filter_blob_writer
        .map(BlobFileWriter::finish)
        .transpose()?
        .unwrap_or_default();

    let (blob_frag_map, released_objects) = ledger.into_inner().finish(&mut *emit)?;
    if let Some(e) = ledger_error.into_inner() {
        return Err(e);
    }
    let carried = match (matcher.take(), compactor.carrying()) {
        (Some(matcher), Some(flavour)) => matcher.finish(&mut FlavourSink {
            flavour,
            pace: read_pace,
        })?,
        _ => super::carry::Carried::default(),
    };
    let mut produced = compactor.produce(opts, dst_lvl, blob_frag_map, extra_blob_files)?;
    produced.set_released_objects(released_objects);
    produced.set_carried(carried);
    if filter_marker.load(core::sync::atomic::Ordering::Relaxed) > 0 {
        produced.mark_filter_transformed();
    }
    if gc_balance.load(core::sync::atomic::Ordering::Relaxed) > 0 {
        produced.mark_collected_below_watermark();
    }
    Ok(produced)
}

#[expect(
    clippy::significant_drop_tightening,
    reason = "the hidden-set check and the install must run under one version lock"
)]
fn move_tables(
    compaction_state: &CompactionGuard<'_>,
    opts: &Options,
    payload: &CompactionPayload,
) -> crate::Result<CompactionResult> {
    #[cfg(test)]
    opts.config.fire_before_output_install();

    let mut version_history_lock = opts.version_history.write();

    // Fail-safe for buggy compaction strategies
    if compaction_state
        .hidden_set()
        .should_decline_compaction(payload.table_ids.iter().copied())
    {
        log::warn!(
            "Compaction task created by {:?} contained hidden tables, declining to run it - please report this at https://github.com/fjall-rs/lsm-tree/issues/new?template=bug_report.md",
            opts.strategy.get_name(),
        );
        return Ok(CompactionResult::nothing());
    }

    // A table that left the version since the move was planned (a `clear`
    // removed it) has nowhere to move from.
    if !super::flavour::holds_every_input(
        &version_history_lock.latest_version().version,
        payload.table_ids.iter().copied(),
    ) {
        log::warn!("Tables to move left the version before the install, declining the move");
        return Ok(CompactionResult::nothing());
    }

    let table_count = payload.table_ids.len();
    let table_ids = payload.table_ids.iter().copied().collect::<Vec<_>>();
    let estimate = {
        let version = &version_history_lock.latest_version().version;
        super::work::estimate_move(table_ids.iter().filter_map(|&id| version.get_table(id)))
    };
    opts.work.plan(estimate);

    version_history_lock.upgrade_version(
        &opts.config.path,
        |current| {
            let mut copy = current.clone();

            let ctx = crate::version::TransformContext::new(opts.config.comparator.as_ref());
            copy.version = copy
                .version
                .with_moved(&table_ids, payload.dest_level as usize, &ctx);

            Ok(copy)
        },
        &opts.global_seqno,
        &opts.visible_seqno,
        &*opts.config.fs,
        opts.runtime_config.load_full(),
        opts.encryption.clone(),
        // A trivial move rewrites nothing: every version survives as-is.
        crate::version::RetentionEffect::Keep,
    )?;
    // The estimate of a move is the move.
    opts.work.moved(estimate.expected.moved);

    Ok(CompactionResult {
        action: CompactionAction::Moved,
        dest_level: Some(payload.dest_level),
        tables_in: table_count,
        tables_out: table_count,
        work: super::WorkReport::default(),
    })
}

/// Picks blob files to rewrite (defragment): the linked, stale, non-dead blob
/// files a compaction of `picked_tables` would relocate. Also the basis for the
/// `storage_stats` blob-relocation estimate, so the reported status budgets the
/// SAME stale subset the gate does (not every live blob file).
// `pub` (not `pub(crate)`) inside this crate-private module: clippy flags the
// redundant restriction since the module itself is already crate-scoped.
pub fn pick_blob_files_to_rewrite(
    picked_tables: &HashSet<TableId>,
    current_version: &Version,
    blob_opts: &crate::KvSeparationOptions,
) -> crate::Result<Vec<BlobFile>> {
    use crate::Table;

    // We start off by getting all the blob files that are referenced by the tables
    // that we want to compact.
    let linked_blob_files = picked_tables
        .iter()
        .map(|&id| {
            current_version.get_table(id).unwrap_or_else(|| {
                panic!("Table {id} should exist");
            })
        })
        .map(Table::list_blob_file_references)
        .collect::<Result<Vec<_>, _>>()?;

    // Then we filter all blob files that are not fragmented or old enough.
    let mut linked_blob_files = linked_blob_files
        .into_iter()
        .flatten()
        .flatten()
        .map(|blob_file_ref| {
            current_version
                .blob_files
                .get(blob_file_ref.blob_file_id)
                .unwrap_or_else(|| {
                    panic!("Blob file {} should exist", blob_file_ref.blob_file_id);
                })
        })
        .filter(|blob_file| {
            blob_file.is_stale(current_version.gc_stats(), blob_opts.staleness_threshold)
        })
        .filter(|blob_file| {
            // NOTE: Dead blob files are dropped anyway during current_version change commit
            !blob_file.is_dead(current_version.gc_stats())
        })
        .collect::<HashSet<_>>()
        .into_iter()
        .collect::<Vec<_>>();

    linked_blob_files.sort_by_key(|a| a.id());

    #[expect(
        clippy::cast_precision_loss,
        clippy::cast_possible_truncation,
        clippy::cast_sign_loss,
        reason = "precision loss and truncation are acceptable for cutoff calculation"
    )]
    let cutoff_point = {
        let len = linked_blob_files.len() as f32;
        (len * blob_opts.age_cutoff) as usize
    };
    linked_blob_files.drain(cutoff_point..);

    // IMPORTANT: Additionally, we also have to check if any other tables reference any of our candidate blob files.
    // We have to *not* include blob files that are referenced by other tables, because otherwise those
    // blob references would point into nothing (becoming dangling).
    for table in current_version.iter_tables() {
        if picked_tables.contains(&table.id()) {
            continue;
        }

        let other_refs = table
            .list_blob_file_references()?
            .unwrap_or_default()
            .into_iter()
            .filter(|x| linked_blob_files.iter().any(|bf| bf.id() == x.blob_file_id))
            .collect::<Vec<_>>();

        for additional_ref in other_refs {
            linked_blob_files.retain(|x| x.id() != additional_ref.blob_file_id);
        }
    }

    Ok(linked_blob_files.into_iter().cloned().collect::<Vec<_>>())
}

/// The blob files a merge relocates.
struct BlobRewrite {
    files: Vec<BlobFile>,
    /// Some of `files` are relocated for locality rather than for staleness.
    #[cfg_attr(
        not(feature = "std"),
        expect(dead_code, reason = "read only by the std merge-on-read fast path")
    )]
    for_locality: bool,
}

/// Blob files a merge of `tables` relocates: the stale set, then, when locality
/// relocation is on, the files interleaved too deeply for scans.
fn pick_blob_files_for_merge(
    opts: &Options,
    version: &Version,
    payload: &CompactionPayload,
    tables: &[Table],
) -> crate::Result<BlobRewrite> {
    let Some(blob_opts) = &opts.config.kv_separation_opts else {
        return Ok(BlobRewrite {
            files: Vec::new(),
            for_locality: false,
        });
    };
    let mut files = pick_blob_files_to_rewrite(&payload.table_ids, version, blob_opts)?;
    let mut for_locality = false;
    if let Some(locality) = blob_opts.locality_relocation {
        let extra = pick_blob_files_for_locality(opts, version, payload, tables, locality, &files)?;
        for_locality = !extra.is_empty();
        files.extend(extra);
    }
    Ok(BlobRewrite {
        files,
        for_locality,
    })
}

/// `writer`, a relocation's blob output, set up to also encode the new values
/// the pass separates (see [`FreshValues`]): the tree's dictionary set pins
/// the dictionary each output file records, and the dictionary of the
/// snapshot's blob policy, with its two-pass setting, encodes the new values.
#[cfg_attr(
    not(zstd_any),
    expect(
        clippy::missing_const_for_fn,
        reason = "the zstd build reads the dictionaries"
    )
)]
fn with_fresh_encoding(
    writer: BlobFileWriter,
    opts: &Options,
    rc: &crate::runtime_config::RuntimeConfig,
) -> BlobFileWriter {
    #[cfg(zstd_any)]
    {
        let dicts = opts.config.current_zstd_dictionaries();
        // Without the policy's dictionary only a new value's write fails, as
        // the filter's own writer did on its first value: a pass that
        // separates nothing new does not need it.
        let dictionary = dicts.for_compression(rc.blob_compression).ok().flatten();
        writer
            .use_zstd_dictionary(dictionary)
            .use_zstd_two_pass_seed(rc.zstd_two_pass_seed)
            .use_zstd_dictionaries(dicts)
    }
    #[cfg(not(zstd_any))]
    {
        let _ = (opts, rc);
        writer
    }
}

/// Opens a relocation scan of `bf` at `from`, or at the frontier the file
/// itself declares when that lies further.
///
/// A relocation that committed a slice and then aborted leaves the file
/// restricted with its consumed prefix punched. Reading from the data section
/// would hit those zeros, resynchronize byte-wise and taint every surviving
/// frame, so no later merge could relocate the file.
fn open_blob_scanner_at_frontier(
    bf: &BlobFile,
    from: u64,
    opts: &Options,
) -> crate::Result<BlobFileScanner> {
    let off = from.max(bf.live_data_start());
    // The relocation reads the blob files of its inputs, their trailer and
    // table of contents included: they report their latency as the table
    // reads do, and count to the run's work.
    BlobFileScanner::open_paced(
        &bf.0.path,
        &*bf.0.fs,
        bf.id(),
        (off != 0).then_some(off),
        Some(input_pacer(opts)),
    )
}

/// Whether merging a move's tables instead would relocate blob files for
/// locality. Stale files alone do not turn a move into a merge.
fn moves_into_locality_relocation(
    version: &Version,
    opts: &Options,
    payload: &CompactionPayload,
) -> crate::Result<bool> {
    let Some(blob_opts) = &opts.config.kv_separation_opts else {
        return Ok(false);
    };
    let Some(locality) = blob_opts.locality_relocation else {
        return Ok(false);
    };
    let Some(tables) = payload
        .table_ids
        .iter()
        .map(|&id| version.get_table(id).cloned())
        .collect::<Option<Vec<_>>>()
    else {
        return Ok(false);
    };
    let stale = pick_blob_files_to_rewrite(&payload.table_ids, version, blob_opts)?;
    // No room is reserved before the move is decided, so the pick only
    // checks the pool here.
    Ok(
        !pick_blob_files_for_locality(opts, version, payload, &tables, locality, &stale)?
            .is_empty(),
    )
}

/// Blob files to relocate, beyond the stale ones already chosen, so a scan
/// over the merged tables interleaves fewer files.
///
/// Relocation writes the chosen files' values in key order, so files whose
/// key spans overlap come out as consecutive runs. Candidates are the files
/// that take part in a region deeper than the limit, taken by depth per
/// on-disk byte while reading and writing them fits the budget. Only whole
/// files that no table outside the merge references are eligible, and
/// nothing is picked when the output would not leave the reserved free space
/// intact. What is picked is taken into the run's room, when it holds one.
fn pick_blob_files_for_locality(
    opts: &Options,
    version: &Version,
    payload: &CompactionPayload,
    tables: &[Table],
    locality: crate::config::BlobLocalityRelocation,
    stale: &[BlobFile],
) -> crate::Result<Vec<BlobFile>> {
    let max_depth = locality.max_depth.get();

    // Cheapest first: a merge whose inputs stay within the limit, the
    // consecutive layout included, never pays for more than the figure.
    let depths = crate::storage_stats::blob_file_depths(tables)?;
    log::trace!("Blob file depths of the merge: {depths:?}");
    let mut candidates: Vec<(BlobFile, u64, u64)> = Vec::new();
    for (id, depth) in depths {
        if depth <= max_depth || stale.iter().any(|bf| bf.id() == id) {
            continue;
        }
        let Some(blob_file) = version.blob_files.get(id) else {
            continue;
        };
        if blob_file.is_dead(version.gc_stats()) {
            continue;
        }
        candidates.push((blob_file.clone(), depth, blob_file.physical_size()?));
    }
    if candidates.is_empty() {
        return Ok(Vec::new());
    }

    // A file another table still points into cannot move: its references
    // would dangle.
    for table in outside_tables(version, &payload.table_ids) {
        let links = table.blob_links()?;
        candidates.retain(|(bf, _, _)| !links.iter().any(|link| link.blob_file_id == bf.id()));
    }

    // The bytes the merge writes anyway: its tables, bounded by the length of
    // their input files (a restricted input's punched prefix excluded), plus
    // what relocating the stale files writes, which is their values these
    // tables still reference rather than the files' whole length. Table
    // files are measured by length, as blob files are: the recorded table size
    // leaves out sections written after the data. Sums of on-disk sizes are
    // bounded by filesystem capacity and cannot overflow.
    let mut table_bytes = 0u64;
    let mut stale_written = 0u64;
    for table in tables {
        let len = table.file_len();
        table_bytes += len.checked_sub(table.punch_offset()?).unwrap_or(len);
        for link in table.blob_links()? {
            if stale.iter().any(|bf| bf.id() == link.blob_file_id) {
                stale_written += link.on_disk_bytes;
            }
        }
    }
    // A float-to-int cast saturates, and NaN or a negative budget gives zero.
    #[expect(
        clippy::cast_possible_truncation,
        clippy::cast_sign_loss,
        clippy::cast_precision_loss,
        reason = "the budget is a fraction of a byte count; rounding is fine"
    )]
    let budget = ((table_bytes + stale_written) as f64 * f64::from(locality.budget)) as u64;

    // Most depth per byte first; ties go to the older file.
    candidates.sort_by(|(a, a_depth, a_size), (b, b_depth, b_size)| {
        let a_score = u128::from(*a_depth) * u128::from((*b_size).max(1));
        let b_score = u128::from(*b_depth) * u128::from((*a_size).max(1));
        b_score.cmp(&a_score).then_with(|| a.id().cmp(&b.id()))
    });

    // A file rewritten without another it overlaps keeps its span: its
    // interleaving stays, and joining it to a disjoint run would only stretch
    // that run's span across the gap. The partner must share its codec too:
    // relocation copies frames verbatim, one output file per source codec, so
    // files of different codecs stay apart however they overlap.
    let mut codecs: Vec<crate::CompressionType> = Vec::new();
    let mut group_of = |bf: &BlobFile| {
        let codec = bf.compression();
        codecs.iter().position(|c| *c == codec).unwrap_or_else(|| {
            codecs.push(codec);
            codecs.len() - 1
        })
    };
    let mut fixed: Vec<(crate::vlog::BlobFileId, usize)> =
        stale.iter().map(|bf| (bf.id(), group_of(bf))).collect();
    fixed.sort_unstable_by_key(|&(id, _)| id);
    // Each byte is read once and written once.
    let ranked: Vec<(crate::vlog::BlobFileId, usize, u64)> = candidates
        .iter()
        .map(|(bf, _, size)| (bf.id(), group_of(bf), 2 * size))
        .collect();
    let (chosen, spent) =
        crate::storage_stats::pick_overlapping_blob_files(tables, &fixed, &ranked, budget)?;
    let picked: Vec<(BlobFile, u64)> = candidates
        .into_iter()
        .zip(chosen)
        .filter_map(|((bf, _, size), chosen)| chosen.then_some((bf, size)))
        .collect();

    log::debug!(
        "Locality relocation: {} of the candidates fit {spent} of {budget} budget bytes",
        picked.len(),
    );
    if picked.is_empty() {
        return Ok(Vec::new());
    }

    // The merge's own tables and stale files are in the room it reserved, so
    // the files picked here are what it adds: each charged its whole length,
    // since the demand is not known more closely before the merge runs.
    let picked_bytes: u64 = picked.iter().map(|(_, size)| size).sum();
    let picked: Vec<BlobFile> = picked.into_iter().map(|(bf, _)| bf).collect();

    let rc = opts.runtime_config.load_full();
    let quota_headroom = match rc.storage_limit_bytes {
        Some(limit) => limit.saturating_sub(crate::storage_stats::compute_used_bytes(version)?),
        None => u64::MAX,
    };
    let blob_dir = opts.config.path.join(BLOBS_FOLDER);
    let extra = super::capacity::Demand {
        fs: &*opts.config.fs,
        path: &blob_dir,
        bytes: picked_bytes,
    };
    // Optional work: it takes only room the flush floor leaves. Checked and
    // taken into the merge's room in one step, so no other compaction sharing
    // the pool takes the room in between.
    let check = super::capacity::Check {
        quota_headroom,
        floor: super::capacity::Floor::Keep,
    };
    if !opts
        .outputs
        .try_reserve_more(&opts.capacity, opts.tree_id, &extra, check)
    {
        log::debug!("Skipping locality relocation: the output would eat into the reserve");
        return Ok(Vec::new());
    }

    log::debug!(
        "Relocating blob files for locality: {:?}",
        picked.iter().map(BlobFile::id).collect::<Vec<_>>(),
    );
    Ok(picked)
}

fn hidden_guard<T>(
    payload: &CompactionPayload,
    opts: &Options,
    f: impl FnOnce() -> crate::Result<T>,
) -> crate::Result<T> {
    f().inspect_err(|e| {
        log::error!("Compaction failed: {e:?}");

        // IMPORTANT: We need to show tables again on error
        let mut compaction_state = opts.compaction_state.lock();

        compaction_state
            .hidden_set_mut()
            .show(payload.table_ids.iter().copied());
    })
}

/// Decides whether the merge reduces to a merge-on-read relocation and, if so,
/// returns the positional delete-bitmap to materialize.
///
/// The conservative, provably-safe trigger: a SINGLE input segment that is
/// columnar, relocatable (non-encrypted, non-ECC, carries a zone map), under a
/// bitmap-writing strategy, whose own below-watermark range tombstones delete at
/// least one row. A single input means no foreign live keys can be lost and no
/// foreign point tombstone can be missed. The bitmap is built over the segment's
/// own scan (every version in block-index = position order) so it matches the
/// read mask. `None` falls through to the normal copy-on-write merge, including
/// the adaptive purge once the deleted fraction crosses the threshold.
#[cfg(feature = "std")]
fn plan_merge_on_read(
    opts: &Options,
    payload: &CompactionPayload,
    tables: &[Table],
    input_range_tombstones: &[crate::range_tombstone::RangeTombstone],
    pace: &crate::table::util::Pacer,
) -> crate::Result<Option<(Table, crate::table::delete_bitmap::DeleteBitmap)>> {
    let dst_lvl: usize = payload.canonical_level.into();
    let strategy = opts.runtime_config.load_full().delete_strategy.get(dst_lvl);
    if !strategy.writes_bitmap() {
        return Ok(None);
    }
    // Single relocatable columnar segment only; anything else is copy-on-write.
    let [source] = tables else {
        return Ok(None);
    };
    if !source.metadata.columnar
        || source.encryption.is_some()
        || source.metadata.ecc_params.is_some()
        || source.metadata.ecc_unrecognized
        || source.zone_map.is_empty()
        // A source that already carries a delete bitmap reads through the mask,
        // so `scan()` below yields the surviving rows under renumbered ordinals
        // that no longer line up with the verbatim-copied physical blocks. Relocating
        // it would alias the wrong rows; fall back to copy-on-write instead.
        || !source.delete_bitmap().is_empty()
        // A tight-space restricted view renumbers for the same reason — `scan()`
        // starts at its bound — and the relocation would additionally copy the
        // punched prefix and publish it unrestricted.
        || source.restrict_lower_bound().is_some()
    {
        return Ok(None);
    }
    // The relocation keeps the source's blocks as they were written: their
    // compression, row group and page sizes and column encoding, not the
    // destination level's policies. It exists to spare a sparse delete the
    // rewrite of every page, which applying those policies would be; the next
    // merge that copies these rows writes them under the destination's.

    // Positional bitmap from the segment's own below-watermark range tombstones,
    // over every stored version in block-index order (scan order = the writer's
    // position numbering = the read mask's expectation). Keep only the key and
    // seqno per row, dropping each scanned value so planning does not retain a
    // whole segment's values in memory.
    let keys = source
        .scan_paced(Some(pace))?
        .map(|entry| {
            let entry = entry?;
            Ok((entry.key.user_key, entry.key.seqno))
        })
        .collect::<crate::Result<Vec<_>>>()?;
    // The relocation keeps the source's key range and its range tombstones as
    // they are. A source written before the key-domain policy, or under another
    // one, may hold rows of two domains, or a tombstone reaching out of its
    // rows' domain: it is rewritten instead, so no output spans one.
    if let (Some(domains), Some((first, _)), Some((last, _))) =
        (opts.config.key_domains.as_ref(), keys.first(), keys.last())
    {
        use core::cmp::Ordering::{Greater, Less};

        let cmp = opts.config.comparator.as_ref();
        let end = domains.domain_end(first);
        let rows_leave = end
            .as_ref()
            .is_some_and(|end| cmp.compare(last, end) != Less);
        // A domain's end names it: a tombstone starting in another domain
        // answers another end.
        if rows_leave
            || input_range_tombstones.iter().any(|rt| {
                domains.domain_end(&rt.start) != end
                    || end
                        .as_ref()
                        .is_some_and(|end| cmp.compare(&rt.end, end) == Greater)
            })
        {
            return Ok(None);
        }
    }
    let bitmap = super::delete_materialize::build_position_bitmap(
        keys.iter()
            .map(|(user_key, seqno)| (user_key.as_ref(), *seqno)),
        input_range_tombstones,
        opts.gc_watermark,
        &opts.config.comparator,
    )?;
    if bitmap.is_empty() {
        return Ok(None);
    }

    // Adaptive: once the deleted fraction crosses the threshold, purge instead
    // (fall through to the copy-on-write merge, which drops the rows and clears
    // the bitmap).
    if let crate::config::DeleteStrategy::Adaptive {
        purge_threshold_percent,
    } = strategy
    {
        let item_count = source.metadata.item_count.max(1);
        // `bitmap.len() <= item_count`, both far below `u64::MAX / 100`.
        let deleted_percent = bitmap.len() * 100 / item_count;
        if deleted_percent >= u64::from(purge_threshold_percent) {
            return Ok(None);
        }
    }

    Ok(Some((source.clone(), bitmap)))
}

/// Relocates `source` into a new segment that reuses its data blocks verbatim
/// plus `bitmap`, then installs the version edit replacing `source`. The caller
/// has already released the version read lock (the install takes the write lock).
#[cfg(feature = "std")]
fn run_merge_on_read_relocation(
    compaction_state: &mut CompactionGuard<'_>,
    opts: &Options,
    payload: &CompactionPayload,
    source: &Table,
    bitmap: &crate::table::delete_bitmap::DeleteBitmap,
    pace: &crate::table::util::Pacer,
) -> crate::Result<CompactionResult> {
    let dst_lvl: usize = payload.canonical_level.into();
    let (folder, level_fs) = opts.config.tables_folder_for_level(payload.dest_level);

    // Write the relocated SST through the DESTINATION level's filesystem (the
    // same one that recovers and installs it below), not the source table's,
    // into a pooled file numbered above the source, as any compaction output.
    let crate::table::relocate::Relocated {
        id: new_id,
        path: new_path,
        entry_durable,
        checksum,
        copied,
    } = source.relocate_columnar_with_deletes(
        || {
            opts.table_files.claim_as(
                &folder,
                &level_fs,
                Some(source.id()),
                true,
                crate::fs::WriteOrigin::CompactionTable,
            )
        },
        &*level_fs,
        bitmap,
        opts.config.sync_mode,
        Some(pace),
    )?;
    // Recorded once finished: the relocation unlinks its own file on a
    // failure before that.
    opts.outputs
        .record_table(new_id, new_path.clone(), Arc::clone(&level_fs));
    opts.outputs.landed(&*level_fs, &new_path)?;

    let relocated = {
        let mut params = crate::table::RecoverParams::new(
            new_path,
            checksum,
            new_id,
            Arc::clone(&level_fs),
            opts.config.comparator.clone(),
            opts.config.cache.clone(),
        );
        // The copy keeps the source's place in L0: the manifest's recency,
        // which for an ingested source is not the one its file carries.
        params.recency = Some(source.l0_recency());
        params.tree_id = opts.tree_id;
        params
            .descriptor_table
            .clone_from(&opts.config.descriptor_table);
        params.pin_filter = opts.config.filter_block_pinning_policy.get(dst_lvl);
        params.pin_index = opts.config.index_block_pinning_policy.get(dst_lvl);
        params.encryption.clone_from(&opts.config.encryption);
        params
            .prefix_extractor
            .clone_from(&opts.config.prefix_extractor);
        #[cfg(zstd_any)]
        {
            params.zstd_dictionaries = opts.config.current_zstd_dictionaries();
        }
        #[cfg(feature = "metrics")]
        {
            params.metrics = opts.metrics.clone();
        }
        let table = Table::recover(params)?;
        if entry_durable {
            table.mark_entry_durable();
        }
        table
    };

    compaction_state
        .hidden_set_mut()
        .hide(payload.table_ids.iter().copied());

    let produced =
        super::flavour::ProducedOutput::for_relocation(relocated, source.clone(), copied);
    let result = super::flavour::install_merge(
        &mut opts.version_history.write(),
        opts,
        payload,
        vec![produced],
    );

    compaction_state
        .hidden_set_mut()
        .show(payload.table_ids.iter().copied());

    let Some(tables_out) = result? else {
        return Ok(CompactionResult::nothing());
    };
    Ok(CompactionResult {
        action: CompactionAction::Merged,
        dest_level: Some(payload.dest_level),
        tables_in: 1,
        tables_out,
        work: super::WorkReport::default(),
    })
}

#[expect(clippy::too_many_lines)]
fn merge_tables(
    mut compaction_state: CompactionGuard<'_>,
    version_history_lock: VersionsReadGuard<'_>,
    opts: &Options,
    payload: &CompactionPayload,
) -> crate::Result<CompactionResult> {
    if opts.stop_signal.is_stopped() {
        log::debug!("Stopping before compaction because of stop signal");
        return Ok(CompactionResult::nothing());
    }

    // Fail-safe for buggy compaction strategies
    if compaction_state
        .hidden_set()
        .should_decline_compaction(payload.table_ids.iter().copied())
    {
        log::warn!(
            "Compaction task created by {:?} contained hidden tables, declining to run it - please report this at https://github.com/fjall-rs/lsm-tree/issues/new?template=bug_report.md",
            opts.strategy.get_name(),
        );
        return Ok(CompactionResult::nothing());
    }

    // Arc so the snapshot can be shared with sub-compactions running on the
    // configured compaction pool (fire-and-forget executor needs `'static`).
    let current_super_version = Arc::new(version_history_lock.latest_version());

    let Some(tables) = payload
        .table_ids
        .iter()
        .map(|&id| current_super_version.version.get_table(id).cloned())
        .collect::<Option<Vec<_>>>()
    else {
        log::warn!(
            "Compaction task created by {:?} contained tables not referenced in the level manifest",
            opts.strategy.get_name(),
        );
        return Ok(CompactionResult::nothing());
    };

    let tables_in = payload.table_ids.len();

    // Collect range tombstones from input tables before they are moved.
    // Canonicalize to avoid duplicate RTs across input tables (MultiWriter
    // rotation copies the same RT into every output table during flush).
    let mut input_range_tombstones: Vec<crate::range_tombstone::RangeTombstone> = tables
        .iter()
        .flat_map(|t| t.range_tombstones().iter().cloned())
        .collect();
    input_range_tombstones.sort();
    input_range_tombstones.dedup();

    // Chosen once: the fast path and the split decision below and the
    // relocating writer must agree on the set.
    let blob_rewrite =
        pick_blob_files_for_merge(opts, &current_super_version.version, payload, &tables)?;

    let is_last_level = payload.dest_level == opts.config.level_count - 1;
    // Gathered once, for the estimate and for the serial merge below.
    let rt_plan = range_tombstone_plan(&current_super_version.version, is_last_level, &opts.config);
    opts.work.plan(merge_estimate(
        opts,
        &current_super_version.version,
        payload,
        &tables,
        &blob_rewrite.files,
        rt_plan.as_ref(),
    ));

    // Every read of the inputs, on the fast path as on the merge, reports its
    // latency to the limiter and counts to the run's work (see `input_pacer`).
    let pace = input_pacer(opts);

    // Merge-on-read fast path: a lone columnar segment whose own range
    // tombstones (below the watermark) delete some of its rows is relocated (its
    // data blocks reused verbatim plus a positional delete-bitmap) instead of
    // being re-transposed. Detection is read-only, so it runs under the held
    // read lock; on a hit the lock is released before the relocation installs
    // its own version edit. Multi-input merges and any non-relocatable segment
    // fall through to the normal copy-on-write merge below. So does a segment
    // whose blob files are relocated for locality: the verbatim copy would keep
    // its blob links, and with them the interleaving.
    #[cfg(feature = "std")]
    if !blob_rewrite.for_locality
        && let Some((source, bitmap)) =
            plan_merge_on_read(opts, payload, &tables, &input_range_tombstones, &pace)?
    {
        drop(current_super_version);
        drop(version_history_lock);
        return run_merge_on_read_relocation(
            &mut compaction_state,
            opts,
            payload,
            &source,
            &bitmap,
            &pace,
        );
    }
    let blob_files_to_rewrite = blob_rewrite.files;

    // ---- Parallel sub-compaction (std only) ----
    // A non-relocating compaction can be split into disjoint key ranges that
    // compact in parallel — each writes its own SSTs, then all merge into one
    // atomic version edit. Active blob relocation stays single-threaded and
    // falls through to the serial path below.
    #[cfg(feature = "std")]
    {
        let threads = opts.config.compaction_threads;
        let dst_lvl: usize = payload.canonical_level.into();

        // Only KV-separated trees with blob files to relocate get here with a
        // non-empty set; that path is not split.
        let relocating = !blob_files_to_rewrite.is_empty();

        let total_input_bytes: u64 = tables.iter().map(Table::file_size).sum();

        let boundaries = if threads > 1
            && !relocating
            && total_input_bytes >= opts.config.subcompaction_min_bytes
        {
            subcompaction_boundaries(
                &current_super_version.version,
                payload.dest_level as usize,
                threads,
                &opts.config.comparator,
            )
        } else {
            Vec::new()
        };

        if !boundaries.is_empty() {
            let blobs_folder = opts.config.path.join(BLOBS_FOLDER);
            let ranges = ranges_from_boundaries(&boundaries);

            // Hand off: hide inputs, release the exclusive + version locks for
            // the CPU-heavy parallel phase.
            drop(version_history_lock);
            compaction_state
                .hidden_set_mut()
                .hide(payload.table_ids.iter().copied());
            drop(compaction_state);

            // Run the sub-compactions on the configured compaction pool (NOT raw
            // threads), so a caller-shared pool bounds total threads across
            // trees. The executor is fire-and-forget + `'static`, so each task
            // owns an Arc'd snapshot of the context; results come back over an
            // mpsc channel, indexed by range. Only range 0 carries the input
            // tables to delete, so they are dropped exactly once at install. No
            // pool configured (e.g. caller injected none) → run sequentially.
            let num_ranges = ranges.len();
            let only_first_owns_inputs =
                |idx: usize| if idx == 0 { tables.clone() } else { Vec::new() };
            // One runtime snapshot for the whole compaction, taken before it
            // splits: every range writes under the same configuration.
            let rc = opts.runtime_config.load_full();
            // One filter plan too, so the ranges share one filter budget, and
            // price by what every range has still to write.
            let filter_sizing = super::flavour::plan_filters(
                &current_super_version.version,
                opts,
                payload,
                &boundaries,
                None,
                &rc,
            );
            // The ranges read one version: the keys the tables outside the
            // compaction hold are found once for all of them.
            let outside_spans = (!input_range_tombstones.is_empty()).then(|| {
                Arc::new(OutsideSpans::new(
                    &current_super_version.version,
                    &payload.table_ids,
                    opts.config.comparator.as_ref(),
                ))
            });

            let outputs: Vec<crate::Result<super::flavour::ProducedOutput>> =
                if let Some(spawner) = opts.config.compaction_pool.clone() {
                    let opts = Arc::new(opts.clone());
                    let payload = Arc::new(payload.clone());
                    let version = Arc::clone(&current_super_version);
                    let rts = Arc::new(input_range_tombstones.clone());
                    let blobs = Arc::new(blobs_folder);

                    let (tx, rx) = std::sync::mpsc::channel();
                    for (idx, range) in ranges.iter().cloned().enumerate() {
                        let tx = tx.clone();
                        let opts = Arc::clone(&opts);
                        let payload = Arc::clone(&payload);
                        let version = Arc::clone(&version);
                        let rts = Arc::clone(&rts);
                        let outside_spans = outside_spans.clone();
                        let blobs = Arc::clone(&blobs);
                        let rc = Arc::clone(&rc);
                        let filter_sizing = filter_sizing.clone();
                        let tables_for_deletion = only_first_owns_inputs(idx);
                        spawner.spawn(Box::new(move || {
                            let out = run_subcompaction(
                                &opts,
                                &payload,
                                &version.version,
                                tables_for_deletion,
                                &rts,
                                outside_spans.as_deref(),
                                range,
                                dst_lvl,
                                is_last_level,
                                true,
                                &blobs,
                                // The parallel path never relocates blobs; tight
                                // blob defrag is the serial slice loop's domain.
                                None,
                                &rc,
                                filter_sizing,
                            );
                            // The receiver outlives every send (it drains N
                            // items below), so this cannot fail.
                            let _ = tx.send((idx, out));
                        }));
                    }
                    drop(tx);

                    let mut slots: Vec<Option<crate::Result<super::flavour::ProducedOutput>>> =
                        (0..num_ranges).map(|_| None).collect();
                    for (idx, out) in rx {
                        if let Some(slot) = slots.get_mut(idx) {
                            *slot = Some(out);
                        }
                    }
                    slots
                        .into_iter()
                        .map(|slot| {
                            slot.unwrap_or_else(|| {
                                // A worker panicked before sending: treat as a
                                // failed sub-compaction so install is skipped.
                                Err(crate::Error::from(crate::io::Error::other(
                                    "sub-compaction worker did not report a result",
                                )))
                            })
                        })
                        .collect()
                } else {
                    ranges
                        .iter()
                        .cloned()
                        .enumerate()
                        .map(|(idx, range)| {
                            run_subcompaction(
                                opts,
                                payload,
                                &current_super_version.version,
                                only_first_owns_inputs(idx),
                                &input_range_tombstones,
                                outside_spans.as_deref(),
                                range,
                                dst_lvl,
                                is_last_level,
                                true,
                                &blobs_folder,
                                None,
                                &rc,
                                filter_sizing.clone(),
                            )
                        })
                        .collect()
                };

            // One failed range aborts the install. The files its siblings
            // finished are not named by any version, so the run removes them
            // with everything else it never installed.
            let outputs = match outputs.into_iter().collect::<crate::Result<Vec<_>>>() {
                Ok(outputs) => outputs,
                Err(err) => {
                    log::error!("Sub-compaction failed: {err:?}");
                    opts.compaction_state
                        .lock()
                        .hidden_set_mut()
                        .show(payload.table_ids.iter().copied());
                    return Err(err);
                }
            };

            #[cfg(test)]
            opts.config.fire_before_output_install();

            // Re-acquire locks and install one atomic version edit for all outputs.
            let mut compaction_state = opts.compaction_state.lock();
            let mut version_history_lock = opts.version_history.write();

            let tables_out =
                super::flavour::install_merge(&mut version_history_lock, opts, payload, outputs)
                    .inspect_err(|e| {
                        // The run removes the outputs it failed to install; the
                        // inputs only need to be shown again.
                        log::error!("Sub-compaction install failed: {e:?}");
                        compaction_state
                            .hidden_set_mut()
                            .show(payload.table_ids.iter().copied());
                    })?;
            // Held until the install: the installed outputs name their
            // dictionaries from here on (see `WritePin`).
            drop(rc);

            compaction_state
                .hidden_set_mut()
                .show(payload.table_ids.iter().copied());

            drop(version_history_lock);
            drop(compaction_state);

            let Some(tables_out) = tables_out else {
                return Ok(CompactionResult::nothing());
            };
            log::trace!("Parallel compaction done in {num_ranges} sub-ranges");

            return Ok(CompactionResult {
                action: CompactionAction::Merged,
                dest_level: Some(payload.dest_level),
                tables_in,
                tables_out,
                work: super::WorkReport::default(),
            });
        }
    }

    // The blob accounting of the dropped versions, shared by the stream that
    // reports them and the loop that settles a key's ownership before its
    // kept rows are written (see `ownership`).
    let ledger = core::cell::RefCell::new(super::ownership::OwnershipLedger::default());
    let ledger_error = core::cell::RefCell::new(None);
    let mut ledger_hook = super::ownership::LedgerHook {
        ledger: &ledger,
        error: &ledger_error,
    };

    // Declared before the stream that borrows it, so it outlives it.
    let completeness = input_completeness_for(
        &current_super_version.version,
        &payload.table_ids,
        &opts.config,
    );

    // One runtime snapshot for everything this compaction writes, its tables
    // and its filter's blob files, held until its output is installed.
    let rc = opts.runtime_config.load_full();

    // Row groups and data blocks the merge leaves as they were are copied into
    // the output instead of being encoded again (see `carry`). A compaction
    // relocating blob files writes every row itself, so its scans record
    // nothing.
    let carry = if blob_files_to_rewrite.is_empty() {
        super::carry::plan(&opts.config, &rc, payload.canonical_level.into(), &tables)
    } else {
        None
    };

    let Some(mut merge_iter) = create_compaction_stream(
        &current_super_version.version,
        &payload.table_ids.iter().copied().collect::<Vec<_>>(),
        opts.gc_watermark,
        opts.config.merge_operator.clone(),
        opts.config.comparator.clone(),
        Some(&pace),
        carry.as_ref(),
    )?
    else {
        log::warn!(
            "Compaction task tried to compact tables that do not exist, declining to run it"
        );
        return Ok(CompactionResult::nothing());
    };

    let dst_lvl = payload.canonical_level.into();

    // See `holds_every_surviving_version`: the destination level alone does not
    // prove that a key missing from the inputs is missing from the tree.
    let holds_everything = holds_every_surviving_version(
        &current_super_version.version,
        &payload.table_ids,
        is_last_level,
        opts.config.comparator.as_ref(),
    );

    merge_iter = merge_iter
        .evict_tombstones(holds_everything)
        .zero_seqnos(false);

    // Whole-version tombstones for compaction-time RT application (drop covered
    // KVs in the merge) and the bottommost seqno-zeroing gate, gathered above
    // from every level so coverage outside this compaction is respected. The
    // scan and the seqno-zeroer wrapper are `core` + `alloc`, so the bottommost
    // RT-application + zeroing runs on the `no_std` serial path too (see the
    // zeroer wrap below).
    if let Some((tombstones, rt_use)) = &rt_plan {
        merge_iter = merge_iter.with_range_tombstones(
            tombstones.clone(),
            opts.config.comparator.clone(),
            *rt_use,
        );
    }
    if let Some(predicate) = &completeness {
        merge_iter = merge_iter.with_input_completeness(predicate);
    }
    let zeroing_tombstones = zeroing_tombstones_from(rt_plan.as_ref());

    let blobs_folder = opts.config.path.join(BLOBS_FOLDER);

    let filter_ctx = Context { is_last_level };

    // Construct the compaction filter
    let mut compaction_filter = opts.config.compaction_filter_factory.as_ref().map(|f| {
        log::trace!("Installing custom compaction filter {:?}", f.name());
        f.make_filter(&filter_ctx)
    });
    // Shared transform counter, created for EVERY compaction: the filter
    // adapter ticks it on non-Keep verdicts, the merge stream ticks it on
    // its own visibility-changing drops (bottommost tombstone elision,
    // range-tombstone application, weak-tombstone annihilation), and the
    // table writer marks transformed the lineage of any output whose window
    // saw one (see `prepare_table_writer`).
    let transform_marker = Some(Arc::new(portable_atomic::AtomicU64::new(0)));
    if let Some(marker) = &transform_marker {
        merge_iter = merge_iter.with_transform_marker(Arc::clone(marker));
    }

    // Versions that go in and do not come out: read after `produce` to tell a
    // run that collected history from one that collected none.
    let gc_balance = merge_iter.gc_balance();

    // Where the compaction filter writes the new values it separates, when
    // this pass has no blob output of its own; a relocating pass separates
    // them into its output, beside the frames it relocates.
    let mut filter_blob_writer = None;
    let relocating = opts.config.kv_separation_opts.is_some() && !blob_files_to_rewrite.is_empty();
    // Filter-only transform counter (see `TransformCounters::filter`): read
    // after `produce` to decide the install's retention effect.
    let filter_marker = Arc::new(portable_atomic::AtomicU64::new(0));
    let mut merge_iter = merge_iter.with_filter(StreamFilterAdapter::new(
        compaction_filter.as_deref_mut(),
        opts,
        &current_super_version.version,
        FilterBlobOutput {
            folder: &blobs_folder,
            writer: (!relocating).then_some(&mut filter_blob_writer),
            runtime: &rc,
        },
        &filter_ctx,
        TransformCounters {
            transform: transform_marker.clone(),
            filter: Arc::clone(&filter_marker),
        },
    ));

    // Serial (single-stream) compaction: block compression may use the pool,
    // and the single writer owns the whole run (its final output carries the
    // `lineage_last` marker).
    let table_writer = super::flavour::prepare_table_writer(
        &current_super_version.version,
        opts,
        payload,
        super::flavour::plan_filters(
            &current_super_version.version,
            opts,
            payload,
            &[],
            None,
            &rc,
        ),
        true,
        transform_marker,
        true,
        &rc,
    )?;

    let start = Instant::now();

    let mut compactor = match &opts.config.kv_separation_opts {
        Some(blob_opts) => {
            merge_iter = merge_iter.with_drop_callback(&mut ledger_hook);

            if blob_files_to_rewrite.is_empty() {
                log::debug!("No blob relocation needed");

                Box::new(StandardCompaction::new(table_writer, tables))
                    as Box<dyn super::flavour::CompactionFlavour>
            } else {
                log::debug!(
                    "Relocate blob files: {:?}",
                    blob_files_to_rewrite
                        .iter()
                        .map(BlobFile::id)
                        .collect::<Vec<_>>(),
                );
                ledger
                    .borrow_mut()
                    .relocating(blob_files_to_rewrite.iter().map(BlobFile::id));

                let scanner = BlobFileMergeScanner::new(
                    blob_files_to_rewrite
                        .iter()
                        .map(|bf| open_blob_scanner_at_frontier(bf, 0, opts))
                        .collect::<crate::Result<Vec<_>>>()?,
                    opts.config.comparator.clone(),
                );

                let writer = BlobFileWriter::new_for(
                    opts.blob_file_id_generator.clone(),
                    &blobs_folder,
                    opts.tree_id,
                    opts.config.descriptor_table.clone(),
                    opts.config.fs.clone(),
                    crate::fs::WriteOrigin::CompactionBlob,
                )?
                .use_output_ledger(opts.outputs.clone())
                .use_target_size(blob_opts.file_target_size)
                .use_passthrough_compression(rc.blob_compression)
                .use_sync_mode(opts.config.sync_mode)
                .use_writeback_bytes(opts.config.writeback_bytes)
                .use_lifetime_groups(blob_opts.lifetime_groups)
                .use_clock(opts.config.clock.clone())
                .use_key_domains(
                    opts.config.key_domains.clone(),
                    opts.config.comparator.clone(),
                );
                // Same as the tight-space relocation above, through the same
                // `RelocatingCompaction`: the policy is only the opening value,
                // each output file records the codec of the source its frames
                // came from, and the set pins the dictionary that names.
                let writer = with_fresh_encoding(writer, opts, &rc);

                let inner = StandardCompaction::new(table_writer, tables);

                Box::new(
                    RelocatingCompaction::new(
                        inner,
                        scanner.peekable(),
                        writer,
                        blob_files_to_rewrite,
                        opts.rate_limiter.clone(),
                        opts.stop_signal.clone(),
                        opts.config.comparator.clone(),
                        opts.tree_id,
                    )
                    .with_fresh_values(FreshValues {
                        options: blob_opts.clone(),
                        compression: rc.blob_compression,
                    }),
                )
            }
        }
        None => Box::new(StandardCompaction::new(table_writer, tables)),
    };

    log::trace!("Blob file GC preparation done in {:?}", start.elapsed());

    drop(version_history_lock);

    {
        compaction_state
            .hidden_set_mut()
            .hide(payload.table_ids.iter().copied());
    }

    // IMPORTANT: Unlock exclusive compaction lock as we are now doing the actual (CPU-intensive) compaction
    drop(compaction_state);

    // Propagate range tombstones to output tables BEFORE writing KV items,
    // so that if the compactor rotates tables during the merge loop,
    // earlier tables already carry the RT metadata.
    //
    // NOTE: this path does NOT GC fully-applied tombstones (unlike the
    // parallel sub-compaction path). The serial stop-signal handling below
    // commits whatever was written so far (`return Ok(())`), so a stop
    // landing after this write but before the covered tail is processed
    // could drop a below-watermark tombstone while some covered keys were
    // never visited — resurrecting them. Tombstone GC therefore only runs in
    // the sub-compaction path, which is atomic (it returns an error and
    // rolls back on stop). Covered keys are still physically dropped here by
    // the merge stream; keeping the tombstone is the conservative, correct
    // choice when the compaction may commit partial output.
    if !input_range_tombstones.is_empty() {
        log::debug!(
            "Propagating {} range tombstones to compaction output",
            input_range_tombstones.len(),
        );
        let outside = tombstones_outside(
            &input_range_tombstones,
            &OutsideSpans::new(
                &current_super_version.version,
                &payload.table_ids,
                opts.config.comparator.as_ref(),
            ),
            (None, None),
            &opts.config.comparator,
        );
        compactor.write_range_tombstones(&input_range_tombstones, outside);
    }

    // Every row the merge emits goes to the output through here: written, or,
    // when a run of them is an input's row group or data block as it was
    // read, copied as it lies.
    let mut matcher = compactor.carrying().and(carry.as_ref()).map(|carry| {
        super::carry::CarryMatcher::new(Arc::clone(&carry.queue), opts.config.comparator.clone())
    });
    let read_pace: Option<&dyn crate::table::util::ReadPacer> = Some(&*pace);
    // Chosen once, so a compaction that copies nothing pays no check per row.
    let mut carrying_emit;
    let mut plain_emit;
    let emit: &mut dyn FnMut(InternalValue) -> crate::Result<()> = if let Some(matcher) =
        matcher.as_mut()
        && let Some(flavour) = compactor.carrying()
    {
        carrying_emit = move |row: InternalValue| {
            matcher.write(
                row,
                &mut FlavourSink {
                    flavour: &mut *flavour,
                    pace: read_pace,
                },
            )
        };
        &mut carrying_emit
    } else {
        plain_emit = |row: InternalValue| compactor.write(row);
        &mut plain_emit
    };

    hidden_guard(payload, opts, || {
        // Bottommost seqno-zeroing: at the last level, entries below the GC
        // watermark and not covered by any range tombstone get seqno 0. The
        // zeroer and the whole-version tombstone scan are `core` + `alloc`, so
        // this runs on the `no_std` serial path too.
        let merge_iter = super::seqno_zeroer::BottommostSeqnoZeroer::new(
            merge_iter,
            holds_everything,
            zeroing_tombstones,
            opts.gc_watermark,
            opts.config.comparator.clone(),
            Arc::clone(&gc_balance),
        );

        for (idx, item) in merge_iter.enumerate() {
            let item = item?;

            // Pace compaction I/O so it cannot saturate the device and
            // starve user reads. Short-circuits to a single relaxed atomic
            // load when `compaction_rate_limit` is 0 (the default), so the
            // unthrottled hot path stays cheap. The wait is interruptible by
            // the stop signal so a low limit plus a large item can't stall
            // tree drop / shutdown for the whole wait; a stopped item is not
            // written, so its debit goes back to the shared budget.
            //
            // Accounting here covers the SST entry's key + value bytes
            // (for KV-separated entries `item.value` is the encoded handle).
            // The relocated blob payload of KV-separated compactions is
            // debited separately at its write site in
            // `RelocatingCompaction::write`, where the real moved bytes are
            // known.
            // One key + value length; the sum is far below u64::MAX → plain add.
            let io_bytes = item.key.user_key.len() as u64 + item.value.len() as u64;
            if opts
                .rate_limiter
                .request_abortable(io_bytes, || opts.stop_signal.is_stopped())
            {
                // Same reason as the stop check below: the item was not even
                // written, so committing here would drop it and everything
                // after it.
                log::debug!("Stopping amidst compaction because of stop signal (I/O throttle)");
                return Err(cancelled_compaction());
            }

            ledger.borrow_mut().admit(item, &mut *emit)?;
            if let Some(e) = ledger_error.borrow_mut().take() {
                return Err(e);
            }

            // Test-only failpoint: raise the stop signal after the first written
            // item, so the interrupted-merge path below is reachable without
            // racing a tree drop against an in-flight compaction.
            #[cfg(all(test, feature = "std"))]
            if opts
                .config
                .stop_serial_merge_after_first_item
                .swap(false, core::sync::atomic::Ordering::SeqCst)
            {
                opts.stop_signal.send();
            }

            // Abort, do not produce. The install swaps EVERY input table named
            // in the payload for this output, so committing a merge that never
            // read its input to the end drops the unread tail out of the tree
            // entirely. The parallel path refuses for the same reason; this one
            // used to commit and call the result `Merged`.
            if idx % 1_000_000 == 0 && opts.stop_signal.is_stopped() {
                log::debug!("Stopping amidst compaction because of stop signal");
                return Err(cancelled_compaction());
            }
        }

        Ok(())
    })?;

    // The rows the ledger still holds go out before the outputs are finished,
    // guarded like the merge: a failure here shows the inputs again.
    let (blob_frag_map, released_objects) = hidden_guard(payload, opts, || {
        let settled = ledger.into_inner().finish(&mut *emit)?;
        if let Some(e) = ledger_error.into_inner() {
            return Err(e);
        }
        Ok(settled)
    })?;

    // What the matcher still holds goes out after the last row, guarded alike.
    let carried = match (matcher.take(), compactor.carrying()) {
        (Some(matcher), Some(flavour)) => hidden_guard(payload, opts, || {
            matcher.finish(&mut FlavourSink {
                flavour,
                pace: read_pace,
            })
        })?,
        _ => super::carry::Carried::default(),
    };
    log::debug!(
        "Carried {} row groups and {} data blocks ({} rows, {} bytes) into the output whole",
        carried.groups,
        carried.blocks,
        carried.rows,
        carried.bytes,
    );

    if let Some(filter) = compaction_filter {
        filter.finish();
    }

    #[cfg(test)]
    opts.config.fire_before_output_install();

    let mut compaction_state = opts.compaction_state.lock();

    log::trace!("Acquiring super version write lock");
    let mut version_history_lock = opts.version_history.write();
    log::trace!("Acquired super version write lock");

    log::trace!("Blob fragmentation diff: {blob_frag_map:#?}");

    let extra_blob_files = filter_blob_writer
        .map(BlobFileWriter::finish)
        .transpose()
        .inspect_err(|e| {
            // NOTE: We cannot use hidden_guard here because we already locked the compaction state

            log::error!("Compaction failed while finishing filter blob writer: {e:?}");

            compaction_state
                .hidden_set_mut()
                .show(payload.table_ids.iter().copied());
        })?
        .unwrap_or_default();

    // Phase split: `produce` finalizes this compaction's output files (no
    // version touch); `install_merge` commits one atomic version edit. With a
    // single output the result is identical to the old combined `finish`; the
    // split is what lets parallel sub-compactions each produce independently
    // and then share one install.
    let mut produce_output = compactor
        .produce(opts, dst_lvl, blob_frag_map, extra_blob_files)
        .inspect_err(|e| {
            // NOTE: We cannot use hidden_guard here because we already locked the compaction state

            log::error!("Compaction failed: {e:?}");

            compaction_state
                .hidden_set_mut()
                .show(payload.table_ids.iter().copied());
        })?;
    produce_output.set_released_objects(released_objects);
    produce_output.set_carried(carried);
    if filter_marker.load(core::sync::atomic::Ordering::Relaxed) > 0 {
        produce_output.mark_filter_transformed();
    }
    if gc_balance.load(core::sync::atomic::Ordering::Relaxed) > 0 {
        produce_output.mark_collected_below_watermark();
    }

    let tables_out = super::flavour::install_merge(
        &mut version_history_lock,
        opts,
        payload,
        vec![produce_output],
    )
    .inspect_err(|e| {
        // NOTE: We cannot use hidden_guard here because we already locked the compaction state

        log::error!("Compaction failed: {e:?}");

        compaction_state
            .hidden_set_mut()
            .show(payload.table_ids.iter().copied());
    })?;
    // Held until the install: the installed outputs name their dictionaries
    // from here on (see `WritePin`).
    drop(rc);

    compaction_state
        .hidden_set_mut()
        .show(payload.table_ids.iter().copied());

    drop(version_history_lock);
    drop(compaction_state);

    let Some(tables_out) = tables_out else {
        return Ok(CompactionResult::nothing());
    };
    log::trace!("Compaction successful");

    Ok(CompactionResult {
        action: CompactionAction::Merged,
        dest_level: Some(payload.dest_level),
        tables_in,
        tables_out,
        work: super::WorkReport::default(),
    })
}

/// A compaction's output as the carry matcher sees it.
struct FlavourSink<'a> {
    flavour: &'a mut StandardCompaction,
    /// Told how long reading a carried group takes.
    pace: Option<&'a dyn crate::table::util::ReadPacer>,
}

impl super::carry::CarrySink for FlavourSink<'_> {
    fn write(&mut self, row: InternalValue) -> crate::Result<()> {
        super::flavour::CompactionFlavour::write(self.flavour, row)
    }

    fn carry(
        &mut self,
        candidate: &crate::table::group_carry::CarryCandidate,
    ) -> crate::Result<bool> {
        self.flavour.carry(candidate, self.pace)
    }

    fn carry_pages(
        &mut self,
        candidate: &crate::table::group_carry::CarryCandidate,
        emitted: &[InternalValue],
    ) -> crate::Result<Option<u64>> {
        self.flavour.carry_pages(candidate, emitted, self.pace)
    }

    fn transforms_seen(&self) -> u64 {
        self.flavour.transforms_seen()
    }

    #[inline]
    fn continues_last(&self, key: &[u8]) -> bool {
        self.flavour.continues_last(key)
    }

    fn settle_transforms(&mut self, seen: u64) {
        self.flavour.settle_transforms(seen);
    }
}

/// Unlinks `ids_to_drop` in one version edit whose `effect` says what the drop
/// owes older snapshots.
fn drop_tables(
    compaction_state: &CompactionGuard<'_>,
    opts: &Options,
    ids_to_drop: &[TableId],
    effect: crate::version::RetentionEffect,
) -> crate::Result<CompactionResult> {
    // Nothing to drop: no version edit either. Installing an empty drop would
    // cost a manifest append AND raise the persisted retention floor to an
    // install that removed nothing, refusing every older snapshot after a
    // reopen for no reason. A strategy that picks no table is a no-op, not a
    // drop.
    if ids_to_drop.is_empty() {
        return Ok(CompactionResult::nothing());
    }

    let mut version_history_lock = opts.version_history.write();

    // Fail-safe for buggy compaction strategies
    if compaction_state
        .hidden_set()
        .should_decline_compaction(ids_to_drop.iter().copied())
    {
        log::warn!(
            "Compaction task created by {:?} contained hidden tables, declining to run it - please report this at https://github.com/fjall-rs/lsm-tree/issues/new?template=bug_report.md",
            opts.strategy.get_name(),
        );
        return Ok(CompactionResult::nothing());
    }

    let Some(tables) = ids_to_drop
        .iter()
        .map(|&id| {
            version_history_lock
                .latest_version()
                .version
                .get_table(id)
                .cloned()
        })
        .collect::<Option<Vec<_>>>()
    else {
        log::warn!(
            "Compaction task created by {:?} contained tables not referenced in the level manifest",
            opts.strategy.get_name(),
        );
        return Ok(CompactionResult::nothing());
    };

    log::debug!("Dropping tables: {ids_to_drop:?}");

    let mut dropped_blob_files = vec![];

    // The objects the dropped tables owned, read from their owned-objects
    // sections before the edit: nothing that can fail may run once the drop
    // is published, or an error would report a drop that happened and leave
    // its objects unrecorded. Read only while some read holds references:
    // with none, no reference can be stale through this drop.
    let mut released_tables = Vec::new();
    if version_history_lock.released().holds_reads() {
        for table in &tables {
            let owned = table.owned_blob_objects()?;
            if !owned.is_empty() {
                released_tables.push(crate::blob_tree::released::OwnedObjects::from(owned));
            }
        }
    }

    // IMPORTANT: Write the manifest with the removed tables first
    // Otherwise the table files are deleted, but are still referenced!
    version_history_lock.upgrade_version(
        &opts.config.path,
        |current| {
            let mut copy = current.clone();

            let ctx = crate::version::TransformContext::new(opts.config.comparator.as_ref());
            copy.version = copy.version.with_dropped(
                ids_to_drop,
                &mut dropped_blob_files,
                &|id| current.memtables_reference_blob_file(id),
                &ctx,
            )?;

            Ok(copy)
        },
        &opts.global_seqno,
        &opts.visible_seqno,
        &*opts.config.fs,
        opts.runtime_config.load_full(),
        opts.encryption.clone(),
        effect,
    )?;

    // Still under the write lock that published the drop: a reference read
    // before it to an object a dropped table owned is stale.
    if !released_tables.is_empty() {
        let published = version_history_lock.latest_version().version.id();
        version_history_lock
            .released()
            .record(published, [], released_tables);
    }

    drop(version_history_lock);

    // NOTE: If the application were to crash >here< it's fine
    // The tables are not referenced anymore, and will be
    // cleaned up upon recovery
    for table in tables {
        table.mark_as_deleted();
    }

    for blob_file in dropped_blob_files {
        blob_file.mark_as_deleted();
    }

    let tables_dropped = ids_to_drop.len();

    log::trace!("Dropped {tables_dropped} tables");

    Ok(CompactionResult {
        action: CompactionAction::Dropped,
        dest_level: None,
        tables_in: tables_dropped,
        tables_out: 0,
        // A drop reads and writes nothing, and moves nothing.
        work: super::WorkReport::default(),
    })
}

#[cfg(test)]
mod tests;

#[cfg(test)]
#[expect(clippy::unwrap_used, reason = "test code")]
mod alignment_tests;

#[cfg(test)]
#[expect(clippy::expect_used, reason = "test code")]
mod move_tests;