mito2/compaction/

twcs.rs

// Copyright 2023 Greptime Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

mod units;

use std::collections::hash_map::Entry;
use std::collections::{BTreeMap, HashMap, HashSet};
use std::fmt::Debug;
use std::sync::{Arc, LazyLock};

use common_base::readable_size::ReadableSize;
use common_telemetry::{debug, info};
use common_time::Timestamp;
use common_time::range::TimestampRange;
use common_time::timestamp::TimeUnit;
use common_time::timestamp_millis::BucketAligned;
use prometheus::{HistogramVec, IntCounterVec};
use snafu::ResultExt;
use store_api::storage::RegionId;

use crate::compaction::CompactionOutput;
use crate::compaction::buckets::infer_time_bucket;
use crate::compaction::compactor::CompactionRegion;
use crate::compaction::picker::{Picker, PickerOutput, get_expired_ssts};
use crate::compaction::run::{
    Ranged, SortedRun, files_overlap, files_overlap_inclusive, find_sorted_runs,
    find_sorted_runs_by_time_range, merge_primary_key_ranges, primary_key_ranges_overlap,
};
use crate::compaction::twcs::units::{ClosedMergeOptions, PickLimits};
use crate::error::{JoinSnafu, Result};
use crate::metrics::{TWCS_PICK_ELAPSED, TWCS_SCANNED_WINDOWS};
use crate::region::options::MergeMode;
use crate::sst::file::{FileHandle, Level, overlaps};
use crate::sst::primary_key::PrimaryKeyRangeMapper;
use crate::sst::version::LevelMeta;

const LEVEL_COMPACTED: Level = 1;

/// Scheduling priority of a window's pick. Earlier classes are scheduled first,
/// and newer windows first within a class.
#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub(crate) enum PickClass {
    /// Merges L0 files, which also absorbs late data of inactive windows.
    L0Merge,
    /// Merges files to reduce the window's file count.
    FileReduction,
    /// Merges files of an inactive window to resolve their overlaps.
    OverlapResolution,
}

impl PickClass {
    pub(crate) fn as_str(self) -> &'static str {
        match self {
            Self::L0Merge => "l0_merge",
            Self::FileReduction => "file_reduction",
            Self::OverlapResolution => "overlap_resolution",
        }
    }
}

/// Compaction inputs picked for one window.
#[derive(Debug)]
pub(crate) struct WindowPick {
    pub(crate) class: PickClass,
    /// Name of the strategy that picked the inputs, for logs.
    branch: &'static str,
    pub(crate) inputs: Vec<FileHandle>,
    /// Number of sorted runs or units the strategy considered, for logs.
    found_runs: usize,
}

/// A discovered alternative, not yet selected or admitted by the scheduler.
#[derive(Debug)]
pub(crate) struct WindowCandidate {
    pub(crate) window: i64,
    pub(crate) pick: WindowPick,
}

/// Discovery snapshot retained until the scheduler chooses its alternatives.
/// Only selected candidates incur tombstone checks and pick-result logging.
pub(crate) struct TwcsCandidates {
    pub(crate) candidates: Vec<WindowCandidate>,
    expired_ssts: Vec<FileHandle>,
    time_window_size: i64,
    windows: Arc<BTreeMap<i64, Window>>,
    active_window: Option<i64>,
    primary_key_mapper: Arc<PrimaryKeyRangeMapper>,
    region_id: RegionId,
    append_mode: bool,
    max_output_file_size: Option<u64>,
}

impl TwcsCandidates {
    /// Materializes the selected candidates off the async runtime. The snapshot
    /// supplies unselected peers for tombstone safety, even after truncation.
    pub(crate) async fn finish(self) -> Result<PickerOutput> {
        common_runtime::spawn_blocking_compact(move || {
            let outputs = self
                .candidates
                .into_iter()
                .rev()
                .map(|candidate| {
                    let WindowCandidate { window, pick } = candidate;
                    let files = &self.windows[&window];
                    let filter_deleted = !self.append_mode
                        && !window_has_overlap(files, &self.windows)
                        && !selected_overlaps_unselected(
                            &pick.inputs,
                            files,
                            &self.primary_key_mapper,
                        );
                    if pick.inputs.len() > 1 {
                        log_pick_result(
                            self.region_id,
                            window,
                            self.active_window,
                            &pick,
                            files.files.len(),
                            self.max_output_file_size,
                            filter_deleted,
                        );
                    }
                    CompactionOutput {
                        output_level: LEVEL_COMPACTED,
                        inputs: pick.inputs,
                        filter_deleted,
                        output_time_range: None,
                    }
                })
                .collect();
            PickerOutput {
                outputs,
                expired_ssts: self.expired_ssts,
                time_window_size: self.time_window_size,
                // The picker treats zero as unlimited; the writer must not split every batch.
                max_file_size: self
                    .max_output_file_size
                    .filter(|size| *size > 0)
                    .map(|size| size as usize),
            }
        })
        .await
        .context(JoinSnafu)
    }
}

/// Snapshot-wide inputs shared by each independent window strategy.
struct WindowPickContext<'a> {
    all_candidates: bool,
    region_id: RegionId,
    active_window: Option<i64>,
    time_window_size: Option<i64>,
    region_sequence: u64,
    mapper: &'a PrimaryKeyRangeMapper,
}

struct WindowOutputContext {
    all_candidates: bool,
    expired_ssts: Vec<FileHandle>,
    active_window: Option<i64>,
    time_window_size: Option<i64>,
    max_outputs: Option<usize>,
    primary_key_mapper: Arc<PrimaryKeyRangeMapper>,
}

/// A mixed L0/L1 compaction may rewrite at most this many L1 rows per L0 row.
const MAX_L1_L0_ROW_RATIO: usize = 2;

/// Default maximum number of input SST files in one compaction input.
const DEFAULT_MAX_INPUT_FILES: usize = 16;

const MAX_INPUT_FILES_ENV: &str = "GREPTIME_TWCS_MAX_INPUT_FILES";

/// Maximum number of input SST files in one compaction input.
/// Configurable via [`MAX_INPUT_FILES_ENV`].
static MAX_INPUT_FILES: LazyLock<usize> = LazyLock::new(|| {
    let env_value = std::env::var(MAX_INPUT_FILES_ENV).ok();
    parse_max_input_files(env_value.as_deref())
});

/// A closed merge in an inactive window may take this many times
/// [`MAX_INPUT_FILES`]. Resolving overlaps across a whole window needs all of
/// its overlapping files at once, and each byte is rewritten once per window.
const CLOSED_MERGE_INPUT_FACTOR: usize = 4;

fn parse_max_input_files(env_value: Option<&str>) -> usize {
    env_value
        .and_then(|env_value| env_value.parse().ok())
        .filter(|max_input_files| *max_input_files >= 2)
        .unwrap_or(DEFAULT_MAX_INPUT_FILES)
}

/// Cloneable planning metric handles, shared with spawned window evaluations.
#[derive(Clone, Debug)]
pub(crate) struct TwcsPickMetrics {
    elapsed: HistogramVec,
    scanned_windows: IntCounterVec,
}

impl Default for TwcsPickMetrics {
    fn default() -> Self {
        Self {
            elapsed: TWCS_PICK_ELAPSED.clone(),
            scanned_windows: TWCS_SCANNED_WINDOWS.clone(),
        }
    }
}

/// `TwcsPicker` picks files of which the max timestamp are in the same time window as compaction
/// candidates.
#[derive(Clone, Debug)]
pub struct TwcsPicker {
    /// Production uses registered handles; tests can use unregistered instances.
    pub(crate) metrics: TwcsPickMetrics,
    /// Minimum file num to trigger a compaction in the active window.
    pub trigger_file_num: usize,
    /// Minimum L1 file num to allow a safety compaction in the active window.
    pub active_window_l1_merge_trigger: usize,
    /// Minimum file num to trigger a compaction in an inactive window.
    pub inactive_window_trigger_file_num: usize,
    /// Minimum L1 file num to trigger a compaction in an inactive window.
    pub inactive_window_l1_merge_trigger: usize,
    /// Compaction time window in seconds.
    pub time_window_seconds: Option<i64>,
    /// Max allowed compaction output file size. The picker also uses it to predict
    /// output splits when scoring candidates.
    pub max_output_file_size: Option<u64>,
    /// Whether the target region is in append mode.
    pub append_mode: bool,
    /// LastNonNull picks expand to an overlap closure, so cannot use partial
    /// small-file reductions that must leave large peers untouched.
    pub(crate) merge_mode: MergeMode,
    /// Max background compaction tasks.
    pub max_background_tasks: Option<usize>,
    /// Optional time range that constrains candidate compaction windows.
    pub(crate) time_range: Option<TimestampRange>,
    /// Picks the active window like an inactive one. Manual compactions set it,
    /// so a window that stopped receiving writes still gets its overlaps
    /// resolved.
    pub(crate) all_windows_inactive: bool,
}

impl TwcsPicker {
    /// Discovers window alternatives. Stateless callers retain their legacy
    /// first-strategy ordering and early output limit; full discovery does not
    /// rank or truncate candidates.
    async fn build_candidates(
        &self,
        region_id: RegionId,
        time_windows: BTreeMap<i64, Window>,
        context: WindowOutputContext,
    ) -> Result<TwcsCandidates> {
        let all_candidates = context.all_candidates;
        let mode = if all_candidates { "fair" } else { "stateless" };
        let _timer = self
            .metrics
            .elapsed
            .with_label_values(&[mode])
            .start_timer();
        let scanned_windows = self.metrics.scanned_windows.with_label_values(&[mode]);
        let WindowOutputContext {
            all_candidates: _,
            expired_ssts,
            active_window,
            time_window_size,
            max_outputs,
            primary_key_mapper,
        } = context;
        let region_sequence = time_windows
            .values()
            .flat_map(|window| window.files())
            .filter_map(|file| file.meta_ref().sequence)
            .map(|sequence| sequence.get())
            .max()
            .unwrap_or(0);
        let windows = time_windows
            .values()
            .rev()
            .filter(|window| {
                !window.files.is_empty()
                    && self.time_range.as_ref().is_none_or(|time_range| {
                        time_window_size.is_none_or(|time_window_size| {
                            time_window_intersects_range(
                                window.time_window,
                                time_window_size,
                                time_range,
                            )
                        })
                    })
            })
            .map(|window| window.time_window)
            .collect::<Vec<_>>();
        let time_windows = Arc::new(time_windows);
        let max_outputs = max_outputs.unwrap_or(usize::MAX);
        // Enumerating all LastNonNull seeds must not increase the number of
        // concurrently planned windows along with the output limit.
        let chunk_size = self.max_background_tasks.unwrap_or(windows.len()).max(1);
        let mut picks = Vec::new();
        for chunk in windows.chunks(chunk_size) {
            let mut handles = Vec::with_capacity(chunk.len());
            for &window in chunk {
                let picker = self.clone();
                let time_windows = time_windows.clone();
                let primary_key_mapper = primary_key_mapper.clone();
                let scanned_windows = scanned_windows.clone();
                handles.push(common_runtime::spawn_blocking_compact(move || {
                    let mut picks = Vec::new();
                    if let Some(files) = time_windows.get(&window) {
                        scanned_windows.inc();
                        picks.extend(
                            picker
                                .pick_window(
                                    files,
                                    &WindowPickContext {
                                        all_candidates,
                                        region_id,
                                        active_window,
                                        time_window_size,
                                        region_sequence,
                                        mapper: &primary_key_mapper,
                                    },
                                )
                                .into_iter()
                                .map(|pick| WindowCandidate { window, pick }),
                        );
                    }
                    picks
                }));
                tokio::task::yield_now().await;
            }
            for result in futures::future::join_all(handles).await {
                picks.extend(result.context(JoinSnafu)?);
            }
            // Windows are visited newest first, so older windows cannot displace
            // a full set of picks of the first class.
            if !all_candidates
                && picks
                    .iter()
                    .filter(|candidate| candidate.pick.class == PickClass::L0Merge)
                    .count()
                    >= max_outputs
            {
                break;
            }
        }

        if !all_candidates {
            // Manual/stateless callers retain their existing early-exit policy.
            picks.sort_by_key(|candidate| candidate.pick.class);
            if picks.len() > max_outputs {
                debug!(
                    "Region ({:?}) compaction output limit ({}) reached, remaining candidates discarded",
                    region_id, max_outputs
                );
                picks.truncate(max_outputs);
            }
        }
        Ok(TwcsCandidates {
            candidates: picks,
            expired_ssts,
            // Snapshot discovery always supplies the width; direct window tests
            // may omit it to disable settling-window behavior.
            time_window_size: time_window_size.unwrap_or_default(),
            windows: time_windows,
            active_window,
            primary_key_mapper,
            region_id,
            append_mode: self.append_mode,
            max_output_file_size: self.max_output_file_size,
        })
    }

    /// Prepares one window once. Fair discovery exposes independent alternatives;
    /// stateless/manual discovery retains the first eligible strategy.
    fn pick_window(&self, window: &Window, context: &WindowPickContext<'_>) -> Vec<WindowPick> {
        let WindowPickContext {
            all_candidates,
            region_id,
            active_window,
            time_window_size,
            region_sequence,
            mapper,
        } = *context;
        let mut files: Vec<_> = window.files().cloned().collect();
        let quiet = is_quiet_window(&files, region_sequence);

        // Filter out large files in append mode - they won't benefit from compaction
        if self.append_mode
            && let Some(max_size) = self.max_output_file_size
        {
            let (kept_files, ignored_files) =
                files.into_iter().partition(|file| file.size() <= max_size);
            files = kept_files;
            if !ignored_files.is_empty() {
                info!(
                    "Skipped {} large files in append mode for region {}, window {}, max_size: {}",
                    ignored_files.len(),
                    region_id,
                    window.time_window,
                    max_size
                );
            }
        }

        let (l0_files, l1_files): (Vec<_>, Vec<_>) =
            files.into_iter().partition(|file| file.level() == 0);
        if !self.all_windows_inactive && active_window == Some(window.time_window) {
            self.pick_active_window(l0_files, l1_files, mapper, all_candidates)
        } else if window.files.len() < self.inactive_window_trigger_file_num
            && l1_files.len() < self.inactive_window_l1_merge_trigger
        {
            Vec::new()
        } else {
            let settling = !self.all_windows_inactive
                && is_settling_window(window.time_window, active_window, time_window_size);
            let num_l1_files = l1_files.len();
            let mut picks = self.pick_inactive_window(
                l0_files,
                l1_files,
                mapper,
                settling,
                quiet,
                all_candidates,
            );
            let needs_reduction = if all_candidates {
                !picks
                    .iter()
                    .any(|pick| pick.class == PickClass::FileReduction)
            } else {
                picks.is_empty()
            };
            if needs_reduction
                && self.merge_mode != MergeMode::LastNonNull
                && num_l1_files >= self.inactive_window_l1_merge_trigger
                // Include even append-mode files excluded from normal picks:
                // a small-file merge must not add conflicts with large peers.
                && let Some(pick) = units::pick_small_l1(&window.files, mapper, self.pick_limits())
            {
                picks.push(WindowPick {
                    class: PickClass::FileReduction,
                    branch: "inactive:pick_small_l1",
                    inputs: pick.inputs,
                    found_runs: pick.units,
                });
            }
            picks
        }
    }

    /// The active window keeps receiving writes, so every pick must reduce its
    /// file count:
    ///
    /// 1. Merge L0 files once they reach the trigger.
    /// 2. Otherwise, merge L1 files once they reach the L1 safety trigger. L1
    ///    files are merged separately from L0 files to avoid chained rewrites.
    ///
    /// Fair discovery considers both strategies; stateless/manual discovery
    /// stops at the first eligible one.
    ///
    /// Overlaps are left for the window to resolve once it becomes inactive:
    /// resolving them while data keeps arriving rewrites the window repeatedly,
    /// and leaves a dominant run that blocks the inactive-window merge.
    fn pick_active_window(
        &self,
        l0_files: Vec<FileHandle>,
        l1_files: Vec<FileHandle>,
        mapper: &PrimaryKeyRangeMapper,
        all_candidates: bool,
    ) -> Vec<WindowPick> {
        let mut picks = Vec::new();
        if l0_files.len() >= self.trigger_file_num {
            let (inputs, found_runs) = pick_candidate_files(
                l0_files,
                self.max_output_file_size,
                pick_count_first,
                mapper,
            );
            if !inputs.is_empty() {
                picks.push(WindowPick {
                    class: PickClass::L0Merge,
                    branch: "active:pick_count_first",
                    inputs,
                    found_runs,
                });
                if !all_candidates {
                    return picks;
                }
            }
        }
        // Fair discovery must expose L1 even when L0 has an eligible merge.
        if l1_files.len() >= self.active_window_l1_merge_trigger
            && let Some(pick) = units::pick_active_l1(&l1_files, mapper, self.pick_limits())
        {
            picks.push(WindowPick {
                class: PickClass::FileReduction,
                branch: "active:pick_l1_file_reduction",
                inputs: pick.inputs,
                found_runs: pick.units,
            });
        }
        picks
    }

    /// An inactive window should converge at a bounded rewrite cost:
    ///
    /// 1. Absorb L0 files (late data), see [`pick_inactive_l0`].
    /// 2. Otherwise, merge a closed group of files that resolves overlaps or
    ///    removes files, see [`units::pick_closed_merge`]. Pure file reduction
    ///    waits for the L1 trigger; overlap resolution does not.
    ///
    /// Fair discovery keeps the L0 candidate and both closed-merge classes as
    /// alternatives; stateless/manual discovery retains the fallback order.
    ///
    /// A `settling` window is the one right before the active window. Late data
    /// still lands there, and any rewrite of its overlapping files would be
    /// repeated once more late data arrives. It only merges L0 files among
    /// themselves and conflict-free files, and leaves overlaps for when the
    /// window is older.
    ///
    /// Late data can also land in older windows. Resolving such an overlap
    /// rewrites a dominant sorted run for a small peer, which would repeat for
    /// every late flush, so it waits until the window is `quiet`: the region
    /// has written a window's worth of rows elsewhere since the window last
    /// received data. Each quiet period then pays for at most one such rewrite.
    /// A manual compaction is a one-off request to resolve overlaps, so it does
    /// not wait.
    ///
    /// Without new writes each pick consumes an L0 file or makes progress on
    /// the closed merge's termination measure, so the window stops compacting.
    fn pick_inactive_window(
        &self,
        l0_files: Vec<FileHandle>,
        l1_files: Vec<FileHandle>,
        mapper: &PrimaryKeyRangeMapper,
        settling: bool,
        quiet: bool,
        all_candidates: bool,
    ) -> Vec<WindowPick> {
        let num_l1_files = l1_files.len();
        let mut picks = Vec::new();
        if !l0_files.is_empty()
            && let Some((inputs, found_runs, branch)) = pick_inactive_l0(
                l0_files.clone(),
                (!settling).then_some(l1_files.as_slice()),
                self.max_output_file_size,
                mapper,
            )
        {
            picks.push(WindowPick {
                class: PickClass::L0Merge,
                branch,
                inputs,
                found_runs,
            });
            if !all_candidates {
                return picks;
            }
        }

        let options = ClosedMergeOptions {
            limits: self.closed_merge_limits(),
            allow_file_reduction: num_l1_files >= self.inactive_window_l1_merge_trigger,
            allow_conflicts: !settling,
            allow_unbalanced: (quiet || self.all_windows_inactive) && !settling,
        };
        // Both classes must see L0 peers: they can absorb late files stranded
        // by the L0 strategies, and must not introduce conflicts with them.
        let mut files = l1_files;
        files.extend(l0_files);
        let candidates = if all_candidates {
            units::pick_closed_merge_by_class(&files, mapper, options)
        } else {
            [units::pick_closed_merge(&files, mapper, options), None]
        };
        for pick in candidates.into_iter().flatten() {
            let (class, branch) = if pick.dense_fallback {
                (PickClass::FileReduction, "inactive:dense_file_reduction")
            } else if pick.reduces_files {
                (PickClass::FileReduction, "inactive:pick_l1_file_reduction")
            } else {
                (PickClass::OverlapResolution, "inactive:pick_l1_overlap")
            };
            picks.push(WindowPick {
                class,
                branch,
                inputs: pick.inputs,
                found_runs: pick.units,
            });
        }
        picks
    }

    fn pick_limits(&self) -> PickLimits {
        PickLimits {
            max_output_file_size: self.max_output_file_size,
            max_input_files: *MAX_INPUT_FILES,
        }
    }

    fn closed_merge_limits(&self) -> PickLimits {
        PickLimits {
            max_input_files: MAX_INPUT_FILES.saturating_mul(CLOSED_MERGE_INPUT_FACTOR),
            ..self.pick_limits()
        }
    }
}

/// Absorbs L0 files of an inactive window, in order of preference:
///
/// 1. Balanced merges of L0 files, which leave L1 files untouched.
/// 2. Mixed merges that bypass the balance checks, but only when the total
///    rewrite fits in the output file budget, bounding write amplification.
///    Skipped without `l1_files`.
/// 3. Merges of L0 files among themselves regardless of balance. The rewrite
///    is bounded by the L0 bytes and leaves large compacted files untouched.
///
/// Every pick consumes L0 files.
fn pick_inactive_l0(
    l0_files: Vec<FileHandle>,
    l1_files: Option<&[FileHandle]>,
    max_output_file_size: Option<u64>,
    mapper: &PrimaryKeyRangeMapper,
) -> Option<(Vec<FileHandle>, usize, &'static str)> {
    let (inputs, found_runs) = pick_candidate_files(
        l0_files.clone(),
        max_output_file_size,
        pick_count_first,
        mapper,
    );
    if !inputs.is_empty() {
        return Some((inputs, found_runs, "inactive:pick_count_first"));
    }
    if let Some(l1_files) = l1_files {
        let mut all_files = l0_files.clone();
        all_files.extend_from_slice(l1_files);
        let (inputs, found_runs) = pick_candidate_files(
            all_files,
            max_output_file_size,
            pick_mixed_within_budget,
            mapper,
        );
        if !inputs.is_empty() {
            return Some((inputs, found_runs, "inactive:pick_mixed_within_budget"));
        }
    }
    let (inputs, found_runs) = pick_candidate_files(
        l0_files,
        max_output_file_size,
        pick_unbalanced_count_first,
        mapper,
    );
    (!inputs.is_empty()).then_some((inputs, found_runs, "inactive:pick_unbalanced_count_first"))
}

type CandidatePicker =
    fn(Vec<SortedRun<FileHandle>>, Option<u64>, &PrimaryKeyRangeMapper) -> Vec<FileHandle>;

fn pick_candidate_files(
    mut files: Vec<FileHandle>,
    max_output_file_size: Option<u64>,
    picker: CandidatePicker,
    mapper: &PrimaryKeyRangeMapper,
) -> (Vec<FileHandle>, usize) {
    let sorted_runs = if files.len() < 1024 {
        find_sorted_runs(&mut files, |lhs, rhs| files_overlap(lhs, rhs, mapper))
    } else {
        find_sorted_runs_by_time_range(&mut files)
    };
    let found_runs = sorted_runs.len();
    (
        picker(sorted_runs, max_output_file_size, mapper),
        found_runs,
    )
}

#[derive(Debug)]
struct OrderedFile<'a> {
    file: &'a FileHandle,
    run_id: usize,
    position_in_run: usize,
    /// Primary key range in the pick's schema, resolved once per pick.
    primary_key_range: Option<(bytes::Bytes, bytes::Bytes)>,
}

impl OrderedFile<'_> {
    /// Same as [`files_overlap_inclusive`] with the cached key ranges.
    fn overlaps_inclusive(&self, other: &Self) -> bool {
        self.file.overlap_inclusive(other.file)
            && match (&self.primary_key_range, &other.primary_key_range) {
                (Some(lhs), Some(rhs)) => primary_key_ranges_overlap(lhs, rhs),
                _ => true,
            }
    }
}

/// Metrics of a candidate compaction input, accumulated one file at a time.
#[derive(Debug, Default)]
struct Candidate {
    /// Number of input files.
    num_files: usize,
    /// Total input bytes.
    total_size: usize,
    /// Size of the largest single input file.
    largest_file_size: usize,
    /// Files in the interval that overlap another interval file from a different
    /// sorted run. Resolving these overlaps is what reduces sorted runs.
    overlap_participants: usize,
    /// Number of rows contributed by uncompacted files.
    l0_rows: usize,
    /// Number of rows contributed by compacted files.
    l1_rows: usize,
    has_l0: bool,
    has_l1: bool,
    /// Whether at least one file uses 0 to represent an unknown row count.
    has_unknown_rows: bool,
}

impl Candidate {
    /// Absorbs `file` into the candidate. `preceding` holds the interval files added
    /// before it and `participations` tracks which of them already participate in an
    /// intra-interval overlap; both exist only for overlap accounting.
    fn absorb(
        &mut self,
        file: &OrderedFile,
        preceding: &[OrderedFile],
        participations: &mut Vec<bool>,
    ) {
        self.absorb_file(file.file);

        let mut participates = false;
        for (offset, other) in preceding.iter().enumerate() {
            if file.run_id != other.run_id && file.overlaps_inclusive(other) {
                if !participations[offset] {
                    participations[offset] = true;
                    self.overlap_participants += 1;
                }
                participates = true;
            }
        }
        participations.push(participates);
        if participates {
            self.overlap_participants += 1;
        }
    }

    /// Accumulates file statistics independently of overlap accounting.
    fn absorb_file(&mut self, file: &FileHandle) {
        self.num_files += 1;
        let file_size = file.size() as usize;
        self.total_size += file_size;
        self.largest_file_size = self.largest_file_size.max(file_size);
        self.absorb_level_rows(file);
    }

    fn absorb_level_rows(&mut self, file: &FileHandle) {
        let num_rows = file.num_rows();
        self.has_unknown_rows |= num_rows == 0;
        if file.level() == 0 {
            self.has_l0 = true;
            self.l0_rows = self.l0_rows.saturating_add(num_rows);
        } else {
            self.has_l1 = true;
            self.l1_rows = self.l1_rows.saturating_add(num_rows);
        }
    }

    /// Predicted number of output files after compacting this candidate. Compaction
    /// output holds at most the input bytes, so a split threshold of
    /// `max_output_file_size` yields at most `ceil(total_size / threshold)` files.
    /// Without a threshold the output is a single file.
    fn predicted_output_files(&self, max_output_file_size: Option<u64>) -> usize {
        match max_output_file_size {
            Some(max) if max > 0 => self.total_size.div_ceil(max as usize).max(1),
            _ => 1,
        }
    }

    /// Net file count reduction compacting this candidate would achieve, i.e. how
    /// many fewer physical SSTs the window holds afterwards. Zero when the output
    /// would be split back into at least as many files as the input.
    fn file_reduction(&self, max_output_file_size: Option<u64>) -> usize {
        self.num_files
            .saturating_sub(self.predicted_output_files(max_output_file_size))
    }

    /// A large historical file is only rewritten once its peers contribute at least
    /// the same amount of data, so every rewrite moves it into a larger size tier.
    fn is_balanced(&self) -> bool {
        self.largest_file_size <= self.total_size - self.largest_file_size
    }

    /// Bounds rewrite amplification for mixed L0/L1 candidates. Legacy files with
    /// unknown row counts retain the byte-balance behavior above.
    fn has_balanced_level_rows(&self) -> bool {
        !self.has_l0
            || !self.has_l1
            || self.has_unknown_rows
            || self.l1_rows <= self.l0_rows.saturating_mul(MAX_L1_L0_ROW_RATIO)
    }

    fn has_mixed_levels(&self) -> bool {
        self.has_l0 && self.has_l1
    }

    /// Whether rewriting this candidate stays within the output file budget.
    /// Inactive-window merges use this to bound write amplification when they
    /// bypass the balance checks. An unset budget means no bound: the operator
    /// removed the output size limit explicitly.
    fn within_rewrite_budget(&self, max_output_file_size: Option<u64>) -> bool {
        match max_output_file_size {
            Some(limit) if limit > 0 => self.total_size <= limit as usize,
            _ => true,
        }
    }

    /// A candidate is worth compacting only if it makes progress on at least one
    /// axis: it reduces the physical file count, or it resolves at least one
    /// overlap (merging sorted runs and reducing read amplification). A pure
    /// rewrite that achieves neither only burns I/O.
    fn makes_progress(&self, max_output_file_size: Option<u64>) -> bool {
        self.file_reduction(max_output_file_size) > 0 || self.overlap_participants > 0
    }

    fn score(&self, max_output_file_size: Option<u64>) -> CandidateScore {
        CandidateScore {
            file_reduction: self.file_reduction(max_output_file_size),
            overlap_participants: self.overlap_participants,
            total_size: self.total_size,
        }
    }
}

/// Score of a candidate compaction input. Higher is better: first by the predicted
/// net file count reduction, then by the number of files participating in an
/// overlap, then by smaller input bytes.
#[derive(Debug)]
struct CandidateScore {
    file_reduction: usize,
    overlap_participants: usize,
    total_size: usize,
}

impl CandidateScore {
    fn is_better_than(&self, other: &Self) -> bool {
        self.file_reduction
            .cmp(&other.file_reduction)
            .then_with(|| self.overlap_participants.cmp(&other.overlap_participants))
            .then_with(|| other.total_size.cmp(&self.total_size))
            .is_gt()
    }
}

/// Picks a contiguous (in global time order) interval of files to compact.
///
/// The picker reorders all files from all sorted runs by `(start asc, end desc)` and
/// enumerates every interval of at most [`MAX_INPUT_FILES`] files as a [`Candidate`].
/// An interval is eligible when it
///
/// - holds at least 2 files,
/// - passes the caller-supplied eligibility predicate (e.g. the byte- and
///   row-balance checks for regular picks, the rewrite budget for inactive
///   window fallbacks),
/// - makes progress on at least one axis: it reduces the physical file count given
///   the output split threshold `max_output_file_size`, or it resolves at least one
///   overlap between sorted runs.
///
/// The best interval wins by [`CandidateScore`]; ties keep the earliest interval.
fn pick_count_first(
    sorted_runs: Vec<SortedRun<FileHandle>>,
    max_output_file_size: Option<u64>,
    mapper: &PrimaryKeyRangeMapper,
) -> Vec<FileHandle> {
    pick_count_first_where(
        sorted_runs,
        max_output_file_size,
        mapper,
        is_balanced_candidate,
    )
}

#[cfg(test)]
fn pick_mixed_count_first(
    sorted_runs: Vec<SortedRun<FileHandle>>,
    max_output_file_size: Option<u64>,
    mapper: &PrimaryKeyRangeMapper,
) -> Vec<FileHandle> {
    pick_count_first_where(sorted_runs, max_output_file_size, mapper, |candidate| {
        candidate.has_mixed_levels() && is_balanced_candidate(candidate)
    })
}

/// Mixed-level fallback for inactive windows: bypasses the balance checks but
/// bounds the total rewrite to the output file budget.
fn pick_mixed_within_budget(
    sorted_runs: Vec<SortedRun<FileHandle>>,
    max_output_file_size: Option<u64>,
    mapper: &PrimaryKeyRangeMapper,
) -> Vec<FileHandle> {
    pick_count_first_where(sorted_runs, max_output_file_size, mapper, |candidate| {
        candidate.has_mixed_levels() && candidate.within_rewrite_budget(max_output_file_size)
    })
}

/// Last-resort pick for inactive windows with no balance requirement at all.
/// Callers must only pass single-level (L0) files so the rewrite stays bounded
/// by their bytes.
fn pick_unbalanced_count_first(
    sorted_runs: Vec<SortedRun<FileHandle>>,
    max_output_file_size: Option<u64>,
    mapper: &PrimaryKeyRangeMapper,
) -> Vec<FileHandle> {
    pick_count_first_where(sorted_runs, max_output_file_size, mapper, |_| true)
}

fn is_balanced_candidate(candidate: &Candidate) -> bool {
    candidate.is_balanced() && candidate.has_balanced_level_rows()
}

fn pick_count_first_where(
    sorted_runs: Vec<SortedRun<FileHandle>>,
    max_output_file_size: Option<u64>,
    mapper: &PrimaryKeyRangeMapper,
    is_eligible: impl Fn(&Candidate) -> bool,
) -> Vec<FileHandle> {
    let files = ordered_files(&sorted_runs, mapper);

    let mut best = None;
    for left in 0..files.len() {
        let mut candidate = Candidate::default();
        let right_bound = left.saturating_add(*MAX_INPUT_FILES).min(files.len());
        let mut participations: Vec<bool> = Vec::with_capacity(right_bound - left);
        for right in left..right_bound {
            candidate.absorb(&files[right], &files[left..right], &mut participations);
            if candidate.num_files < 2
                || !is_eligible(&candidate)
                || !candidate.makes_progress(max_output_file_size)
            {
                continue;
            }

            let score = candidate.score(max_output_file_size);
            if best
                .as_ref()
                .is_none_or(|(best_score, _)| score.is_better_than(best_score))
            {
                best = Some((score, &files[left..=right]));
            }
        }
    }

    let Some((_, best)) = best else {
        return vec![];
    };
    best.iter().map(|file| file.file.clone()).collect()
}

/// Flattens sorted runs into files ordered by `(start asc, end desc)`, breaking ties
/// by run and position to keep the order deterministic.
fn ordered_files<'a>(
    sorted_runs: &'a [SortedRun<FileHandle>],
    mapper: &PrimaryKeyRangeMapper,
) -> Vec<OrderedFile<'a>> {
    let mut files = sorted_runs
        .iter()
        .enumerate()
        .flat_map(|(run_id, run)| {
            run.items()
                .iter()
                .enumerate()
                .map(move |(position_in_run, file)| OrderedFile {
                    file,
                    run_id,
                    position_in_run,
                    primary_key_range: file.primary_key_range(mapper),
                })
        })
        .collect::<Vec<_>>();
    files.sort_unstable_by(|lhs, rhs| {
        let (lhs_start, lhs_end) = lhs.file.range();
        let (rhs_start, rhs_end) = rhs.file.range();
        lhs_start
            .cmp(&rhs_start)
            .then_with(|| rhs_end.cmp(&lhs_end))
            .then_with(|| lhs.run_id.cmp(&rhs.run_id))
            .then_with(|| lhs.position_in_run.cmp(&rhs.position_in_run))
    });
    files
}

fn selected_overlaps_unselected(
    selected: &[FileHandle],
    window: &Window,
    mapper: &PrimaryKeyRangeMapper,
) -> bool {
    // The overall time span of the selection: a file outside it cannot overlap any
    // selected file (ranges are inclusive), so it needs no precise overlap check.
    let Some((span_start, span_end)) = selected
        .iter()
        .map(Ranged::range)
        .reduce(|(start_a, end_a), (start_b, end_b)| (start_a.min(start_b), end_a.max(end_b)))
    else {
        return false;
    };
    let selected_file_ids = selected
        .iter()
        .map(FileHandle::file_id)
        .collect::<HashSet<_>>();
    window
        .files()
        .filter(|file| {
            let (start, end) = file.range();
            start <= span_end && span_start <= end
        })
        .filter(|file| !selected_file_ids.contains(&file.file_id()))
        .any(|unselected| {
            selected
                .iter()
                .any(|selected| files_overlap_inclusive(selected, unselected, mapper))
        })
}

fn log_pick_result(
    region_id: RegionId,
    window: i64,
    active_window: Option<i64>,
    pick: &WindowPick,
    file_num: usize,
    max_output_file_size: Option<u64>,
    filter_deleted: bool,
) {
    let input_file_str: Vec<String> = pick
        .inputs
        .iter()
        .map(|f| {
            let range = f.range();
            let start = range.0.to_iso8601_string();
            let end = range.1.to_iso8601_string();
            let num_rows = f.num_rows();
            format!(
                "File{{id: {:?}, range: ({}, {}), size: {}, num rows: {} }}",
                f.file_id(),
                start,
                end,
                ReadableSize(f.size()),
                num_rows
            )
        })
        .collect();
    let window_str = Timestamp::new_second(window).to_iso8601_string();
    let active_window_str = active_window.map(|s| Timestamp::new_second(s).to_iso8601_string());
    let max_output_file_size = max_output_file_size.map(|size| ReadableSize(size).to_string());
    info!(
        "Region ({:?}) compaction pick result: current window: {}, active window: {:?}, branch: {}, \
            found runs: {}, file num: {}, max output file size: {:?}, filter deleted: {}, \
            input files: {:?}",
        region_id,
        window_str,
        active_window_str,
        pick.branch,
        pick.found_runs,
        file_num,
        max_output_file_size,
        filter_deleted,
        input_file_str
    );
}

#[async_trait::async_trait]
impl Picker for TwcsPicker {
    async fn pick(&self, compaction_region: &CompactionRegion) -> Result<Option<PickerOutput>> {
        self.pick_with_output_limit(compaction_region, self.max_background_tasks)
            .await
    }
}

impl TwcsPicker {
    pub(crate) async fn pick_with_output_limit(
        &self,
        compaction_region: &CompactionRegion,
        max_outputs: Option<usize>,
    ) -> Result<Option<PickerOutput>> {
        let candidates = self
            .discover_with_limit(compaction_region, max_outputs, false)
            .await?;
        let output = candidates.finish().await?;
        Ok((!output.outputs.is_empty() || !output.expired_ssts.is_empty()).then_some(output))
    }

    /// Enumerates all eligible classes/windows without applying scheduling policy.
    pub(crate) async fn discover(&self, region: &CompactionRegion) -> Result<TwcsCandidates> {
        self.discover_with_limit(region, None, true).await
    }

    async fn discover_with_limit(
        &self,
        compaction_region: &CompactionRegion,
        max_outputs: Option<usize>,
        all_candidates: bool,
    ) -> Result<TwcsCandidates> {
        let region_id = compaction_region.region_id;
        let picker = self.clone();
        let compaction_region = compaction_region.clone();
        let primary_key_mapper = compaction_region.current_version.ssts.primary_key_mapper();
        let window_mapper = primary_key_mapper.clone();
        let (expired_ssts, time_window_size, active_window, windows) =
            common_runtime::spawn_blocking_compact(move || {
                let levels = compaction_region.current_version.ssts.levels();
                let expired_ssts = get_expired_ssts(
                    levels,
                    compaction_region.ttl,
                    Timestamp::current_millis(),
                );
                if !expired_ssts.is_empty() {
                    info!("Expired SSTs in region {}: {:?}", region_id, expired_ssts);
                }
                let expired_file_ids = expired_ssts
                    .iter()
                    .map(|file| file.file_id())
                    .collect::<HashSet<_>>();

                let compaction_time_window = compaction_region
                    .current_version
                    .compaction_time_window
                    .map(|window| window.as_secs() as i64);
                let time_window_size = compaction_time_window
                    .or(picker.time_window_seconds)
                    .unwrap_or_else(|| {
                        let inferred = infer_time_bucket(levels[0].files());
                        info!(
                            "Compaction window for region {} is not present, inferring from files: {:?}",
                            region_id, inferred
                        );
                        inferred
                    });

                let windows = assign_to_windows(
                    levels
                        .iter()
                        .flat_map(LevelMeta::files)
                        .filter(|file| !expired_file_ids.contains(&file.file_id())),
                    time_window_size,
                    &window_mapper,
                );
                // Compute activity from the candidate files so expired or
                // compacting files cannot identify a window absent from `windows`.
                let active_window = find_active_window_by_sequence(
                    windows.values().flat_map(Window::files),
                    time_window_size,
                )
                .or_else(|| {
                    find_latest_window_in_seconds(
                        windows
                            .values()
                            .flat_map(Window::files)
                            .filter(|file| file.level() == 0),
                        time_window_size,
                    )
                });

                (expired_ssts, time_window_size, active_window, windows)
            })
            .await
            .context(JoinSnafu)?;

        self.build_candidates(
            region_id,
            windows,
            WindowOutputContext {
                all_candidates,
                expired_ssts,
                active_window,
                time_window_size: Some(time_window_size),
                max_outputs,
                primary_key_mapper,
            },
        )
        .await
    }
}

#[derive(Clone)]
struct Window {
    start: Timestamp,
    end: Timestamp,
    files: Vec<FileHandle>,
    /// Start of the half-open window, in seconds.
    time_window: i64,
    primary_key_range: Option<(bytes::Bytes, bytes::Bytes)>,
}

impl Window {
    /// Creates a new [Window] with given file.
    fn new_with_file(file: FileHandle, mapper: &PrimaryKeyRangeMapper) -> Self {
        let (start, end) = file.time_range();
        let primary_key_range = file.primary_key_range(mapper);
        Self {
            start,
            end,
            files: vec![file],
            time_window: 0,
            primary_key_range,
        }
    }

    /// Returns the time range of all files in current window (inclusive).
    fn range(&self) -> (Timestamp, Timestamp) {
        (self.start, self.end)
    }

    /// Adds a new file to window and updates time range.
    fn add_file(&mut self, file: FileHandle, mapper: &PrimaryKeyRangeMapper) {
        let (start, end) = file.time_range();
        self.start = self.start.min(start);
        self.end = self.end.max(end);
        self.primary_key_range = merge_primary_key_ranges(
            self.primary_key_range.take(),
            file.primary_key_range(mapper),
        );
        self.files.push(file);
    }

    fn files(&self) -> impl Iterator<Item = &FileHandle> {
        self.files.iter()
    }
}

/// Whether `window` is the one right before `active_window`, where late data
/// still lands.
fn is_settling_window(
    window: i64,
    active_window: Option<i64>,
    time_window_size: Option<i64>,
) -> bool {
    active_window
        .zip(time_window_size)
        .is_some_and(|(active, size)| {
            // Windows at the timestamp limits may be too far apart to subtract.
            active
                .checked_sub(window)
                .is_some_and(|distance| (1..=size).contains(&distance))
        })
}

/// Whether the region has written at least as many rows as `files` hold, and at
/// least one row, since the newest of them was written. Sequences are assigned
/// per row and compaction keeps the inputs' maximum, so this needs no state.
/// Files written before row counts were recorded have zero rows, so a window
/// holding them is not quiet.
fn is_quiet_window(files: &[FileHandle], region_sequence: u64) -> bool {
    if files.iter().any(|file| file.num_rows() == 0) {
        return false;
    }
    let Some(window_sequence) = files
        .iter()
        .filter_map(|file| file.meta_ref().sequence)
        .map(|sequence| sequence.get())
        .max()
    else {
        return false;
    };
    let rows: u64 = files.iter().map(|file| file.num_rows() as u64).sum();
    region_sequence.saturating_sub(window_sequence) >= rows.max(1)
}

/// Assigns files to windows with predefined window size (in seconds) by their max timestamps.
fn assign_to_windows<'a>(
    files: impl Iterator<Item = &'a FileHandle>,
    time_window_size: i64,
    mapper: &PrimaryKeyRangeMapper,
) -> BTreeMap<i64, Window> {
    let mut windows: HashMap<i64, Window> = HashMap::new();
    // Iterates all files and assign to time windows according to max timestamp
    for f in files {
        if f.compacting() {
            continue;
        }
        let (_, end) = f.time_range();
        let time_window = time_window_start(end, time_window_size);

        match windows.entry(time_window) {
            Entry::Occupied(mut e) => {
                e.get_mut().add_file(f.clone(), mapper);
            }
            Entry::Vacant(e) => {
                let mut window = Window::new_with_file(f.clone(), mapper);
                window.time_window = time_window;
                e.insert(window);
            }
        }
    }
    windows.into_iter().collect()
}

/// Uses the same half-open windows as memtables, including timestamps in the first
/// second after a boundary. An unrepresentable window start is clamped to `i64::MIN`.
fn time_window_start(timestamp: Timestamp, time_window_size: i64) -> i64 {
    // Conversion to seconds cannot overflow.
    timestamp
        .convert_to(TimeUnit::Second)
        .and_then(|ts| ts.value().align_by_bucket(time_window_size))
        .unwrap_or(i64::MIN)
}

fn time_window_intersects_range(
    window_start: i64,
    time_window_size: i64,
    time_range: &TimestampRange,
) -> bool {
    if time_range.is_empty() {
        return false;
    }
    let first_window = time_range
        .start()
        .map_or(i64::MIN, |start| time_window_start(start, time_window_size));
    let last_window = match time_range.end() {
        None => i64::MAX,
        Some(end) => {
            // The range end is exclusive; keep the last second that intersects it.
            let Some(last_second) = end
                .convert_to_ceil(TimeUnit::Second)
                .and_then(|timestamp| timestamp.value().checked_sub(1))
            else {
                return false;
            };
            time_window_start(Timestamp::new_second(last_second), time_window_size)
        }
    };
    (first_window..=last_window).contains(&window_start)
}

fn window_has_overlap(this: &Window, windows: &BTreeMap<i64, Window>) -> bool {
    windows
        .values()
        .filter(|that| this.time_window != that.time_window)
        .any(|that| {
            overlaps(&this.range(), &that.range()) && {
                match (&this.primary_key_range, &that.primary_key_range) {
                    (Some(l), Some(r)) => primary_key_ranges_overlap(l, r),
                    _ => true,
                }
            }
        })
}

/// Finds the latest active writing window among all files.
/// Returns `None` when there are no files.
fn find_latest_window_in_seconds<'a>(
    files: impl Iterator<Item = &'a FileHandle>,
    time_window_size: i64,
) -> Option<i64> {
    let mut latest_timestamp = None;
    for f in files {
        let (_, end) = f.time_range();
        if let Some(latest) = latest_timestamp {
            if end > latest {
                latest_timestamp = Some(end);
            }
        } else {
            latest_timestamp = Some(end);
        }
    }
    latest_timestamp.map(|ts| time_window_start(ts, time_window_size))
}

/// Finds the active window from the file with the highest sequence number.
///
/// Flush and compaction outputs inherit the max input sequence, so the
/// max-sequence file always tracks the most recent write: the active window
/// survives the transient state where level 0 is empty right after its files
/// were compacted away. Returns `None` when no file carries a sequence.
///
/// The window key is its inclusive start, as in [`assign_to_windows`].
fn find_active_window_by_sequence<'a>(
    files: impl Iterator<Item = &'a FileHandle>,
    time_window_size: i64,
) -> Option<i64> {
    files
        .filter(|f| f.meta_ref().sequence.is_some())
        .max_by_key(|f| f.meta_ref().sequence)
        .map(|f| time_window_start(f.time_range().1, time_window_size))
}

#[cfg(test)]
mod tests {
    use std::collections::HashSet;
    use std::num::NonZeroU64;
    use std::sync::Arc;
    use std::time::{Duration, Instant};

    use common_base::Plugins;
    use common_time::range::TimestampRange;
    use store_api::storage::FileId;

    use super::*;
    use crate::cache::CacheManager;
    use crate::compaction::compactor::CompactionVersion;
    use crate::compaction::scheduler::history::{PickHistory, PickKey};
    use crate::compaction::scheduler::pick_automatic_for_test;
    use crate::compaction::test_util::{
        compaction_region_with_ssts, new_file_handle, new_file_handle_with_sequence,
        new_file_handle_with_size_and_sequence,
        new_file_handle_with_size_sequence_and_primary_key_range, pk_range,
        primary_key_mapper_for_test,
    };
    use crate::config::MitoConfig;
    use crate::region::options::RegionOptions;
    use crate::sst::file::{FileMeta, Level};
    use crate::sst::version::SstVersion;
    use crate::test_util::memtable_util::metadata_for_test;
    use crate::test_util::scheduler_util::SchedulerEnv;

    fn assign_to_windows<'a>(
        files: impl Iterator<Item = &'a FileHandle>,
        time_window_size: i64,
    ) -> BTreeMap<i64, Window> {
        super::assign_to_windows(files, time_window_size, &primary_key_mapper_for_test())
    }

    fn pick_count_first(
        sorted_runs: Vec<SortedRun<FileHandle>>,
        max_output_file_size: Option<u64>,
    ) -> Vec<FileHandle> {
        super::pick_count_first(
            sorted_runs,
            max_output_file_size,
            &primary_key_mapper_for_test(),
        )
    }

    fn pick_mixed_count_first(
        sorted_runs: Vec<SortedRun<FileHandle>>,
        max_output_file_size: Option<u64>,
    ) -> Vec<FileHandle> {
        super::pick_mixed_count_first(
            sorted_runs,
            max_output_file_size,
            &primary_key_mapper_for_test(),
        )
    }

    impl TwcsPicker {
        async fn build_output_with_time_range(
            &self,
            region_id: RegionId,
            time_windows: BTreeMap<i64, Window>,
            active_window: Option<i64>,
            time_window_size: Option<i64>,
        ) -> Result<Vec<CompactionOutput>> {
            self.build_candidates(
                region_id,
                time_windows,
                WindowOutputContext {
                    all_candidates: false,
                    expired_ssts: Vec::new(),
                    active_window,
                    time_window_size,
                    max_outputs: self.max_background_tasks,
                    primary_key_mapper: Arc::new(primary_key_mapper_for_test()),
                },
            )
            .await?
            .finish()
            .await
            .map(|output| output.outputs)
        }

        async fn build_output_with_history(
            &self,
            time_windows: BTreeMap<i64, Window>,
            active_window: Option<i64>,
            history: &mut PickHistory,
        ) -> Result<Vec<CompactionOutput>> {
            let mut candidates = self
                .build_candidates(
                    RegionId::from_u64(1),
                    time_windows,
                    WindowOutputContext {
                        all_candidates: true,
                        expired_ssts: Vec::new(),
                        active_window,
                        time_window_size: None,
                        max_outputs: None,
                        primary_key_mapper: Arc::new(primary_key_mapper_for_test()),
                    },
                )
                .await?;
            let now = Instant::now();
            candidates.candidates = history.rank_candidates(candidates.candidates, now);
            candidates
                .candidates
                .truncate(self.max_background_tasks.unwrap_or(usize::MAX));
            history.acknowledge(candidates.candidates.iter().map(PickKey::from), now);
            candidates.finish().await.map(|output| output.outputs)
        }
    }

    #[test]
    fn test_cross_schema_pk_overlap_before_window_aggregation() {
        use crate::test_util::sst_util::{new_primary_key, sst_region_metadata};

        let metadata = Arc::new(sst_region_metadata());
        let mut old = FileMeta {
            region_id: metadata.region_id,
            file_id: FileId::random(),
            time_range: (
                Timestamp::new_millisecond(0),
                Timestamp::new_millisecond(1000),
            ),
            primary_key_min: Some(new_primary_key(&["a"]).into()),
            primary_key_max: Some(new_primary_key(&["b"]).into()),
            ..Default::default()
        };
        let mut completed = new_primary_key(&["b"]);
        completed.push(0); // The appended nullable tag's default.
        let new = FileMeta {
            file_id: FileId::random(),
            time_range: (
                Timestamp::new_millisecond(500),
                Timestamp::new_millisecond(2000),
            ),
            primary_key_min: Some(completed.clone().into()),
            primary_key_max: Some(completed.clone().into()),
            ..old.clone()
        };
        assert!(old.primary_key_max < new.primary_key_min);
        let mut ssts = SstVersion::new(metadata);
        let old_id = old.file_id;
        let new_id = new.file_id;
        ssts.add_files(
            crate::test_util::new_noop_file_purger(),
            [old.clone(), new].into_iter(),
        );
        let old_file = &ssts.levels()[0].files[&old_id];
        let new_file = &ssts.levels()[0].files[&new_id];
        let ranges = ssts.primary_key_mapper();
        let windows = super::assign_to_windows([old_file, new_file].into_iter(), 1, &ranges);
        assert_eq!(2, windows.len());
        assert!(
            windows
                .values()
                .all(|window| window_has_overlap(window, &windows))
        );

        let mut window = Window::new_with_file(old_file.clone(), &ranges);
        window.add_file(new_file.clone(), &ranges);
        assert!(selected_overlaps_unselected(
            std::slice::from_ref(new_file),
            &window,
            &ranges,
        ));
        assert_eq!(
            Some(completed.into()),
            window
                .primary_key_range
                .as_ref()
                .map(|range| range.1.clone())
        );

        // In the same target schema, a genuinely disjoint range still prunes.
        old.file_id = FileId::random();
        old.primary_key_max = old.primary_key_min.clone();
        let disjoint_id = old.file_id;
        ssts.add_files(crate::test_util::new_noop_file_purger(), [old].into_iter());
        let files = &ssts.levels()[0].files;
        assert!(!files_overlap_inclusive(
            &files[&disjoint_id],
            &files[&new_id],
            &ranges
        ));
    }

    #[test]
    fn test_valid_max_input_files_env_overrides_default() {
        assert_eq!(64, parse_max_input_files(Some("64")));
    }

    #[test]
    fn test_invalid_max_input_files_env_falls_back_to_default() {
        for env_value in [None, Some(""), Some("invalid"), Some("0"), Some("1")] {
            assert_eq!(16, parse_max_input_files(env_value));
        }
    }

    async fn compaction_region_with_expired_sst() -> CompactionRegion {
        compaction_region_with_ssts(
            (1..=4).map(|sequence| FileMeta {
                file_id: FileId::random(),
                time_range: (
                    Timestamp::new_millisecond(0),
                    Timestamp::new_millisecond(10),
                ),
                level: 0,
                sequence: NonZeroU64::new(sequence),
                ..Default::default()
            }),
            Duration::from_millis(1),
        )
        .await
    }

    #[tokio::test]
    async fn test_pick_normalizes_zero_output_size_to_unlimited() {
        let mut compaction_region = compaction_region_with_ssts(
            (1..=4).map(|sequence| new_file(0, 10, sequence, 100).meta_ref().clone()),
            Duration::from_secs(3600),
        )
        .await;
        compaction_region.ttl = None;

        for (max_output_file_size, expected_max_file_size) in
            [(Some(0), None), (None, None), (Some(1024), Some(1024))]
        {
            let picker = TwcsPicker {
                trigger_file_num: 4,
                active_window_l1_merge_trigger: 8,
                inactive_window_trigger_file_num: 4,
                inactive_window_l1_merge_trigger: 8,
                time_window_seconds: Some(3),
                max_output_file_size,
                append_mode: false,
                merge_mode: MergeMode::LastRow,
                max_background_tasks: None,
                time_range: None,
                all_windows_inactive: false,
                metrics: Default::default(),
            };

            let output = picker.pick(&compaction_region).await.unwrap().unwrap();

            assert_eq!(output.outputs.len(), 1);
            assert_eq!(output.outputs[0].inputs.len(), 4);
            assert!(output.expired_ssts.is_empty());
            assert_eq!(output.max_file_size, expected_max_file_size);
        }
    }

    #[tokio::test]
    async fn test_pick_expired_ssts_without_marking_compacting() {
        let picker = TwcsPicker {
            trigger_file_num: 4,
            active_window_l1_merge_trigger: 8,
            inactive_window_trigger_file_num: 4,
            inactive_window_l1_merge_trigger: 8,
            time_window_seconds: Some(3),
            max_output_file_size: None,
            append_mode: false,
            merge_mode: MergeMode::LastRow,
            max_background_tasks: None,
            time_range: None,
            all_windows_inactive: false,
            metrics: Default::default(),
        };
        let compaction_region = compaction_region_with_expired_sst().await;

        let output = picker.pick(&compaction_region).await.unwrap().unwrap();

        assert!(output.outputs.is_empty());
        assert!(!output.expired_ssts.is_empty());
        assert!(output.expired_ssts.iter().all(|file| !file.compacting()));
    }

    #[tokio::test]
    async fn test_expired_sst_does_not_determine_active_window() {
        let now = Timestamp::current_millis().value();
        let files = [
            FileMeta {
                file_id: FileId::random(),
                time_range: (
                    Timestamp::new_millisecond(0),
                    Timestamp::new_millisecond(10),
                ),
                level: 0,
                sequence: NonZeroU64::new(100),
                ..Default::default()
            },
            FileMeta {
                file_id: FileId::random(),
                time_range: (
                    Timestamp::new_millisecond(now - 1000),
                    Timestamp::new_millisecond(now),
                ),
                level: 0,
                sequence: NonZeroU64::new(1),
                ..Default::default()
            },
            FileMeta {
                file_id: FileId::random(),
                time_range: (
                    Timestamp::new_millisecond(now - 1000),
                    Timestamp::new_millisecond(now),
                ),
                level: 0,
                sequence: NonZeroU64::new(2),
                ..Default::default()
            },
        ];
        let compaction_region = compaction_region_with_ssts(files, Duration::from_secs(60)).await;
        let picker = TwcsPicker {
            trigger_file_num: 4,
            active_window_l1_merge_trigger: 8,
            inactive_window_trigger_file_num: 2,
            inactive_window_l1_merge_trigger: 8,
            time_window_seconds: Some(3),
            max_output_file_size: None,
            append_mode: false,
            merge_mode: MergeMode::LastRow,
            max_background_tasks: None,
            time_range: None,
            all_windows_inactive: false,
            metrics: Default::default(),
        };

        let output = picker.pick(&compaction_region).await.unwrap().unwrap();

        assert_eq!(1, output.expired_ssts.len());
        assert!(output.outputs.is_empty());
    }

    #[tokio::test]
    async fn test_active_window_survives_l0_consumption() {
        // Simulates the transient state right after an L0 compaction: level 0 is
        // empty and only L1 files remain. The active window must still be the
        // window of the most recent write (tracked by the max-sequence file), so
        // the window is NOT compacted under the inactive rules.
        let env = SchedulerEnv::new().await;
        let metadata = metadata_for_test();
        let manifest_ctx = env.mock_manifest_context(metadata.clone()).await;
        let mut ssts = SstVersion::new(metadata.clone());
        ssts.add_files(
            Arc::new(crate::sst::file_purger::NoopFilePurger),
            [100, 101].into_iter().map(|sequence| FileMeta {
                region_id: metadata.region_id,
                file_id: FileId::random(),
                time_range: (
                    Timestamp::new_millisecond(0),
                    Timestamp::new_millisecond(10),
                ),
                level: 1,
                sequence: NonZeroU64::new(sequence),
                ..Default::default()
            }),
        );
        let compaction_region = CompactionRegion {
            region_id: metadata.region_id,
            region_options: RegionOptions::default(),
            engine_config: Arc::new(MitoConfig::default()),
            region_metadata: metadata.clone(),
            cache_manager: Arc::new(CacheManager::default()),
            access_layer: env.access_layer,
            manifest_ctx,
            current_version: CompactionVersion {
                metadata,
                options: RegionOptions::default(),
                ssts: Arc::new(ssts),
                memtable_min_sequence: None,
                compaction_time_window: None,
            },
            file_purger: None,
            ttl: None,
            max_parallelism: 1,
            plugins: Plugins::new(),
        };
        let picker = TwcsPicker {
            trigger_file_num: 4,
            active_window_l1_merge_trigger: 8,
            inactive_window_trigger_file_num: 2,
            inactive_window_l1_merge_trigger: 8,
            time_window_seconds: Some(3600),
            max_output_file_size: None,
            append_mode: false,
            merge_mode: MergeMode::LastRow,
            max_background_tasks: None,
            time_range: None,
            all_windows_inactive: false,
            metrics: Default::default(),
        };

        assert!(picker.pick(&compaction_region).await.unwrap().is_none());
    }

    #[test]
    fn test_find_active_window_by_sequence() {
        // The max-sequence file tracks the most recent write regardless of
        // level: here a compacted L1 file ending at 2999ms (window key 2) is
        // newer than the L0 file ending at 999ms (window key 0). Window keys
        // follow the assign_to_windows convention (truncate to seconds).
        let files = [
            new_file_handle_with_sequence(FileId::random(), 0, 999, 0, 1),
            new_file_handle_with_sequence(FileId::random(), 2000, 2999, 1, 10),
        ];
        assert_eq!(Some(2), find_active_window_by_sequence(files.iter(), 1));

        // Backfill: the newest write lands in an older window, and that window
        // is the active one.
        let files = [
            new_file_handle_with_sequence(FileId::random(), 0, 999, 0, 10),
            new_file_handle_with_sequence(FileId::random(), 2000, 2999, 0, 1),
        ];
        assert_eq!(Some(0), find_active_window_by_sequence(files.iter(), 1));

        // Files without a sequence are skipped; if none have one, the caller
        // falls back to the L0-based rule.
        let files = [new_file_handle_with_sequence(
            FileId::random(),
            0,
            999,
            1,
            0,
        )];
        assert_eq!(None, find_active_window_by_sequence(files.iter(), 1));
        assert!(find_active_window_by_sequence(Vec::<FileHandle>::new().iter(), 1).is_none());
    }

    #[test]
    fn test_active_window_falls_back_to_l0_rule_without_sequence() {
        let active_window_of = |files: &[FileHandle]| {
            find_active_window_by_sequence(files.iter(), 1).or_else(|| {
                find_latest_window_in_seconds(files.iter().filter(|f| f.level() == 0), 1)
            })
        };

        // Legacy files without sequence: the L0-based rule still applies.
        let files = [
            new_file_handle_with_sequence(FileId::random(), 0, 999, 1, 0),
            new_file_handle_with_sequence(FileId::random(), 2000, 2999, 0, 0),
        ];
        assert_eq!(Some(2), active_window_of(&files));

        // No sequence and no L0: None, exactly as before.
        let files = [new_file_handle_with_sequence(
            FileId::random(),
            0,
            999,
            1,
            0,
        )];
        assert_eq!(None, active_window_of(&files));
    }

    #[test]
    fn test_get_latest_window_in_seconds() {
        assert_eq!(
            Some(0),
            find_latest_window_in_seconds([new_file_handle(FileId::random(), 0, 999, 0)].iter(), 1)
        );
        assert_eq!(
            Some(1),
            find_latest_window_in_seconds(
                [new_file_handle(FileId::random(), 0, 1000, 0)].iter(),
                1
            )
        );

        assert_eq!(
            Some(-9223372036857600),
            find_latest_window_in_seconds(
                [new_file_handle(FileId::random(), i64::MIN, i64::MIN + 1, 0)].iter(),
                3600,
            )
        );

        assert_eq!(
            (i64::MAX / 10000000) * 10000,
            find_latest_window_in_seconds(
                [new_file_handle(FileId::random(), i64::MIN, i64::MAX, 0)].iter(),
                10000,
            )
            .unwrap()
        );

        assert_eq!(
            Some((i64::MAX / 3600000) * 3600),
            find_latest_window_in_seconds(
                [
                    new_file_handle(FileId::random(), i64::MIN, i64::MAX, 0),
                    new_file_handle(FileId::random(), 0, 1000, 0)
                ]
                .iter(),
                3600
            )
        );
    }

    #[test]
    fn test_window_assignment_and_activity_at_boundaries() {
        // Before, at, and after positive/negative boundaries, including the first
        // fractional second: all picker paths must agree with half-open partitions.
        for (unit, ticks_per_second) in [
            (TimeUnit::Second, 1),
            (TimeUnit::Millisecond, 1_000),
            (TimeUnit::Microsecond, 1_000_000),
            (TimeUnit::Nanosecond, 1_000_000_000),
        ] {
            for window_size in [1, 3, 86_400] {
                for boundary in [-window_size, 0, window_size] {
                    for (offset, expected_window) in [
                        (-1, boundary - window_size),
                        (0, boundary),
                        (ticks_per_second - 1, boundary),
                        (ticks_per_second, boundary + i64::from(window_size == 1)),
                    ] {
                        let timestamp = Timestamp::new(boundary * ticks_per_second + offset, unit);
                        let file = FileHandle::new(
                            FileMeta {
                                time_range: (timestamp, timestamp),
                                sequence: NonZeroU64::new(1),
                                ..Default::default()
                            },
                            crate::test_util::new_noop_file_purger(),
                        );
                        let files = [file];
                        let windows = assign_to_windows(files.iter(), window_size);
                        assert_eq!(
                            vec![expected_window],
                            windows.keys().copied().collect::<Vec<_>>(),
                            "{timestamp:?}, window size {window_size}"
                        );
                        assert_eq!(
                            Some(expected_window),
                            find_active_window_by_sequence(files.iter(), window_size)
                        );
                        assert_eq!(
                            Some(expected_window),
                            find_latest_window_in_seconds(files.iter(), window_size)
                        );
                    }
                }
            }
        }
    }

    #[test]
    fn test_settling_window_at_timestamp_limits() {
        assert!(is_settling_window(0, Some(60), Some(60)));
        assert!(!is_settling_window(60, Some(60), Some(60)));
        assert!(!is_settling_window(-60, Some(60), Some(60)));
        assert!(!is_settling_window(0, None, Some(60)));
        // The distance between these windows does not fit in an i64.
        assert!(!is_settling_window(i64::MAX, Some(i64::MIN), Some(1)));
        assert!(!is_settling_window(i64::MIN, Some(i64::MAX), Some(1)));
    }

    #[test]
    fn test_window_assignment_at_timestamp_limits() {
        for (seconds, expected_window) in [(i64::MIN, i64::MIN), (i64::MAX, i64::MAX - 1)] {
            let timestamp = Timestamp::new_second(seconds);
            let files = [FileHandle::new(
                FileMeta {
                    time_range: (timestamp, timestamp),
                    sequence: NonZeroU64::new(1),
                    ..Default::default()
                },
                crate::test_util::new_noop_file_purger(),
            )];
            let windows = assign_to_windows(files.iter(), 3);
            assert_eq!(
                vec![expected_window],
                windows.keys().copied().collect::<Vec<_>>()
            );
            assert_eq!(
                Some(expected_window),
                find_active_window_by_sequence(files.iter(), 3)
            );
            assert_eq!(
                Some(expected_window),
                find_latest_window_in_seconds(files.iter(), 3)
            );
            assert!(time_window_intersects_range(
                expected_window,
                3,
                &TimestampRange::single(timestamp)
            ));
        }
    }

    #[tokio::test]
    async fn test_midnight_file_is_not_compacted_with_previous_day() {
        // Regression for #9441: flush already separates the midnight-only SST.
        let midnight = 86_400_000;
        CompactionPickerTestCase {
            window_size: 86_400,
            input_files: [
                (midnight - 240_000, midnight - 180_000),
                (midnight - 180_000, midnight - 90_000),
                (midnight - 90_000, midnight - 30_000),
                (midnight, midnight),
            ]
            .into_iter()
            .enumerate()
            .map(|(index, (start, end))| {
                new_file_handle_with_sequence(FileId::random(), start, end, 0, index as u64 + 1)
            })
            .collect(),
            expected_outputs: vec![ExpectedOutput {
                input_files: vec![0, 1, 2],
                output_level: 1,
            }],
        }
        .check()
        .await;
    }

    #[test]
    fn test_assign_to_windows() {
        let windows = assign_to_windows(
            [
                new_file_handle(FileId::random(), 0, 999, 0),
                new_file_handle(FileId::random(), 0, 999, 0),
                new_file_handle(FileId::random(), 0, 999, 0),
                new_file_handle(FileId::random(), 0, 999, 0),
                new_file_handle(FileId::random(), 0, 999, 0),
            ]
            .iter(),
            3,
        );
        let fgs = &windows.get(&0).unwrap().files;
        assert_eq!(5, fgs.len());

        let files = [FileId::random(); 3];
        let windows = assign_to_windows(
            [
                new_file_handle(files[0], -2000, -3, 0),
                new_file_handle(files[1], 0, 2999, 0),
                new_file_handle(files[2], 50, 10001, 0),
            ]
            .iter(),
            3,
        );
        assert_eq!(
            files[0],
            windows
                .get(&-3)
                .unwrap()
                .files()
                .next()
                .unwrap()
                .file_id()
                .file_id()
        );
        assert_eq!(
            files[1],
            windows
                .get(&0)
                .unwrap()
                .files()
                .next()
                .unwrap()
                .file_id()
                .file_id()
        );
        assert_eq!(
            files[2],
            windows
                .get(&9)
                .unwrap()
                .files()
                .next()
                .unwrap()
                .file_id()
                .file_id()
        );
    }

    #[test]
    fn test_assign_files_to_windows() {
        let files = [
            FileId::random(),
            FileId::random(),
            FileId::random(),
            FileId::random(),
        ];
        let windows = assign_to_windows(
            [
                new_file_handle_with_sequence(files[0], 0, 999, 0, 1),
                new_file_handle_with_sequence(files[1], 0, 999, 0, 1),
                new_file_handle_with_sequence(files[2], 0, 999, 0, 2),
                new_file_handle_with_sequence(files[3], 0, 999, 0, 2),
            ]
            .iter(),
            3,
        );
        assert_eq!(windows.len(), 1);
        let window_files = &windows.get(&0).unwrap().files;
        assert_eq!(4, window_files.len());
        assert_eq!(
            window_files
                .iter()
                .map(|f| f.file_id().file_id())
                .collect::<HashSet<_>>(),
            files.into_iter().collect()
        );
    }

    #[test]
    fn test_assign_compacting_to_windows() {
        let files = [
            new_file_handle(FileId::random(), 0, 999, 0),
            new_file_handle(FileId::random(), 0, 999, 0),
            new_file_handle(FileId::random(), 0, 999, 0),
            new_file_handle(FileId::random(), 0, 999, 0),
            new_file_handle(FileId::random(), 0, 999, 0),
        ];
        files[0].set_compacting(true);
        files[2].set_compacting(true);
        let mut windows = assign_to_windows(files.iter(), 3);
        let window0 = windows.remove(&0).unwrap();
        assert_eq!(3, window0.files.len());
        let candidates = window0
            .files
            .iter()
            .map(|f| f.file_id().file_id())
            .collect::<HashSet<_>>();
        assert_eq!(candidates.len(), 3);
        assert_eq!(
            candidates,
            [
                files[1].file_id().file_id(),
                files[3].file_id().file_id(),
                files[4].file_id().file_id()
            ]
            .into_iter()
            .collect::<HashSet<_>>()
        );
    }

    /// (Window value, overlapping, files' time ranges in window)
    type ExpectedWindowSpec = (i64, bool, Vec<(i64, i64)>);

    fn check_assign_to_windows_with_overlapping(
        file_time_ranges: &[(i64, i64)],
        time_window: i64,
        expected_files: &[ExpectedWindowSpec],
    ) {
        let files: Vec<_> = (0..file_time_ranges.len())
            .map(|_| FileId::random())
            .collect();

        let file_handles = files
            .iter()
            .zip(file_time_ranges.iter())
            .map(|(file_id, range)| new_file_handle(*file_id, range.0, range.1, 0))
            .collect::<Vec<_>>();

        let windows = assign_to_windows(file_handles.iter(), time_window);
        assert_eq!(expected_files.len(), windows.len());

        for (expected_window, overlapping, window_files) in expected_files {
            let actual_window = windows.get(expected_window).unwrap();
            let actual_overlapping = window_has_overlap(actual_window, &windows);
            assert_eq!(*overlapping, actual_overlapping);
            let mut file_ranges = actual_window
                .files
                .iter()
                .map(|f| {
                    let (s, e) = f.time_range();
                    (s.value(), e.value())
                })
                .collect::<Vec<_>>();
            file_ranges.sort_unstable_by(|l, r| l.0.cmp(&r.0).then(l.1.cmp(&r.1)));
            assert_eq!(window_files, &file_ranges);
        }
    }

    #[test]
    fn test_assign_to_windows_with_overlapping() {
        check_assign_to_windows_with_overlapping(
            &[(0, 999), (1000, 1999), (2000, 2999)],
            2,
            &[
                (0, false, vec![(0, 999), (1000, 1999)]),
                (2, false, vec![(2000, 2999)]),
            ],
        );

        check_assign_to_windows_with_overlapping(
            &[(0, 1), (0, 999), (100, 2999)],
            2,
            &[
                (0, true, vec![(0, 1), (0, 999)]),
                (2, true, vec![(100, 2999)]),
            ],
        );

        check_assign_to_windows_with_overlapping(
            &[(0, 999), (1000, 1999), (2000, 2999), (3000, 3999)],
            2,
            &[
                (0, false, vec![(0, 999), (1000, 1999)]),
                (2, false, vec![(2000, 2999), (3000, 3999)]),
            ],
        );

        check_assign_to_windows_with_overlapping(
            &[
                (0, 999),
                (1000, 1999),
                (2000, 2999),
                (3000, 3999),
                (0, 3999),
            ],
            2,
            &[
                (0, true, vec![(0, 999), (1000, 1999)]),
                (2, true, vec![(0, 3999), (2000, 2999), (3000, 3999)]),
            ],
        );

        check_assign_to_windows_with_overlapping(
            &[
                (0, 999),
                (1000, 1999),
                (2000, 2999),
                (3000, 3999),
                (1999, 3999),
            ],
            2,
            &[
                (0, true, vec![(0, 999), (1000, 1999)]),
                (2, true, vec![(1999, 3999), (2000, 2999), (3000, 3999)]),
            ],
        );

        check_assign_to_windows_with_overlapping(
            &[
                (0, 999),     // window 0
                (1000, 1999), // window 0
                (2000, 2999), // window 2
                (3000, 3999), // window 2
                (2999, 3999), // window 2
            ],
            2,
            &[
                // Overlap within a window does not imply cross-window overlap.
                (0, false, vec![(0, 999), (1000, 1999)]),
                (2, false, vec![(2000, 2999), (2999, 3999), (3000, 3999)]),
            ],
        );

        check_assign_to_windows_with_overlapping(
            &[
                (0, 999),     // window 0
                (1000, 1999), // window 0
                (2000, 2999), // window 2
                (3000, 3999), // window 2
                (0, 2000),    // window 2
            ],
            2,
            &[
                (0, true, vec![(0, 999), (1000, 1999)]),
                (2, true, vec![(0, 2000), (2000, 2999), (3000, 3999)]),
            ],
        );
    }

    #[test]
    fn test_assign_to_windows_not_overlapping_when_pk_disjoint() {
        let files = [
            new_file_handle_with_size_sequence_and_primary_key_range(
                FileId::random(),
                0,
                1000,
                0,
                1,
                10,
                pk_range(b"a", b"f"),
            ),
            new_file_handle_with_size_sequence_and_primary_key_range(
                FileId::random(),
                500,
                2999,
                0,
                2,
                10,
                pk_range(b"x", b"z"),
            ),
        ];

        let windows = assign_to_windows(files.iter(), 2);

        assert_eq!(2, windows.len());
        let overlapping = window_has_overlap(windows.get(&0).unwrap(), &windows);
        assert!(!overlapping);
    }

    #[test]
    fn test_assign_to_windows_pk_unknown_in_earlier_window_does_not_poison_later_windows() {
        let files = [
            new_file_handle(FileId::random(), 0, 1999, 0),
            new_file_handle_with_size_sequence_and_primary_key_range(
                FileId::random(),
                2000,
                3999,
                0,
                1,
                10,
                pk_range(b"a", b"f"),
            ),
            new_file_handle_with_size_sequence_and_primary_key_range(
                FileId::random(),
                3000,
                4999,
                0,
                2,
                10,
                pk_range(b"x", b"z"),
            ),
        ];

        let windows = assign_to_windows(files.iter(), 2);

        let overlapping = window_has_overlap(windows.get(&4).unwrap(), &windows);
        assert!(!overlapping);
    }

    struct CompactionPickerTestCase {
        window_size: i64,
        input_files: Vec<FileHandle>,
        expected_outputs: Vec<ExpectedOutput>,
    }

    impl CompactionPickerTestCase {
        async fn check(&self) {
            let file_id_to_idx = self
                .input_files
                .iter()
                .enumerate()
                .map(|(idx, file)| (file.file_id(), idx))
                .collect::<HashMap<_, _>>();
            let windows = assign_to_windows(self.input_files.iter(), self.window_size);
            let active_window =
                find_active_window_by_sequence(self.input_files.iter(), self.window_size).or_else(
                    || find_latest_window_in_seconds(self.input_files.iter(), self.window_size),
                );
            let output = TwcsPicker {
                trigger_file_num: 2,
                active_window_l1_merge_trigger: 8,
                inactive_window_trigger_file_num: 2,
                inactive_window_l1_merge_trigger: 2,
                time_window_seconds: None,
                max_output_file_size: None,
                append_mode: false,
                merge_mode: MergeMode::LastRow,
                max_background_tasks: None,
                time_range: None,
                all_windows_inactive: false,
                metrics: Default::default(),
            }
            .build_output_with_time_range(RegionId::from_u64(0), windows, active_window, None)
            .await
            .unwrap();

            let output = output
                .iter()
                .map(|o| {
                    let input_file_ids = o
                        .inputs
                        .iter()
                        .map(|f| file_id_to_idx.get(&f.file_id()).copied().unwrap())
                        .collect::<HashSet<_>>();
                    (input_file_ids, o.output_level)
                })
                .collect::<Vec<_>>();

            let expected = self
                .expected_outputs
                .iter()
                .map(|o| {
                    let input_file_ids = o.input_files.iter().copied().collect::<HashSet<_>>();
                    (input_file_ids, o.output_level)
                })
                .collect::<Vec<_>>();
            assert_eq!(expected, output);
        }
    }

    struct ExpectedOutput {
        input_files: Vec<usize>,
        output_level: Level,
    }

    #[tokio::test]
    async fn test_newer_windows_are_placed_last_for_execution() {
        let file_ids = (0..4).map(|_| FileId::random()).collect::<Vec<_>>();

        // Case 1: 2 runs found in each time window.
        CompactionPickerTestCase {
            window_size: 3,
            input_files: [
                new_file_handle_with_sequence(file_ids[0], -2000, -3, 0, 1),
                new_file_handle_with_sequence(file_ids[1], -3000, -100, 0, 2),
                new_file_handle_with_sequence(file_ids[2], 0, 2999, 0, 3), //active windows
                new_file_handle_with_sequence(file_ids[3], 50, 2998, 0, 4), //active windows
            ]
            .to_vec(),
            expected_outputs: vec![
                ExpectedOutput {
                    input_files: vec![0, 1],
                    output_level: 1,
                },
                ExpectedOutput {
                    input_files: vec![2, 3],
                    output_level: 1,
                },
            ],
        }
        .check()
        .await;

        // Case 2:
        //    -2000........-3
        // -3000.....-100
        //                    0..............2999
        //                      50..........2998
        //                     11.........2990
        let file_ids = (0..6).map(|_| FileId::random()).collect::<Vec<_>>();
        CompactionPickerTestCase {
            window_size: 3,
            input_files: [
                new_file_handle_with_sequence(file_ids[0], -2000, -3, 0, 1),
                new_file_handle_with_sequence(file_ids[1], -3000, -100, 0, 2),
                new_file_handle_with_sequence(file_ids[2], 0, 2999, 0, 3),
                new_file_handle_with_sequence(file_ids[3], 50, 2998, 0, 4),
                new_file_handle_with_sequence(file_ids[4], 11, 2990, 0, 5),
            ]
            .to_vec(),
            expected_outputs: vec![
                ExpectedOutput {
                    input_files: vec![0, 1],
                    output_level: 1,
                },
                ExpectedOutput {
                    input_files: vec![2, 3, 4],
                    output_level: 1,
                },
            ],
        }
        .check()
        .await;

        // Case 3:
        // A compaction may split output into several files that have overlapping time ranges and
        // the same sequence. They are ordinary compaction candidates now: the picker merges the
        // overlapping ones in the window that reaches the trigger.
        let file_ids = (0..6).map(|_| FileId::random()).collect::<Vec<_>>();
        CompactionPickerTestCase {
            window_size: 3,
            input_files: [
                new_file_handle_with_sequence(file_ids[0], 0, 2999, 1, 1),
                new_file_handle_with_sequence(file_ids[1], 0, 2998, 1, 1),
                new_file_handle_with_sequence(file_ids[2], 3000, 5999, 1, 2),
                new_file_handle_with_sequence(file_ids[3], 3000, 5000, 1, 2),
                new_file_handle_with_sequence(file_ids[4], 11, 2990, 0, 3),
            ]
            .to_vec(),
            expected_outputs: vec![ExpectedOutput {
                input_files: vec![2, 3],
                output_level: 1,
            }],
        }
        .check()
        .await;
    }

    #[tokio::test]
    async fn test_build_output_skips_pk_disjoint_files() {
        let files = [
            new_file_handle_with_size_sequence_and_primary_key_range(
                FileId::random(),
                0,
                2999,
                0,
                1,
                10,
                pk_range(b"a", b"f"),
            ),
            new_file_handle_with_size_sequence_and_primary_key_range(
                FileId::random(),
                50,
                2998,
                0,
                2,
                10,
                pk_range(b"x", b"z"),
            ),
        ];
        let windows = assign_to_windows(files.iter(), 3);
        let active_window = find_latest_window_in_seconds(files.iter(), 3);
        let output = TwcsPicker {
            trigger_file_num: 4,
            active_window_l1_merge_trigger: 8,
            inactive_window_trigger_file_num: 4,
            inactive_window_l1_merge_trigger: 8,
            time_window_seconds: None,
            max_output_file_size: None,
            append_mode: false,
            merge_mode: MergeMode::LastRow,
            max_background_tasks: None,
            time_range: None,
            all_windows_inactive: false,
            metrics: Default::default(),
        }
        .build_output_with_time_range(RegionId::from_u64(0), windows, active_window, None)
        .await
        .unwrap();

        assert!(output.is_empty());
    }

    #[test]
    fn test_append_mode_filter_large_files() {
        let file_ids = (0..4).map(|_| FileId::random()).collect::<Vec<_>>();
        let max_output_file_size = 1000u64;

        // Create files with different sizes
        let small_file_1 = new_file_handle_with_size_and_sequence(file_ids[0], 0, 999, 0, 1, 500);
        let large_file_1 = new_file_handle_with_size_and_sequence(file_ids[1], 0, 999, 0, 2, 1500);
        let small_file_2 = new_file_handle_with_size_and_sequence(file_ids[2], 0, 999, 0, 3, 800);
        let large_file_2 = new_file_handle_with_size_and_sequence(file_ids[3], 0, 999, 0, 4, 2000);

        let mut files_to_merge = vec![small_file_1, large_file_1, small_file_2, large_file_2];

        // Test filtering logic directly
        let original_count = files_to_merge.len();

        // Apply append mode filtering
        files_to_merge.retain(|file| file.size() <= max_output_file_size);

        // Should have filtered out 2 large files, leaving 2 small files
        assert_eq!(files_to_merge.len(), 2);
        assert_eq!(original_count, 4);

        // Verify the remaining files are the small ones
        for file in &files_to_merge {
            assert!(
                file.size() <= max_output_file_size,
                "File size {} should be <= {}",
                file.size(),
                max_output_file_size
            );
        }
    }

    #[tokio::test]
    async fn test_build_output_multiple_windows_with_zero_runs() {
        let file_ids = (0..7).map(|_| FileId::random()).collect::<Vec<_>>();

        let files = [
            // Window 0: Contains 3 files but not forming any runs (not enough files in sequence to reach trigger_file_num)
            new_file_handle_with_sequence(file_ids[0], 0, 999, 0, 1),
            new_file_handle_with_sequence(file_ids[1], 0, 999, 0, 2),
            new_file_handle_with_sequence(file_ids[2], 0, 999, 0, 3),
            // Window 3: Contains files that will form 2 runs
            new_file_handle_with_sequence(file_ids[3], 3000, 3999, 0, 4),
            new_file_handle_with_sequence(file_ids[4], 3000, 3999, 0, 5),
            new_file_handle_with_sequence(file_ids[5], 3000, 3999, 0, 6),
            new_file_handle_with_sequence(file_ids[6], 3000, 3999, 0, 7),
        ];

        let windows = assign_to_windows(files.iter(), 3);

        // Create picker with trigger_file_num of 4 so single files won't form runs in first window
        let picker = TwcsPicker {
            trigger_file_num: 4, // High enough to prevent runs in first window
            active_window_l1_merge_trigger: 8,
            inactive_window_trigger_file_num: 4,
            inactive_window_l1_merge_trigger: 8,
            time_window_seconds: Some(3),
            max_output_file_size: None,
            append_mode: false,
            merge_mode: MergeMode::LastRow,
            max_background_tasks: None,
            time_range: None,
            all_windows_inactive: false,
            metrics: Default::default(),
        };

        let active_window = find_latest_window_in_seconds(files.iter(), 3);
        let output = picker
            .build_output_with_time_range(RegionId::from_u64(123), windows, active_window, None)
            .await
            .unwrap();

        assert!(
            !output.is_empty(),
            "Should have output from windows with runs, even when one window has 0 runs"
        );

        let all_output_files: Vec<_> = output
            .iter()
            .flat_map(|o| o.inputs.iter())
            .map(|f| f.file_id().file_id())
            .collect();

        assert!(
            all_output_files.contains(&file_ids[3])
                || all_output_files.contains(&file_ids[4])
                || all_output_files.contains(&file_ids[5]),
            "Output should contain files from the window with runs"
        );
    }

    #[tokio::test]
    async fn test_build_output_single_window_zero_runs() {
        let file_ids = (0..2).map(|_| FileId::random()).collect::<Vec<_>>();

        let large_file_1 = new_file_handle_with_size_and_sequence(file_ids[0], 0, 999, 0, 1, 2000); // 2000 bytes
        let large_file_2 = new_file_handle_with_size_and_sequence(file_ids[1], 0, 999, 0, 2, 2500); // 2500 bytes

        let files = [large_file_1, large_file_2];

        let windows = assign_to_windows(files.iter(), 3);

        let picker = TwcsPicker {
            trigger_file_num: 2,
            active_window_l1_merge_trigger: 8,
            inactive_window_trigger_file_num: 2,
            inactive_window_l1_merge_trigger: 8,
            time_window_seconds: Some(3),
            max_output_file_size: Some(1000),
            append_mode: true,
            merge_mode: MergeMode::LastRow,
            max_background_tasks: None,
            time_range: None,
            all_windows_inactive: false,
            metrics: Default::default(),
        };

        let active_window = find_latest_window_in_seconds(files.iter(), 3);
        let output = picker
            .build_output_with_time_range(RegionId::from_u64(456), windows, active_window, None)
            .await
            .unwrap();

        // Should return empty output (no compaction needed)
        assert!(
            output.is_empty(),
            "Should return empty output when no runs are found after filtering"
        );
    }

    #[tokio::test]
    async fn test_append_mode_can_pick_remaining_single_level_files() {
        let files = [
            new_file_handle_with_size_and_sequence(FileId::random(), 0, 9, 0, 1, 100),
            new_file_handle_with_size_and_sequence(FileId::random(), 20, 29, 0, 2, 100),
            new_file_handle_with_size_and_sequence(FileId::random(), 40, 49, 0, 3, 100),
            new_file_handle_with_size_and_sequence(FileId::random(), 60, 69, 0, 4, 2_000),
        ];
        let windows = assign_to_windows(files.iter(), 1);
        let picker = TwcsPicker {
            trigger_file_num: 4,
            active_window_l1_merge_trigger: 8,
            inactive_window_trigger_file_num: 4,
            inactive_window_l1_merge_trigger: 8,
            time_window_seconds: Some(1),
            max_output_file_size: Some(1_000),
            append_mode: true,
            merge_mode: MergeMode::LastRow,
            max_background_tasks: None,
            time_range: None,
            all_windows_inactive: false,
            metrics: Default::default(),
        };

        let output = picker
            .build_output_with_time_range(RegionId::from_u64(1), windows, Some(1), None)
            .await
            .unwrap();

        assert_eq!(1, output.len());
        assert_eq!(3, output[0].inputs.len());
    }

    #[tokio::test]
    async fn test_max_background_tasks_truncation() {
        let file_ids = (0..10).map(|_| FileId::random()).collect::<Vec<_>>();
        let max_background_tasks = 3;

        // Create files across multiple windows that will generate multiple compaction outputs
        let files = [
            // Window 0: 4 files that will form a run
            new_file_handle_with_sequence(file_ids[0], 0, 999, 0, 1),
            new_file_handle_with_sequence(file_ids[1], 0, 999, 0, 2),
            new_file_handle_with_sequence(file_ids[2], 0, 999, 0, 3),
            new_file_handle_with_sequence(file_ids[3], 0, 999, 0, 4),
            // Window 3: 4 files that will form another run
            new_file_handle_with_sequence(file_ids[4], 3000, 3999, 0, 5),
            new_file_handle_with_sequence(file_ids[5], 3000, 3999, 0, 6),
            new_file_handle_with_sequence(file_ids[6], 3000, 3999, 0, 7),
            new_file_handle_with_sequence(file_ids[7], 3000, 3999, 0, 8),
            // Window 6: 4 files that will form another run
            new_file_handle_with_sequence(file_ids[8], 6000, 6999, 0, 9),
            new_file_handle_with_sequence(file_ids[9], 6000, 6999, 0, 10),
        ];

        let windows = assign_to_windows(files.iter(), 3);

        let picker = TwcsPicker {
            trigger_file_num: 4,
            active_window_l1_merge_trigger: 8,
            inactive_window_trigger_file_num: 4,
            inactive_window_l1_merge_trigger: 8,
            time_window_seconds: Some(3),
            max_output_file_size: None,
            append_mode: false,
            merge_mode: MergeMode::LastRow,
            max_background_tasks: Some(max_background_tasks),
            time_range: None,
            all_windows_inactive: false,
            metrics: Default::default(),
        };

        let active_window = find_latest_window_in_seconds(files.iter(), 3);
        let output = picker
            .build_output_with_time_range(RegionId::from_u64(123), windows, active_window, None)
            .await
            .unwrap();

        // Should have at most max_background_tasks outputs
        assert!(
            output.len() <= max_background_tasks,
            "Output should be truncated to max_background_tasks: expected <= {}, got {}",
            max_background_tasks,
            output.len()
        );

        // Without max_background_tasks, should have more outputs
        let picker_no_limit = TwcsPicker {
            trigger_file_num: 4,
            active_window_l1_merge_trigger: 8,
            inactive_window_trigger_file_num: 4,
            inactive_window_l1_merge_trigger: 8,
            time_window_seconds: Some(3),
            max_output_file_size: None,
            append_mode: false,
            merge_mode: MergeMode::LastRow,
            max_background_tasks: None,
            time_range: None,
            all_windows_inactive: false,
            metrics: Default::default(),
        };

        let windows_no_limit = assign_to_windows(files.iter(), 3);
        let output_no_limit = picker_no_limit
            .build_output_with_time_range(
                RegionId::from_u64(123),
                windows_no_limit,
                active_window,
                None,
            )
            .await
            .unwrap();

        // Without limit, should have more outputs (if there are enough windows)
        if output_no_limit.len() > max_background_tasks {
            assert!(
                output_no_limit.len() > output.len(),
                "Without limit should have more outputs than with limit"
            );
        }
    }

    #[tokio::test]
    async fn test_max_background_tasks_no_truncation_when_under_limit() {
        let file_ids = (0..4).map(|_| FileId::random()).collect::<Vec<_>>();
        let max_background_tasks = 10; // Larger than expected outputs

        // Create files in one window that will generate one compaction output
        let files = [
            new_file_handle_with_sequence(file_ids[0], 0, 999, 0, 1),
            new_file_handle_with_sequence(file_ids[1], 0, 999, 0, 2),
            new_file_handle_with_sequence(file_ids[2], 0, 999, 0, 3),
            new_file_handle_with_sequence(file_ids[3], 0, 999, 0, 4),
        ];

        let windows = assign_to_windows(files.iter(), 3);

        let picker = TwcsPicker {
            trigger_file_num: 4,
            active_window_l1_merge_trigger: 8,
            inactive_window_trigger_file_num: 4,
            inactive_window_l1_merge_trigger: 8,
            time_window_seconds: Some(3),
            max_output_file_size: None,
            append_mode: false,
            merge_mode: MergeMode::LastRow,
            max_background_tasks: Some(max_background_tasks),
            time_range: None,
            all_windows_inactive: false,
            metrics: Default::default(),
        };

        let active_window = find_latest_window_in_seconds(files.iter(), 3);
        let output = picker
            .build_output_with_time_range(RegionId::from_u64(123), windows, active_window, None)
            .await
            .unwrap();

        // Should have all outputs since we're under the limit
        assert!(
            output.len() <= max_background_tasks,
            "Output should be within limit"
        );
        // Should have at least one output
        assert!(!output.is_empty(), "Should have at least one output");
    }

    #[tokio::test]
    async fn test_pick_multiple_runs() {
        common_telemetry::init_default_ut_logging();

        let num_files = 8;
        let file_ids = (0..num_files).map(|_| FileId::random()).collect::<Vec<_>>();

        // Create files with different sequences so they form multiple runs
        let files: Vec<_> = file_ids
            .iter()
            .enumerate()
            .map(|(idx, file_id)| {
                new_file_handle_with_size_and_sequence(
                    *file_id,
                    0,
                    999,
                    0,
                    (idx + 1) as u64,
                    1024 * 1024,
                )
            })
            .collect();

        let windows = assign_to_windows(files.iter(), 3);

        let picker = TwcsPicker {
            trigger_file_num: 4,
            active_window_l1_merge_trigger: 8,
            inactive_window_trigger_file_num: 4,
            inactive_window_l1_merge_trigger: 8,
            time_window_seconds: Some(3),
            max_output_file_size: None,
            append_mode: false,
            merge_mode: MergeMode::LastRow,
            max_background_tasks: None,
            time_range: None,
            all_windows_inactive: false,
            metrics: Default::default(),
        };

        let active_window = find_latest_window_in_seconds(files.iter(), 3);
        let output = picker
            .build_output_with_time_range(RegionId::from_u64(123), windows, active_window, None)
            .await
            .unwrap();

        assert_eq!(1, output.len());
        assert_eq!(output[0].inputs.len(), num_files);
    }

    #[tokio::test]
    async fn test_window_trigger_can_exceed_input_limit() {
        common_telemetry::init_default_ut_logging();

        let num_files = 50;
        let file_ids = (0..num_files).map(|_| FileId::random()).collect::<Vec<_>>();

        // Create files with different sequences so they form 2 runs
        let files: Vec<_> = file_ids
            .iter()
            .enumerate()
            .map(|(idx, file_id)| {
                new_file_handle_with_size_and_sequence(
                    *file_id,
                    (idx / 2 * 10) as i64,
                    (idx / 2 * 10 + 5) as i64,
                    0,
                    (idx + 1) as u64,
                    1024 * 1024,
                )
            })
            .collect();

        let windows = assign_to_windows(files.iter(), 3);

        let picker = TwcsPicker {
            trigger_file_num: num_files,
            active_window_l1_merge_trigger: 8,
            inactive_window_trigger_file_num: num_files,
            inactive_window_l1_merge_trigger: 8,
            time_window_seconds: Some(3),
            max_output_file_size: None,
            append_mode: false,
            merge_mode: MergeMode::LastRow,
            max_background_tasks: None,
            time_range: None,
            all_windows_inactive: false,
            metrics: Default::default(),
        };

        let active_window = find_latest_window_in_seconds(files.iter(), 3);
        let output = picker
            .build_output_with_time_range(RegionId::from_u64(123), windows, active_window, None)
            .await
            .unwrap();

        assert_eq!(1, output.len());
        assert_eq!(output[0].inputs.len(), num_files.min(*MAX_INPUT_FILES));
    }

    #[tokio::test]
    async fn test_limit_max_input_files_keeps_deletion_markers() {
        common_telemetry::init_default_ut_logging();

        // One large file group spanning the whole window plus enough small ones to fill the
        // input limit. That is exactly 2 runs, so the window allows filtering deletions.
        let mut files = vec![new_file_handle_with_size_and_sequence(
            FileId::random(),
            0,
            3_000_000,
            0,
            1,
            1024 * 1024 * 1024,
        )];
        files.extend((0..DEFAULT_MAX_INPUT_FILES as i64).map(|idx| {
            new_file_handle_with_size_and_sequence(
                FileId::random(),
                (idx + 1) * 10_000,
                (idx + 1) * 10_000 + 1_000,
                0,
                (idx + 2) as u64,
                1024,
            )
        }));

        let windows = assign_to_windows(files.iter(), 3600);

        let picker = TwcsPicker {
            trigger_file_num: 4,
            active_window_l1_merge_trigger: 8,
            inactive_window_trigger_file_num: 4,
            inactive_window_l1_merge_trigger: 8,
            time_window_seconds: Some(3600),
            max_output_file_size: None,
            append_mode: false,
            merge_mode: MergeMode::LastRow,
            max_background_tasks: None,
            time_range: None,
            all_windows_inactive: false,
            metrics: Default::default(),
        };

        let active_window = find_latest_window_in_seconds(files.iter(), 3600);
        let output = picker
            .build_output_with_time_range(RegionId::from_u64(123), windows, active_window, None)
            .await
            .unwrap();

        assert_eq!(1, output.len());
        // The input file num limit picks the smallest groups first, so the large group is
        // left behind while the small groups that overlap it are compacted.
        assert_eq!(DEFAULT_MAX_INPUT_FILES, output[0].inputs.len());
        assert!(
            !output[0].filter_deleted,
            "deletion markers must be kept once the file num limit drops files they may mask"
        );
    }

    #[tokio::test]
    async fn test_limit_max_input_files_still_filters_without_overlap() {
        common_telemetry::init_default_ut_logging();

        // More disjoint file groups than the input limit, i.e. a single run. Nothing left
        // behind can hold a row masked by a deletion marker we compact.
        let files: Vec<_> = (0..DEFAULT_MAX_INPUT_FILES as i64 + 8)
            .map(|idx| {
                new_file_handle_with_size_and_sequence(
                    FileId::random(),
                    (idx + 1) * 10_000,
                    (idx + 1) * 10_000 + 1_000,
                    0,
                    (idx + 1) as u64,
                    1024,
                )
            })
            .collect();

        let windows = assign_to_windows(files.iter(), 3600);

        let picker = TwcsPicker {
            trigger_file_num: 4,
            active_window_l1_merge_trigger: 8,
            inactive_window_trigger_file_num: 4,
            inactive_window_l1_merge_trigger: 8,
            time_window_seconds: Some(3600),
            max_output_file_size: Some(1024 * 1024 * 1024),
            append_mode: false,
            merge_mode: MergeMode::LastRow,
            max_background_tasks: None,
            time_range: None,
            all_windows_inactive: false,
            metrics: Default::default(),
        };

        let active_window = find_latest_window_in_seconds(files.iter(), 3600);
        let output = picker
            .build_output_with_time_range(RegionId::from_u64(123), windows, active_window, None)
            .await
            .unwrap();

        assert_eq!(1, output.len());
        assert_eq!(DEFAULT_MAX_INPUT_FILES, output[0].inputs.len());
        assert!(output[0].filter_deleted);
    }

    #[tokio::test]
    async fn test_newer_windows_have_priority() {
        let older_file_ids = [FileId::random(), FileId::random()];
        let newer_file_ids = [FileId::random(), FileId::random()];
        let files = [
            new_file_handle_with_sequence(older_file_ids[0], 1_000, 1_999, 0, 1),
            new_file_handle_with_sequence(older_file_ids[1], 1_000, 1_999, 0, 2),
            new_file_handle_with_sequence(newer_file_ids[0], 7_000, 7_999, 0, 3),
            new_file_handle_with_sequence(newer_file_ids[1], 7_000, 7_999, 0, 4),
        ];
        let windows = assign_to_windows(files.iter(), 3);
        let picker = TwcsPicker {
            trigger_file_num: 2,
            active_window_l1_merge_trigger: 8,
            inactive_window_trigger_file_num: 2,
            inactive_window_l1_merge_trigger: 8,
            time_window_seconds: Some(3),
            max_output_file_size: None,
            append_mode: false,
            merge_mode: MergeMode::LastRow,
            max_background_tasks: Some(1),
            time_range: None,
            all_windows_inactive: false,
            metrics: Default::default(),
        };

        let output = picker
            .build_output_with_time_range(RegionId::from_u64(123), windows, Some(6), None)
            .await
            .unwrap();

        assert_eq!(1, output.len());
        assert_eq!(
            newer_file_ids.into_iter().collect::<HashSet<_>>(),
            output[0]
                .inputs
                .iter()
                .map(|file| file.file_id().file_id())
                .collect::<HashSet<_>>()
        );
    }

    #[test]
    fn test_filter_time_windows_by_time_range() {
        // Interior, exclusive boundary ends, fractional boundary ends, negative
        // timestamps, and empty ranges must all use the same half-open windows.
        for (start, end, expected_windows) in [
            (1_200, 1_800, vec![0]),
            (0, 3_000, vec![0]),
            (3_000, 6_000, vec![3]),
            (2_999, 3_000, vec![0]),
            (2_999, 3_001, vec![0, 3]),
            (3_000, 3_001, vec![3]),
            (3_001, 3_002, vec![3]),
            (-3_000, 0, vec![-3]),
            (-3_001, -2_999, vec![-6, -3]),
            (0, 0, vec![]),
            (1_500, 1_500, vec![]),
        ] {
            let time_range = TimestampRange::with_unit(start, end, TimeUnit::Millisecond).unwrap();
            let selected = [-6, -3, 0, 3, 6]
                .into_iter()
                .filter(|window| time_window_intersects_range(*window, 3, &time_range))
                .collect::<Vec<_>>();
            assert_eq!(expected_windows, selected, "{time_range}");
        }
    }

    #[test]
    fn test_filter_time_windows_with_unbounded_and_extreme_ranges() {
        for (range, expected_windows) in [
            (TimestampRange::min_to_max(), vec![-3, 0, 3]),
            (
                TimestampRange::from_start(Timestamp::new_second(0)),
                vec![0, 3],
            ),
            (
                TimestampRange::until_end(Timestamp::new_second(0), false),
                vec![-3],
            ),
        ] {
            let selected = [-3, 0, 3]
                .into_iter()
                .filter(|window| time_window_intersects_range(*window, 3, &range))
                .collect::<Vec<_>>();
            assert_eq!(expected_windows, selected, "{range}");
        }

        let upper_range = TimestampRange::new(
            Timestamp::new_second(i64::MAX - 1),
            Timestamp::new_second(i64::MAX),
        )
        .unwrap();
        assert!(!time_window_intersects_range(0, 4, &upper_range));
        assert!(time_window_intersects_range(i64::MAX - 3, 4, &upper_range));

        let lower_range = TimestampRange::until_end(Timestamp::new_second(i64::MIN), false);
        assert!(!time_window_intersects_range(i64::MIN, 3, &lower_range));
    }

    #[tokio::test]
    async fn test_time_range_filter_precedes_background_task_limit() {
        let early_file_ids = [FileId::random(), FileId::random()];
        let selected_file_ids = [FileId::random(), FileId::random()];
        let files = [
            new_file_handle_with_sequence(early_file_ids[0], 1_000, 1_999, 0, 1),
            new_file_handle_with_sequence(early_file_ids[1], 1_000, 1_999, 0, 2),
            new_file_handle_with_sequence(selected_file_ids[0], 7_000, 7_999, 0, 3),
            new_file_handle_with_sequence(selected_file_ids[1], 7_000, 7_999, 0, 4),
        ];
        let windows = assign_to_windows(files.iter(), 3);
        let picker = TwcsPicker {
            trigger_file_num: 2,
            active_window_l1_merge_trigger: 8,
            inactive_window_trigger_file_num: 2,
            inactive_window_l1_merge_trigger: 8,
            time_window_seconds: Some(3),
            max_output_file_size: None,
            append_mode: false,
            merge_mode: MergeMode::LastRow,
            max_background_tasks: Some(1),
            time_range: TimestampRange::new(
                Timestamp::new_millisecond(7_200),
                Timestamp::new_millisecond(7_800),
            ),
            all_windows_inactive: false,
            metrics: Default::default(),
        };

        let output = picker
            .build_output_with_time_range(RegionId::from_u64(123), windows, Some(6), Some(3))
            .await
            .unwrap();

        assert_eq!(1, output.len());
        assert_eq!(
            selected_file_ids.into_iter().collect::<HashSet<_>>(),
            output[0]
                .inputs
                .iter()
                .map(|file| file.file_id().file_id())
                .collect::<HashSet<_>>()
        );
    }

    #[tokio::test]
    async fn test_inactive_window_uses_its_trigger_file_num() {
        let files = [
            new_file_handle_with_size_and_sequence(FileId::random(), 0, 10, 0, 1, 10),
            new_file_handle_with_size_and_sequence(FileId::random(), 20, 30, 0, 2, 10),
        ];
        let windows = assign_to_windows(files.iter(), 100);
        let picker = TwcsPicker {
            trigger_file_num: 4,
            active_window_l1_merge_trigger: 8,
            inactive_window_trigger_file_num: 2,
            inactive_window_l1_merge_trigger: 8,
            time_window_seconds: Some(100),
            max_output_file_size: None,
            append_mode: false,
            merge_mode: MergeMode::LastRow,
            max_background_tasks: None,
            time_range: None,
            all_windows_inactive: false,
            metrics: Default::default(),
        };

        let output = picker
            .build_output_with_time_range(RegionId::from_u64(1), windows, Some(100), None)
            .await
            .unwrap();

        assert_eq!(1, output.len());
        assert_eq!(2, output[0].inputs.len());
    }

    #[tokio::test]
    async fn test_inactive_window_requires_a_level_trigger_before_mixed_fallback() {
        let files = [
            new_file_handle_with_size_and_sequence(FileId::random(), 0, 10, 0, 1, 10),
            new_file_handle_with_size_and_sequence(FileId::random(), 0, 10, 1, 2, 10),
        ];
        let windows = assign_to_windows(files.iter(), 100);
        let picker = TwcsPicker {
            trigger_file_num: 4,
            active_window_l1_merge_trigger: 8,
            inactive_window_trigger_file_num: 8,
            inactive_window_l1_merge_trigger: 2,
            time_window_seconds: Some(100),
            max_output_file_size: Some(1_000),
            append_mode: false,
            merge_mode: MergeMode::LastRow,
            max_background_tasks: None,
            time_range: None,
            all_windows_inactive: false,
            metrics: Default::default(),
        };

        let output = picker
            .build_output_with_time_range(RegionId::from_u64(1), windows, Some(100), None)
            .await
            .unwrap();

        assert!(output.is_empty());
    }

    #[tokio::test]
    async fn test_count_first_prefers_more_files_over_smaller_overlap() {
        let files = [
            new_file_handle_with_size_and_sequence(FileId::random(), 0, 10, 0, 1, 10),
            new_file_handle_with_size_and_sequence(FileId::random(), 20, 30, 0, 2, 10),
            new_file_handle_with_size_and_sequence(FileId::random(), 40, 50, 0, 3, 10),
            new_file_handle_with_size_and_sequence(FileId::random(), 5, 15, 0, 4, 10),
        ];
        let windows = assign_to_windows(files.iter(), 100);
        let picker = TwcsPicker {
            trigger_file_num: 2,
            active_window_l1_merge_trigger: 8,
            inactive_window_trigger_file_num: 2,
            inactive_window_l1_merge_trigger: 8,
            time_window_seconds: Some(100),
            max_output_file_size: None,
            append_mode: false,
            merge_mode: MergeMode::LastRow,
            max_background_tasks: None,
            time_range: None,
            all_windows_inactive: false,
            metrics: Default::default(),
        };

        let output = picker
            .build_output_with_time_range(RegionId::from_u64(1), windows, Some(0), None)
            .await
            .unwrap();

        assert_eq!(1, output.len());
        assert_eq!(4, output[0].inputs.len());
    }

    #[tokio::test]
    async fn test_count_first_trigger_counts_physical_ssts() {
        let files = [
            new_file_handle_with_size_and_sequence(FileId::random(), 0, 10, 0, 1, 10),
            new_file_handle_with_size_and_sequence(FileId::random(), 0, 10, 0, 1, 10),
            new_file_handle_with_size_and_sequence(FileId::random(), 20, 30, 0, 2, 10),
        ];
        let windows = assign_to_windows(files.iter(), 100);
        let picker = TwcsPicker {
            trigger_file_num: 3,
            active_window_l1_merge_trigger: 8,
            inactive_window_trigger_file_num: 3,
            inactive_window_l1_merge_trigger: 8,
            time_window_seconds: Some(100),
            max_output_file_size: None,
            append_mode: false,
            merge_mode: MergeMode::LastRow,
            max_background_tasks: None,
            time_range: None,
            all_windows_inactive: false,
            metrics: Default::default(),
        };

        let output = picker
            .build_output_with_time_range(RegionId::from_u64(1), windows, Some(0), None)
            .await
            .unwrap();

        assert_eq!(1, output.len());
        assert_eq!(3, output[0].inputs.len());
    }

    #[tokio::test]
    async fn test_count_first_does_not_compact_overlap_below_trigger() {
        let files = [
            new_file_handle_with_size_and_sequence(FileId::random(), 0, 20, 0, 1, 10),
            new_file_handle_with_size_and_sequence(FileId::random(), 10, 30, 0, 2, 10),
        ];
        let windows = assign_to_windows(files.iter(), 100);
        let picker = TwcsPicker {
            trigger_file_num: 3,
            active_window_l1_merge_trigger: 8,
            inactive_window_trigger_file_num: 3,
            inactive_window_l1_merge_trigger: 8,
            time_window_seconds: Some(100),
            max_output_file_size: None,
            append_mode: false,
            merge_mode: MergeMode::LastRow,
            max_background_tasks: None,
            time_range: None,
            all_windows_inactive: false,
            metrics: Default::default(),
        };

        let output = picker
            .build_output_with_time_range(RegionId::from_u64(1), windows, Some(0), None)
            .await
            .unwrap();

        assert!(output.is_empty());
    }

    #[tokio::test]
    async fn test_filter_deleted_is_false_when_selected_files_overlap_unselected_file() {
        let mut files = (0..32)
            .map(|idx| {
                new_file_handle_with_size_and_sequence(
                    FileId::random(),
                    idx * 10,
                    idx * 10 + 9,
                    0,
                    idx as u64 + 1,
                    10,
                )
            })
            .collect::<Vec<_>>();
        files.push(new_file_handle_with_size_and_sequence(
            FileId::random(),
            0,
            320,
            0,
            33,
            10,
        ));
        let windows = assign_to_windows(files.iter(), 1000);
        let picker = TwcsPicker {
            trigger_file_num: 2,
            active_window_l1_merge_trigger: 8,
            inactive_window_trigger_file_num: 2,
            inactive_window_l1_merge_trigger: 8,
            time_window_seconds: Some(1000),
            max_output_file_size: None,
            append_mode: false,
            merge_mode: MergeMode::LastRow,
            max_background_tasks: None,
            time_range: None,
            all_windows_inactive: false,
            metrics: Default::default(),
        };

        let output = picker
            .build_output_with_time_range(RegionId::from_u64(1), windows, Some(0), None)
            .await
            .unwrap();

        assert_eq!(1, output.len());
        assert_eq!(DEFAULT_MAX_INPUT_FILES, output[0].inputs.len());
        assert!(!output[0].filter_deleted);
    }

    #[rstest::rstest]
    #[case::historical_l1(0, PickClass::FileReduction, 1)]
    #[case::same_window_l1(10, PickClass::FileReduction, 1)]
    #[case::older_l0(0, PickClass::L0Merge, 0)]
    #[case::historical_overlap(0, PickClass::OverlapResolution, 1)]
    #[tokio::test]
    async fn test_fair_picker_promotes_waiting_work_with_one_slot(
        #[case] waiting_window: i64,
        #[case] class: PickClass,
        #[case] level: Level,
    ) {
        let overlap = class == PickClass::OverlapResolution;
        let files = priority_test_files(10, 0, 40, false)
            .into_iter()
            .chain(priority_test_files(
                waiting_window,
                level,
                if overlap { 100 } else { 40 },
                overlap,
            ))
            .collect::<Vec<_>>();
        let mut history = PickHistory::default();
        let picker = priority_test_picker(1);
        let first = picker
            .build_output_with_history(assign_to_windows(files.iter(), 1), Some(10), &mut history)
            .await
            .unwrap();
        assert_eq!(1, first.len());
        assert!(first[0].inputs.iter().all(|f| f.level() == 0));
        assert_eq!(10, output_window(&first[0]));

        // Advance the waiting candidate's age without sleeping. Discovery must
        // preserve it even while the hot window keeps supplying eligible L0.
        let key = PickKey {
            window: waiting_window,
            class,
        };
        history = PickHistory::default();
        history.observe_for_test([key].into_iter(), Instant::now() - Duration::from_secs(120));
        let output = picker
            .build_output_with_history(assign_to_windows(files.iter(), 1), Some(10), &mut history)
            .await
            .unwrap();
        assert_eq!(1, output.len());
        assert_eq!(waiting_window, output_window(&output[0]));
        assert!(output[0].inputs.iter().all(|f| f.level() == level));
        let expected: HashSet<_> = files
            .iter()
            .filter(|f| f.level() == level && f.time_range().1.value() / 1000 == waiting_window)
            .map(FileHandle::file_id)
            .collect();
        assert_eq!(
            expected,
            output[0].inputs.iter().map(FileHandle::file_id).collect()
        );
    }

    #[tokio::test]
    async fn test_overdue_candidates_alternate_without_completion() {
        // A plan that is accepted but never completes must still consume the
        // fairness turn. Otherwise an older candidate that keeps failing keeps
        // the first slot and starves every other eligible window.
        let files = priority_test_files(0, 0, 40, false)
            .into_iter()
            .chain(priority_test_files(10, 0, 40, false))
            .collect::<Vec<_>>();
        let mut region = compaction_region_with_ssts(
            files.iter().map(|file| file.meta_ref().clone()),
            Duration::from_secs(60),
        )
        .await;
        region.ttl = None;
        let crate::region::options::CompactionOptions::Twcs(options) =
            &mut region.region_options.compaction;
        options.time_window = Some(Duration::from_secs(1));
        options.active_window_trigger_file_num = 2;
        let mut history = PickHistory::default();
        let start = Instant::now();
        history.seed_for_test(
            PickKey {
                window: 0,
                class: PickClass::L0Merge,
            },
            start - Duration::from_secs(120),
        );
        let younger = PickKey {
            window: 10,
            class: PickClass::L0Merge,
        };
        let waiting_since = start - Duration::from_secs(90);
        history.seed_for_test(younger, waiting_since);

        // Exercise the production orchestration, not the test helper's copy of
        // rank/truncate/acknowledge: only the selected key may consume a turn.
        let first = pick_automatic_for_test(&region, &mut history, Some(1))
            .await
            .unwrap()
            .unwrap();
        assert_eq!(1, first.outputs.len());
        assert_eq!(0, output_window(&first.outputs[0]));
        assert_eq!(
            Some(waiting_since),
            history.promoted_since_for_test(younger, start)
        );

        // No completion acknowledgement: selecting the older window already
        // consumed its turn, so the younger overdue window must go next.
        let second = pick_automatic_for_test(&region, &mut history, Some(1))
            .await
            .unwrap()
            .unwrap();
        assert_eq!(1, second.outputs.len());
        assert_eq!(10, output_window(&second.outputs[0]));
    }

    #[rstest::rstest]
    #[case::stateless_early_exit(false, false, true)]
    #[case::fair_full_scan(true, false, true)]
    #[case::promoted_selection(true, true, true)]
    #[case::empty_plan(true, false, false)]
    #[tokio::test]
    async fn test_pick_metrics_cover_scanned_windows_and_selected_seeds(
        #[case] fair: bool,
        #[case] promoted: bool,
        #[case] eligible: bool,
    ) {
        let mut files = priority_test_files(10, 0, 40, false).to_vec();
        files.extend(priority_test_files(10, 1, 40, false));
        files.extend(priority_test_files(0, 1, 40, false));
        if !eligible {
            files = vec![files[0].clone(), files[4].clone()];
        }
        // Unregistered instances cannot be changed by other libtest threads.
        let selected_metrics = IntCounterVec::new(
            prometheus::Opts::new("test_twcs_selected", "selected seeds"),
            &["class", "priority"],
        )
        .unwrap();
        let picker = TwcsPicker {
            metrics: TwcsPickMetrics {
                elapsed: HistogramVec::new(
                    prometheus::HistogramOpts::new("test_twcs_pick_elapsed", "pick elapsed"),
                    &["mode"],
                )
                .unwrap(),
                scanned_windows: IntCounterVec::new(
                    prometheus::Opts::new("test_twcs_scanned_windows", "scanned windows"),
                    &["mode"],
                )
                .unwrap(),
            },
            ..priority_test_picker(1)
        };
        let mut history = PickHistory::with_selected_metrics_for_test(selected_metrics.clone());
        if fair && promoted {
            history.observe_for_test(
                [PickKey {
                    window: 0,
                    class: PickClass::FileReduction,
                }]
                .into_iter(),
                Instant::now() - Duration::from_secs(120),
            );
        }
        let mode = if fair { "fair" } else { "stateless" };
        let elapsed = picker.metrics.elapsed.with_label_values(&[mode]);
        let windows = picker.metrics.scanned_windows.with_label_values(&[mode]);
        let elapsed_before = elapsed.get_sample_count();
        let windows_before = windows.get();
        let selected = [
            PickClass::L0Merge,
            PickClass::FileReduction,
            PickClass::OverlapResolution,
        ]
        .into_iter()
        .flat_map(|class| {
            let selected_metrics = &selected_metrics;
            ["normal", "promoted"].into_iter().map(move |priority| {
                let counter = selected_metrics.with_label_values(&[class.as_str(), priority]);
                let before = counter.get();
                (class, priority, counter, before)
            })
        })
        .collect::<Vec<_>>();

        let output = if fair {
            picker
                .build_output_with_history(
                    assign_to_windows(files.iter(), 1),
                    Some(10),
                    &mut history,
                )
                .await
                .unwrap()
        } else {
            picker
                .build_output_with_time_range(
                    RegionId::from_u64(1),
                    assign_to_windows(files.iter(), 1),
                    Some(10),
                    None,
                )
                .await
                .unwrap()
        };
        assert_eq!(usize::from(eligible), output.len());
        if eligible {
            assert_eq!(if promoted { 0 } else { 10 }, output_window(&output[0]));
        }
        assert_eq!(elapsed_before + 1, elapsed.get_sample_count());
        assert_eq!(windows_before + if fair { 2 } else { 1 }, windows.get());
        // Count only the single selected seed, not the other discovered classes
        // or the older window discarded by the output limit. Stateless picks
        // never contribute to automatic selection/promotion metrics.
        for (class, priority, counter, before) in selected {
            let chosen = if promoted {
                (PickClass::FileReduction, "promoted")
            } else {
                (PickClass::L0Merge, "normal")
            };
            let expected = u64::from(fair && eligible && (class, priority) == chosen);
            assert_eq!(before + expected, counter.get(), "{class:?}/{priority}");
        }
    }

    #[tokio::test]
    async fn test_fair_inactive_closed_merge_absorbs_stranded_l0() {
        // R2-C1: one L0 cannot merge alone, and the mixed L0 fallback exceeds
        // its one-output budget. A closed overlap merge must still see both levels.
        let files = [
            new_file_handle_with_size_and_sequence(FileId::random(), 0, 499, 0, 1, 100),
            new_file_handle_with_size_and_sequence(FileId::random(), 250, 749, 1, 1, 100),
        ];
        let picker = priority_test_picker(1);
        let mut history = PickHistory::default();
        let output = picker
            .build_output_with_history(assign_to_windows(files.iter(), 1), Some(10), &mut history)
            .await
            .unwrap();
        assert_eq!(1, output.len());
        assert_eq!(
            files
                .iter()
                .map(FileHandle::file_id)
                .collect::<HashSet<_>>(),
            output[0].inputs.iter().map(FileHandle::file_id).collect()
        );
    }

    #[rstest::rstest]
    #[case::stranded_l0(1)]
    #[case::eligible_l0_merge(2)]
    #[tokio::test]
    async fn test_fair_closed_merge_does_not_add_conflicts_with_l0(#[case] l0_count: usize) {
        // R2-C1: merging the two L1 files alone encloses previously disjoint L0
        // peers. Even an aged FileReduction must not displace safe L0 work with
        // such a merge; without L0 work this snapshot must produce no plan.
        let mut files = vec![
            new_file_handle_with_size_and_sequence(FileId::random(), 0, 99, 1, 1, 40),
            new_file_handle_with_size_and_sequence(FileId::random(), 200, 299, 1, 1, 40),
        ];
        files.extend((0..l0_count).map(|_| {
            new_file_handle_with_size_and_sequence(FileId::random(), 100, 199, 0, 1, 1000)
        }));
        let mut history = PickHistory::default();
        history.observe_for_test(
            [PickKey {
                window: 0,
                class: PickClass::FileReduction,
            }]
            .into_iter(),
            Instant::now() - Duration::from_secs(120),
        );
        let picker = priority_test_picker(1);
        let output = picker
            .build_output_with_history(assign_to_windows(files.iter(), 1), Some(10), &mut history)
            .await
            .unwrap();
        if l0_count == 1 {
            assert!(output.is_empty());
        } else {
            assert_eq!(1, output.len());
            assert_eq!(
                files[2..]
                    .iter()
                    .map(FileHandle::file_id)
                    .collect::<HashSet<_>>(),
                output[0].inputs.iter().map(FileHandle::file_id).collect()
            );
        }
    }

    #[rstest::rstest]
    #[case::active(0, MergeMode::LastRow, "active:pick_l1_file_reduction")]
    #[case::inactive(10, MergeMode::LastRow, "inactive:pick_l1_file_reduction")]
    #[case::settling_fallback(1, MergeMode::LastRow, "inactive:pick_small_l1")]
    #[case::settling_last_non_null(1, MergeMode::LastNonNull, "")]
    fn test_window_discovery_keeps_l1_alongside_l0(
        #[case] active_window: i64,
        #[case] merge_mode: MergeMode,
        #[case] l1_branch: &str,
    ) {
        let files = priority_test_files(0, 0, 40, false)
            .into_iter()
            .chain(priority_test_files(0, 1, 20, true))
            .collect::<Vec<_>>();
        let windows = assign_to_windows(files.iter(), 1);
        let mapper = primary_key_mapper_for_test();
        let mut context = WindowPickContext {
            all_candidates: false,
            region_id: RegionId::from_u64(1),
            active_window: Some(active_window),
            time_window_size: Some(1),
            region_sequence: 100,
            mapper: &mapper,
        };
        let mut picker = priority_test_picker(4);
        picker.merge_mode = merge_mode;
        let legacy = picker.pick_window(&windows[&0], &context);
        assert_eq!(1, legacy.len());
        assert_eq!(PickClass::L0Merge, legacy[0].class);

        context.all_candidates = true;
        let picks = picker.pick_window(&windows[&0], &context);
        assert_eq!(if l1_branch.is_empty() { 1 } else { 2 }, picks.len());
        assert_eq!(PickClass::L0Merge, picks[0].class);
        if !l1_branch.is_empty() {
            assert_eq!(PickClass::FileReduction, picks[1].class);
            assert_eq!(l1_branch, picks[1].branch);
            assert_eq!(
                files
                    .iter()
                    // Inactive closed merges include overlapping L0 peers;
                    // active picks and the small-L1 fallback remain L1-only.
                    .filter(|file| active_window == 10 || file.level() == 1)
                    .map(FileHandle::file_id)
                    .collect::<HashSet<_>>(),
                picks[1]
                    .inputs
                    .iter()
                    .map(FileHandle::file_id)
                    .collect::<HashSet<_>>()
            );
        }
    }

    #[tokio::test]
    async fn test_fair_picker_does_not_schedule_same_window_alternatives_together() {
        let files = priority_test_files(0, 0, 40, false)
            .into_iter()
            .chain(priority_test_files(0, 1, 40, false))
            .collect::<Vec<_>>();
        let picker = priority_test_picker(4);
        let mut history = PickHistory::default();
        let output = picker
            .build_output_with_history(assign_to_windows(files.iter(), 1), Some(0), &mut history)
            .await
            .unwrap();
        assert_eq!(1, output.len());
        assert!(output[0].inputs.iter().all(|f| f.level() == 0));
        assert!(
            !output[0].filter_deleted,
            "unselected overlapping L1 must retain tombstones"
        );
    }

    fn new_file(start: i64, end: i64, sequence: u64, file_size: u64) -> FileHandle {
        new_file_handle_with_size_and_sequence(FileId::random(), start, end, 0, sequence, file_size)
    }

    fn new_file_with_level_and_rows(
        start: i64,
        end: i64,
        level: Level,
        sequence: u64,
        file_size: u64,
        num_rows: u64,
    ) -> FileHandle {
        let file = new_file_handle_with_size_and_sequence(
            FileId::random(),
            start,
            end,
            level,
            sequence,
            file_size,
        );
        let mut meta = file.meta_ref().clone();
        meta.num_rows = num_rows;
        FileHandle::new(meta, crate::test_util::new_noop_file_purger())
    }

    fn picked_ranges(files: &[FileHandle]) -> Vec<(i64, i64)> {
        files
            .iter()
            .map(|file| {
                let (start, end) = file.range();
                (start.value(), end.value())
            })
            .collect()
    }

    #[test]
    fn test_count_first_rejects_dominant_historical_file() {
        let picked = pick_count_first(
            vec![SortedRun::from(vec![
                new_file(0, 9, 1, 400),
                new_file(20, 29, 2, 100),
            ])],
            None,
        );

        assert!(picked.is_empty());
    }

    #[test]
    fn test_count_first_accepts_balanced_historical_file() {
        let picked = pick_count_first(
            vec![SortedRun::from(vec![
                new_file(0, 9, 1, 400),
                new_file(20, 29, 2, 100),
                new_file(40, 49, 3, 100),
                new_file(60, 69, 4, 100),
                new_file(80, 89, 5, 100),
            ])],
            None,
        );

        assert_eq!(
            vec![(0, 9), (20, 29), (40, 49), (60, 69), (80, 89)],
            picked_ranges(&picked)
        );
    }

    #[test]
    fn test_count_first_finds_smaller_balanced_interval_when_larger_one_is_unbalanced() {
        let picked = pick_count_first(
            vec![SortedRun::from(vec![
                new_file(0, 9, 1, 1000),
                new_file(20, 29, 2, 100),
                new_file(40, 49, 3, 100),
                new_file(60, 69, 4, 100),
                new_file(80, 89, 5, 100),
            ])],
            None,
        );

        assert_eq!(
            vec![(20, 29), (40, 49), (60, 69), (80, 89)],
            picked_ranges(&picked)
        );
    }

    #[test]
    fn test_count_first_prefers_overlap_participants_when_file_counts_match() {
        let first_run = (0..DEFAULT_MAX_INPUT_FILES)
            .map(|idx| {
                let start = idx as i64 * 20;
                let end = if idx + 1 == DEFAULT_MAX_INPUT_FILES {
                    700
                } else {
                    start + 9
                };
                new_file(start, end, idx as u64 + 1, 10)
            })
            .collect::<Vec<_>>();
        let overlapping = new_file(690, 710, 100, 10);

        let picked = pick_count_first(
            vec![
                SortedRun::from(first_run),
                SortedRun::from(vec![overlapping]),
            ],
            None,
        );
        let ranges = picked_ranges(&picked);

        assert_eq!(DEFAULT_MAX_INPUT_FILES, ranges.len());
        assert!(!ranges.contains(&(0, 9)));
        assert!(ranges.contains(&(690, 710)));
    }

    #[tokio::test]
    async fn test_picker_avoids_chained_l1_rewrites_by_compacting_levels_separately() {
        let mut enough_l0 = (0..DEFAULT_MAX_INPUT_FILES)
            .map(|idx| {
                let start = idx as i64 * 20;
                new_file_handle_with_size_and_sequence(
                    FileId::random(),
                    start,
                    start + 9,
                    0,
                    idx as u64 + 1,
                    10,
                )
            })
            .collect::<Vec<_>>();
        enough_l0.push(new_file_handle_with_size_and_sequence(
            FileId::random(),
            0,
            700,
            1,
            100,
            100,
        ));
        let mut enough_l1 = (0..4)
            .map(|idx| {
                new_file_handle_with_size_and_sequence(
                    FileId::random(),
                    idx * 100,
                    idx * 100 + 99,
                    1,
                    idx as u64 + 1,
                    100,
                )
            })
            .collect::<Vec<_>>();
        enough_l1.extend((0..3).map(|idx| {
            new_file_handle_with_size_and_sequence(
                FileId::random(),
                idx * 20,
                idx * 20 + 9,
                0,
                idx as u64 + 10,
                10,
            )
        }));
        let picker = TwcsPicker {
            trigger_file_num: 4,
            active_window_l1_merge_trigger: 8,
            inactive_window_trigger_file_num: 4,
            inactive_window_l1_merge_trigger: 4,
            time_window_seconds: Some(1),
            max_output_file_size: None,
            append_mode: false,
            merge_mode: MergeMode::LastRow,
            max_background_tasks: None,
            time_range: None,
            all_windows_inactive: false,
            metrics: Default::default(),
        };

        for (case, files, expected_level, expected_len) in [
            ("L0 reaches trigger", enough_l0, 0, DEFAULT_MAX_INPUT_FILES),
            ("L0 fallback precedes triggered L1", enough_l1, 0, 3),
        ] {
            let windows = assign_to_windows(files.iter(), 1);
            let output = picker
                .build_output_with_time_range(RegionId::from_u64(1), windows, Some(1), None)
                .await
                .unwrap();

            assert_eq!(1, output.len(), "{case}");
            assert_eq!(expected_len, output[0].inputs.len(), "{case}");
            assert!(
                output[0]
                    .inputs
                    .iter()
                    .all(|file| file.level() == expected_level),
                "{case}"
            );
        }
    }

    #[tokio::test]
    async fn test_active_window_defers_l1_below_safety_trigger() {
        let files = (0..7)
            .map(|idx| {
                let start = idx * 100;
                new_file_handle_with_size_and_sequence(
                    FileId::random(),
                    start,
                    start + 99,
                    1,
                    idx as u64 + 1,
                    100,
                )
            })
            .collect::<Vec<_>>();
        let windows = assign_to_windows(files.iter(), 1);
        let active_window = windows.keys().next().copied();
        let picker = TwcsPicker {
            trigger_file_num: 4,
            active_window_l1_merge_trigger: 8,
            inactive_window_trigger_file_num: 2,
            inactive_window_l1_merge_trigger: 8,
            time_window_seconds: Some(1),
            max_output_file_size: None,
            append_mode: false,
            merge_mode: MergeMode::LastRow,
            max_background_tasks: None,
            time_range: None,
            all_windows_inactive: false,
            metrics: Default::default(),
        };

        let output = picker
            .build_output_with_time_range(RegionId::from_u64(1), windows, active_window, None)
            .await
            .unwrap();

        assert!(output.is_empty());
    }

    #[tokio::test]
    async fn test_active_window_l1_safety_trigger_is_independent_of_l0_trigger() {
        let files = (0..8)
            .map(|idx| {
                let start = idx * 100;
                new_file_handle_with_size_and_sequence(
                    FileId::random(),
                    start,
                    start + 99,
                    1,
                    idx as u64 + 1,
                    100,
                )
            })
            .collect::<Vec<_>>();
        let windows = assign_to_windows(files.iter(), 1);
        let active_window = windows.keys().next().copied();
        let picker = TwcsPicker {
            trigger_file_num: 16,
            active_window_l1_merge_trigger: 8,
            inactive_window_trigger_file_num: 2,
            inactive_window_l1_merge_trigger: 8,
            time_window_seconds: Some(1),
            max_output_file_size: None,
            append_mode: false,
            merge_mode: MergeMode::LastRow,
            max_background_tasks: None,
            time_range: None,
            all_windows_inactive: false,
            metrics: Default::default(),
        };

        let output = picker
            .build_output_with_time_range(RegionId::from_u64(1), windows, active_window, None)
            .await
            .unwrap();

        assert_eq!(1, output.len());
        assert_eq!(8, output[0].inputs.len());
        assert!(output[0].inputs.iter().all(|file| file.level() == 1));
    }

    #[tokio::test]
    async fn test_picker_falls_back_to_l1_when_triggered_l0_cannot_make_progress() {
        let mut files = (0..4)
            .map(|idx| {
                let start = idx * 20;
                new_file_handle_with_size_and_sequence(
                    FileId::random(),
                    start,
                    start + 9,
                    0,
                    idx as u64 + 1,
                    600,
                )
            })
            .collect::<Vec<_>>();
        files.extend((0..8).map(|idx| {
            let start = idx * 20 + 100;
            new_file_handle_with_size_and_sequence(
                FileId::random(),
                start,
                start + 9,
                1,
                idx as u64 + 10,
                50,
            )
        }));
        let windows = assign_to_windows(files.iter(), 1);
        let active_window = windows.keys().next().copied();
        let picker = TwcsPicker {
            trigger_file_num: 4,
            active_window_l1_merge_trigger: 8,
            inactive_window_trigger_file_num: 4,
            inactive_window_l1_merge_trigger: 8,
            time_window_seconds: Some(1),
            max_output_file_size: Some(512),
            append_mode: false,
            merge_mode: MergeMode::LastRow,
            max_background_tasks: None,
            time_range: None,
            all_windows_inactive: false,
            metrics: Default::default(),
        };

        let output = picker
            .build_output_with_time_range(RegionId::from_u64(1), windows, active_window, None)
            .await
            .unwrap();

        assert_eq!(1, output.len());
        assert_eq!(8, output[0].inputs.len());
        assert!(output[0].inputs.iter().all(|file| file.level() == 1));
    }

    fn priority_test_files(
        window: i64,
        level: Level,
        file_size: u64,
        overlap: bool,
    ) -> [FileHandle; 2] {
        let start = window * 1_000;
        let ranges = if overlap {
            [(start, start + 499), (start + 250, start + 749)]
        } else {
            [(start, start + 99), (start + 200, start + 299)]
        };
        ranges.map(|(start, end)| {
            new_file_handle_with_size_and_sequence(
                FileId::random(),
                start,
                end,
                level,
                window as u64 + 1,
                file_size,
            )
        })
    }

    fn priority_test_picker(max_background_tasks: usize) -> TwcsPicker {
        TwcsPicker {
            trigger_file_num: 2,
            active_window_l1_merge_trigger: 2,
            inactive_window_trigger_file_num: 2,
            inactive_window_l1_merge_trigger: 2,
            time_window_seconds: Some(1),
            max_output_file_size: Some(100),
            append_mode: false,
            merge_mode: MergeMode::LastRow,
            max_background_tasks: Some(max_background_tasks),
            time_range: None,
            all_windows_inactive: false,
            metrics: Default::default(),
        }
    }

    fn output_window(output: &CompactionOutput) -> i64 {
        output.inputs[0]
            .time_range()
            .1
            .convert_to(TimeUnit::Second)
            .unwrap()
            .value()
    }

    #[tokio::test]
    async fn test_manual_compaction_resolves_active_window_overlaps() {
        let files = priority_test_files(5, 1, 100, true);
        for (all_windows_inactive, expected) in [(false, 0), (true, 1)] {
            let picker = TwcsPicker {
                all_windows_inactive,
                ..priority_test_picker(1)
            };
            let output = picker
                .build_output_with_time_range(
                    RegionId::from_u64(1),
                    assign_to_windows(files.iter(), 1),
                    Some(5),
                    None,
                )
                .await
                .unwrap();
            assert_eq!(expected, output.len(), "manual: {all_windows_inactive}");
        }
    }

    #[tokio::test]
    async fn test_shared_boundary_l1_files_resolve_in_inactive_and_manual_windows() {
        let files = [(b"a", b"m"), (b"m", b"z")].map(|(min, max)| {
            new_file_handle_with_size_sequence_and_primary_key_range(
                FileId::random(),
                0,
                100,
                1,
                1,
                100,
                pk_range(min, max),
            )
        });
        for (active_window, all_windows_inactive, expected) in
            [(0, false, 0), (0, true, 1), (3, false, 1)]
        {
            let picker = TwcsPicker {
                all_windows_inactive,
                ..priority_test_picker(1)
            };
            let output = picker
                .build_output_with_time_range(
                    RegionId::from_u64(1),
                    assign_to_windows(files.iter(), 1),
                    Some(active_window),
                    Some(1),
                )
                .await
                .unwrap();
            assert_eq!(
                expected,
                output.len(),
                "active: {active_window}, manual: {all_windows_inactive}"
            );
            if let Some(output) = output.first() {
                assert_eq!(2, output.inputs.len());
            }
        }
    }

    #[tokio::test]
    async fn test_settling_window_defers_overlap_resolution() {
        let files = priority_test_files(3, 1, 100, true);
        for (active_window, all_windows_inactive, expected) in
            [(4, false, 0), (5, false, 1), (2, false, 1), (4, true, 1)]
        {
            let picker = TwcsPicker {
                all_windows_inactive,
                ..priority_test_picker(1)
            };
            let output = picker
                .build_output_with_time_range(
                    RegionId::from_u64(1),
                    assign_to_windows(files.iter(), 1),
                    Some(active_window),
                    Some(1),
                )
                .await
                .unwrap();
            assert_eq!(
                expected,
                output.len(),
                "active: {active_window}, manual: {all_windows_inactive}"
            );
        }
    }

    #[tokio::test]
    async fn test_inactive_windows_merge_small_l1_without_large_peer() {
        let large =
            new_file_handle_with_size_and_sequence(FileId::random(), 3_000, 3_999, 1, 1, 1_000);
        let small = priority_test_files(3, 1, 20, true);
        let files = [vec![large.clone()], small.to_vec()].concat();
        // The immediate predecessor is settling; the older window is not quiet.
        for (active_window, merge_mode, trigger, expected) in [
            (4, MergeMode::LastRow, 2, 1),
            (5, MergeMode::LastRow, 2, 1),
            (4, MergeMode::LastNonNull, 2, 0),
            (5, MergeMode::LastNonNull, 2, 0),
            (4, MergeMode::LastRow, 4, 0),
        ] {
            let picker = TwcsPicker {
                merge_mode,
                inactive_window_l1_merge_trigger: trigger,
                ..priority_test_picker(1)
            };
            let output = picker
                .build_output_with_time_range(
                    RegionId::from_u64(1),
                    assign_to_windows(files.iter(), 1),
                    Some(active_window),
                    Some(1),
                )
                .await
                .unwrap();
            assert_eq!(
                expected,
                output.len(),
                "active: {active_window}, mode: {merge_mode:?}, trigger: {trigger}"
            );
            if expected == 0 {
                continue;
            }
            let inputs: HashSet<_> = output[0].inputs.iter().map(FileHandle::file_id).collect();
            assert_eq!(
                small
                    .iter()
                    .map(FileHandle::file_id)
                    .collect::<HashSet<_>>(),
                inputs
            );
            assert!(
                !output[0].filter_deleted,
                "the unselected large file still overlaps"
            );
        }
    }

    #[tokio::test]
    async fn test_small_l1_fallback_checks_large_append_mode_peer() {
        // Two conflicting small files form a non-rectangular area. Their union
        // would introduce a conflict with the large file in the empty corner.
        let files = [
            (0, 20, b"a", b"m", 20),
            (10, 30, b"l", b"z", 20),
            (0, 5, b"y", b"z", 1_000),
        ]
        .map(|(start, end, min, max, size)| {
            new_file_handle_with_size_sequence_and_primary_key_range(
                FileId::random(),
                start,
                end,
                1,
                1,
                size,
                pk_range(min, max),
            )
        });
        for append_mode in [false, true] {
            let picker = TwcsPicker {
                append_mode,
                ..priority_test_picker(1)
            };
            let output = picker
                .build_output_with_time_range(
                    RegionId::from_u64(1),
                    assign_to_windows(files.iter(), 1),
                    Some(1),
                    Some(1),
                )
                .await
                .unwrap();
            assert!(output.is_empty(), "append mode: {append_mode}");
        }
    }

    #[tokio::test]
    async fn test_settling_window_does_not_rewrite_l1_for_late_data() {
        // A late L0 file overlapping both L1 files, which a mixed merge absorbs.
        let mut files = priority_test_files(3, 1, 40, false).to_vec();
        files.push(new_file_handle_with_size_and_sequence(
            FileId::random(),
            3_000,
            3_299,
            0,
            10,
            10,
        ));
        for (active_window, expected) in [(4, 0), (5, 1)] {
            let output = priority_test_picker(1)
                .build_output_with_time_range(
                    RegionId::from_u64(1),
                    assign_to_windows(files.iter(), 1),
                    Some(active_window),
                    Some(1),
                )
                .await
                .unwrap();
            assert_eq!(expected, output.len(), "active: {active_window}");
        }
    }

    #[tokio::test]
    async fn test_older_l0_candidate_has_priority_over_newer_pure_l1_candidate() {
        let files = priority_test_files(0, 0, 40, false)
            .into_iter()
            .chain(priority_test_files(1, 1, 40, false))
            .collect::<Vec<_>>();
        let windows = assign_to_windows(files.iter(), 1);

        let output = priority_test_picker(1)
            .build_output_with_time_range(RegionId::from_u64(1), windows, Some(1), None)
            .await
            .unwrap();

        assert_eq!(1, output.len());
        assert_eq!(0, output_window(&output[0]));
        assert!(output[0].inputs.iter().all(|file| file.level() == 0));
    }

    #[tokio::test]
    async fn test_older_l1_file_reduction_has_priority_over_newer_overlap_resolution() {
        let files = priority_test_files(0, 1, 40, false)
            .into_iter()
            .chain(priority_test_files(1, 1, 100, true))
            .collect::<Vec<_>>();
        let windows = assign_to_windows(files.iter(), 1);

        let output = priority_test_picker(1)
            .build_output_with_time_range(RegionId::from_u64(1), windows, Some(1), None)
            .await
            .unwrap();

        assert_eq!(1, output.len());
        assert_eq!(0, output_window(&output[0]));
    }

    #[tokio::test]
    async fn test_multiple_slots_follow_class_priority_and_newer_first_within_class() {
        let files = [
            priority_test_files(0, 0, 40, false),
            priority_test_files(1, 0, 40, false),
            priority_test_files(2, 1, 40, false),
            priority_test_files(3, 1, 40, false),
            priority_test_files(4, 1, 100, true),
            priority_test_files(5, 1, 100, true),
        ]
        .into_iter()
        .flatten()
        .collect::<Vec<_>>();
        let windows = assign_to_windows(files.iter(), 1);

        let output = priority_test_picker(5)
            .build_output_with_time_range(RegionId::from_u64(1), windows, Some(5), None)
            .await
            .unwrap();
        let pop_priority = output
            .iter()
            .rev()
            .map(|output| (output.inputs[0].level(), output_window(output)))
            .collect::<Vec<_>>();

        // The active window 5 only reduces files, so the overlap slot goes to
        // the inactive window 4.
        assert_eq!(vec![(0, 1), (0, 0), (1, 3), (1, 2), (1, 4)], pop_priority);
        let units = crate::compaction::unit::CompactionUnit::from_picker(PickerOutput {
            outputs: output,
            time_window_size: 1,
            ..Default::default()
        })
        .unwrap();
        let unit_priority = units
            .iter()
            .map(|unit| {
                (
                    unit.outputs[0].inputs[0].level(),
                    output_window(&unit.outputs[0]),
                )
            })
            .collect::<Vec<_>>();
        assert_eq!(
            pop_priority, unit_priority,
            "local FIFO must follow the picker's execution priority"
        );
    }

    #[tokio::test]
    async fn test_inactive_window_mixed_fallback_merges_within_rewrite_budget() {
        let files = [
            new_file_with_level_and_rows(0, 99, 1, 1, 1_000, 1_000_000),
            new_file_with_level_and_rows(0, 9, 0, 2, 10, 1),
        ];
        let windows = assign_to_windows(files.iter(), 100);
        let picker = TwcsPicker {
            trigger_file_num: 4,
            active_window_l1_merge_trigger: 8,
            inactive_window_trigger_file_num: 2,
            inactive_window_l1_merge_trigger: 8,
            time_window_seconds: Some(100),
            max_output_file_size: Some(2_000),
            append_mode: false,
            merge_mode: MergeMode::LastRow,
            max_background_tasks: None,
            time_range: None,
            all_windows_inactive: false,
            metrics: Default::default(),
        };

        let output = picker
            .build_output_with_time_range(RegionId::from_u64(1), windows, Some(100), None)
            .await
            .unwrap();

        assert_eq!(1, output.len());
        assert_eq!(2, output[0].inputs.len());
    }

    #[tokio::test]
    async fn test_inactive_window_over_budget_merges_l0_only() {
        // L0 files are unbalanced among themselves, and merging in the huge L1 file
        // would exceed the rewrite budget: only the L0 files are compacted.
        let files = [
            new_file_handle_with_size_and_sequence(FileId::random(), 0, 9, 0, 1, 10_000),
            new_file_handle_with_size_and_sequence(FileId::random(), 20, 29, 0, 2, 10),
            new_file_handle_with_size_and_sequence(FileId::random(), 0, 99, 1, 3, 1_000_000),
        ];
        let windows = assign_to_windows(files.iter(), 100);
        let picker = TwcsPicker {
            trigger_file_num: 4,
            active_window_l1_merge_trigger: 8,
            inactive_window_trigger_file_num: 2,
            inactive_window_l1_merge_trigger: 8,
            time_window_seconds: Some(100),
            max_output_file_size: Some(100_000),
            append_mode: false,
            merge_mode: MergeMode::LastRow,
            max_background_tasks: None,
            time_range: None,
            all_windows_inactive: false,
            metrics: Default::default(),
        };

        let output = picker
            .build_output_with_time_range(RegionId::from_u64(1), windows, Some(100), None)
            .await
            .unwrap();

        assert_eq!(1, output.len());
        assert_eq!(2, output[0].inputs.len());
        assert!(output[0].inputs.iter().all(|file| file.level() == 0));
    }

    #[tokio::test]
    async fn test_inactive_window_over_budget_without_l0_pair_does_not_merge() {
        // A single tiny L0 and a huge L1: merging is over budget and there is no
        // L0 pair to converge, so the window is left as-is.
        let files = [
            new_file_handle_with_size_and_sequence(FileId::random(), 0, 9, 0, 1, 10),
            new_file_handle_with_size_and_sequence(FileId::random(), 0, 99, 1, 2, 1_000_000),
        ];
        let windows = assign_to_windows(files.iter(), 100);
        let picker = TwcsPicker {
            trigger_file_num: 4,
            active_window_l1_merge_trigger: 8,
            inactive_window_trigger_file_num: 2,
            inactive_window_l1_merge_trigger: 8,
            time_window_seconds: Some(100),
            max_output_file_size: Some(100_000),
            append_mode: false,
            merge_mode: MergeMode::LastRow,
            max_background_tasks: None,
            time_range: None,
            all_windows_inactive: false,
            metrics: Default::default(),
        };

        let output = picker
            .build_output_with_time_range(RegionId::from_u64(1), windows, Some(100), None)
            .await
            .unwrap();

        assert!(output.is_empty());
    }

    #[tokio::test]
    async fn test_inactive_window_falls_through_when_triggered_l0_pick_fails() {
        // L0 reaches the inactive trigger but is unbalanced; L1 is below the
        // trigger. The window must still converge through the budgeted mixed
        // fallback instead of being skipped.
        let files = [
            new_file_handle_with_size_and_sequence(FileId::random(), 0, 9, 0, 1, 100_000),
            new_file_handle_with_size_and_sequence(FileId::random(), 20, 29, 0, 2, 10),
            new_file_handle_with_size_and_sequence(FileId::random(), 0, 99, 1, 3, 50_000),
        ];
        let windows = assign_to_windows(files.iter(), 100);
        let picker = TwcsPicker {
            trigger_file_num: 4,
            active_window_l1_merge_trigger: 8,
            inactive_window_trigger_file_num: 2,
            inactive_window_l1_merge_trigger: 8,
            time_window_seconds: Some(100),
            max_output_file_size: Some(1_000_000),
            append_mode: false,
            merge_mode: MergeMode::LastRow,
            max_background_tasks: None,
            time_range: None,
            all_windows_inactive: false,
            metrics: Default::default(),
        };

        let output = picker
            .build_output_with_time_range(RegionId::from_u64(1), windows, Some(100), None)
            .await
            .unwrap();

        assert_eq!(1, output.len());
        assert_eq!(3, output[0].inputs.len());
    }

    #[test]
    fn test_mixed_candidate_row_balance() {
        for (case, l1_rows, l0_rows, expected_len) in [
            ("L1 rows dominate", 1_000_000, 10_000, 0),
            ("L1 rows meet ratio", 60_000, 10_000, 4),
            ("L0 rows are unknown", 1_000_000, 0, 4),
        ] {
            let files = vec![
                new_file_with_level_and_rows(0, 99, 1, 1, 100, l1_rows),
                new_file_with_level_and_rows(0, 9, 0, 2, 100, l0_rows),
                new_file_with_level_and_rows(20, 29, 0, 3, 100, l0_rows),
                new_file_with_level_and_rows(40, 49, 0, 4, 100, l0_rows),
            ];

            let picked = pick_mixed_count_first(vec![SortedRun::from(files)], None);

            assert_eq!(expected_len, picked.len(), "{case}");
        }
    }

    #[test]
    fn test_count_first_prefers_smaller_bytes_when_file_counts_match() {
        let files = (0..=DEFAULT_MAX_INPUT_FILES)
            .map(|idx| {
                let start = idx as i64 * 20;
                let size = if idx == 0 {
                    100
                } else if idx == DEFAULT_MAX_INPUT_FILES {
                    1
                } else {
                    10
                };
                new_file(start, start + 9, idx as u64 + 1, size)
            })
            .collect::<Vec<_>>();

        let picked = pick_count_first(vec![SortedRun::from(files)], None);
        let ranges = picked_ranges(&picked);

        assert_eq!(DEFAULT_MAX_INPUT_FILES, ranges.len());
        assert_eq!(Some(&(20, 29)), ranges.first());
        let last_start = DEFAULT_MAX_INPUT_FILES as i64 * 20;
        assert_eq!(Some(&(last_start, last_start + 9)), ranges.last());
    }

    #[test]
    fn test_count_first_skips_pure_rewrite_without_progress() {
        // Four non-overlapping files whose combined size splits back into at least
        // four outputs: compacting them reduces neither the file count nor any
        // overlap, so there is nothing worth picking.
        let picked = pick_count_first(
            vec![SortedRun::from(vec![
                new_file(0, 9, 1, 600),
                new_file(20, 29, 2, 600),
                new_file(40, 49, 3, 600),
                new_file(60, 69, 4, 600),
            ])],
            Some(512),
        );

        assert!(picked.is_empty());
    }

    #[test]
    fn test_count_first_allows_overlap_resolution_without_file_reduction() {
        // Two large overlapping files from different runs: compacting them cannot
        // reduce the file count (the output splits back into just as many files),
        // but it resolves the overlap and merges two sorted runs into one.
        let picked = pick_count_first(
            vec![
                SortedRun::from(vec![new_file(0, 19, 1, 600)]),
                SortedRun::from(vec![new_file(10, 29, 2, 600)]),
            ],
            Some(512),
        );

        assert_eq!(vec![(0, 19), (10, 29)], picked_ranges(&picked));
    }

    #[tokio::test]
    async fn test_l1_partial_pk_overlap_does_not_rewrite_unrelated_files() {
        // Region 1025's 2026-09-23 R2: A1..A8 are R1's outputs, plus one L1 E.
        // Keep the logged, rounded sizes (tenths of MiB) and normalize timestamps
        // while preserving the observed order: A3, A7, A6, A1, A4, A5, A2, A8, E.
        // The PK ranges are hypothetical: the logs only contain their lengths.
        let a_files = [
            (4, 5249), // A1
            (7, 5218), // A2
            (1, 5255), // A3
            (5, 5263), // A4
            (6, 5131), // A5
            (3, 5143), // A6
            (2, 5197), // A7
            (8, 4716), // A8
        ]
        .into_iter()
        .enumerate()
        .map(|(idx, (start, size))| {
            let key = b'a' + idx as u8;
            new_file_handle_with_size_sequence_and_primary_key_range(
                FileId::random(),
                start,
                10_000,
                1,
                100,
                size * 1024 * 1024 / 10,
                pk_range(&[key, b'0'], &[key, b'9']),
            )
        })
        .collect::<Vec<_>>();
        let mapper = primary_key_mapper_for_test();
        for (idx, file) in a_files.iter().enumerate() {
            for other in &a_files[idx + 1..] {
                assert!(file.overlap_inclusive(other)); // Their time ranges overlap.
                assert!(!files_overlap_inclusive(file, other, &mapper));
            }
        }

        let picker = TwcsPicker {
            trigger_file_num: 4,
            active_window_l1_merge_trigger: 4,
            inactive_window_trigger_file_num: 2,
            inactive_window_l1_merge_trigger: 8,
            time_window_seconds: Some(60),
            max_output_file_size: Some(512 * 1024 * 1024),
            append_mode: false,
            merge_mode: MergeMode::LastRow,
            max_background_tasks: Some(1),
            time_range: None,
            all_windows_inactive: false,
            metrics: Default::default(),
        };

        for (case, key_range, extra_size, overlapping_a, expected_inputs) in [
            (
                "A3 and E skip the unrelated A files",
                Some((b'c', b'c')),
                5155,
                &[2][..],
                &[2, 8][..],
            ),
            (
                "A8 and E must not pull in A2",
                Some((b'h', b'h')),
                5155,
                &[7][..],
                &[7, 8][..],
            ),
            (
                "E overlaps no A: no progress",
                Some((b'z', b'z')),
                5155,
                &[][..],
                &[] as &[usize],
            ),
            (
                "A3 and E may split into one extra tail file",
                Some((b'c', b'c')),
                5255,
                &[2][..],
                &[2, 8][..],
            ),
            (
                "two A files exceed E's run budget",
                Some((b'b', b'c')),
                5155,
                &[1, 2][..],
                &[][..],
            ),
            (
                "sufficient E bytes allow two A files separated in time order",
                Some((b'b', b'c')),
                6000,
                &[1, 2][..],
                &[1, 2, 8][..],
            ),
            (
                "unknown PK bounds include the whole dominant run in the budget",
                None,
                5155,
                &[0, 1, 2, 3, 4, 5, 6, 7][..],
                &[][..],
            ),
        ] {
            let extra = new_file_handle_with_size_sequence_and_primary_key_range(
                FileId::random(),
                1_000,
                9_000,
                1,
                100,
                extra_size * 1024 * 1024 / 10,
                key_range.and_then(|(min, max)| pk_range(&[min, b'0'], &[max, b'9'])),
            );
            for (idx, file) in a_files.iter().enumerate() {
                assert!(file.overlap_inclusive(&extra));
                assert_eq!(
                    overlapping_a.contains(&idx),
                    files_overlap_inclusive(file, &extra, &mapper),
                    "{case}: A{}",
                    idx + 1,
                );
            }

            let mut files = a_files.clone();
            files.push(extra);
            for active_window in [Some(0), None] {
                // The active window leaves pure overlap resolution to the window
                // becoming inactive.
                let expected_inputs = if active_window.is_some() {
                    &[][..]
                } else {
                    expected_inputs
                };
                let expected = expected_inputs
                    .iter()
                    .map(|&idx| files[idx].file_id())
                    .collect::<HashSet<_>>();
                let windows = assign_to_windows(files.iter(), 60);
                let outputs = picker
                    .build_output_with_time_range(
                        RegionId::from_u64(0),
                        windows,
                        active_window,
                        Some(60),
                    )
                    .await
                    .unwrap();

                assert_eq!(
                    usize::from(!expected_inputs.is_empty()),
                    outputs.len(),
                    "{case}, active window: {active_window:?}"
                );
                let actual = outputs
                    .iter()
                    .flat_map(|output| output.inputs.iter().map(FileHandle::file_id))
                    .collect::<HashSet<_>>();
                assert_eq!(expected, actual, "{case}, active window: {active_window:?}");
            }
        }
    }

    #[rstest::rstest]
    #[case::active_filtered(Some(0), false, 1000, &[0, 2])]
    #[case::inactive_filtered(None, false, 1000, &[0, 2])]
    #[case::older_reduction_before_newer_overlap(Some(3), true, 1000, &[0, 2])]
    #[case::contiguous_has_more_reduction(Some(0), false, 90, &[0, 1, 2, 3])]
    #[case::filtered_has_equal_reduction_with_less_io(Some(0), false, 600, &[0, 2])]
    #[tokio::test]
    async fn test_l1_file_reduction_considers_filtered_candidates(
        #[case] active_window: Option<i64>,
        #[case] newer_overlap_window: bool,
        #[case] unrelated_size: u64,
        #[case] expected_indices: &[usize],
    ) {
        // Only A/B overlap. With 1000-byte unrelated files between/after them,
        // no contiguous interval reduces the file count, but A+B alone does.
        let mut files = [
            (b'a', 100),
            (b'b', unrelated_size),
            (b'a', 100),
            (b'c', unrelated_size),
        ]
        .into_iter()
        .enumerate()
        .map(|(idx, (key, size))| {
            new_file_handle_with_size_sequence_and_primary_key_range(
                FileId::random(),
                idx as i64 * 100,
                749,
                1,
                1,
                size,
                pk_range(&[key], &[key]),
            )
        })
        .collect::<Vec<_>>();
        let expected = expected_indices
            .iter()
            .map(|&idx| files[idx].file_id())
            .collect::<Vec<_>>();
        if newer_overlap_window {
            // One slot: a filtered file-reducing pick must beat a newer pure
            // rewrite, rather than being deferred behind overlap resolution.
            files.extend(priority_test_files(1, 1, 512, true));
        }
        let picker = TwcsPicker {
            max_output_file_size: Some(512),
            ..priority_test_picker(1)
        };
        let outputs = picker
            .build_output_with_time_range(
                RegionId::from_u64(0),
                assign_to_windows(files.iter(), 1),
                active_window,
                Some(1),
            )
            .await
            .unwrap();

        assert_eq!(1, outputs.len());
        assert_eq!(
            expected,
            outputs[0]
                .inputs
                .iter()
                .map(FileHandle::file_id)
                .collect::<Vec<_>>()
        );
    }

    #[tokio::test]
    async fn test_overlap_resolution_limits_dominant_run_rewrites() {
        let file_size = 512 * 1024 * 1024;
        // Peers are full-size files in one run, so a pure rewrite keeps the file count.
        for (case, peer_sizes, max_output_size, should_pick) in [
            (
                "small peer cannot rewrite any slice of the large run",
                vec![file_size],
                file_size,
                false,
            ),
            (
                "one byte below the run budget",
                [vec![file_size; 6], vec![file_size - 1]].concat(),
                file_size,
                false,
            ),
            (
                "exact run budget boundary",
                vec![file_size; 7],
                file_size,
                true,
            ),
            ("equal sized runs", vec![file_size; 8], file_size, true),
            (
                "file reduction keeps its own admission",
                vec![file_size],
                2 * file_size,
                true,
            ),
        ] {
            let mut files = (0..8)
                .map(|idx| {
                    let key = b'a' + idx as u8;
                    new_file_handle_with_size_sequence_and_primary_key_range(
                        FileId::random(),
                        idx,
                        10_000,
                        1,
                        100,
                        file_size,
                        pk_range(&[key, b'0'], &[key, b'9']),
                    )
                })
                .collect::<Vec<_>>();
            files.extend(peer_sizes.into_iter().enumerate().map(|(idx, size)| {
                let start = 1_000 * (idx as i64 + 1);
                new_file_handle_with_size_sequence_and_primary_key_range(
                    FileId::random(),
                    start,
                    start + 999,
                    1,
                    100,
                    size,
                    pk_range(b"a0", b"h9"),
                )
            }));
            // Only the last case removes files; the active window leaves pure
            // overlap resolution to the window becoming inactive.
            let reduces_files = max_output_size > file_size;
            let picker = TwcsPicker {
                trigger_file_num: 4,
                active_window_l1_merge_trigger: 4,
                inactive_window_trigger_file_num: 2,
                inactive_window_l1_merge_trigger: 8,
                time_window_seconds: Some(60),
                max_output_file_size: Some(max_output_size),
                append_mode: false,
                merge_mode: MergeMode::LastRow,
                max_background_tasks: Some(1),
                time_range: None,
                all_windows_inactive: false,
                metrics: Default::default(),
            };

            for active_window in [Some(0), None] {
                let expected = if should_pick && (reduces_files || active_window.is_none()) {
                    files
                        .iter()
                        .map(FileHandle::file_id)
                        .collect::<HashSet<_>>()
                } else {
                    HashSet::new()
                };
                let outputs = picker
                    .build_output_with_time_range(
                        RegionId::from_u64(0),
                        assign_to_windows(files.iter(), 60),
                        active_window,
                        Some(60),
                    )
                    .await
                    .unwrap();
                let actual = outputs
                    .iter()
                    .flat_map(|output| output.inputs.iter().map(FileHandle::file_id))
                    .collect::<HashSet<_>>();
                assert_eq!(expected, actual, "{case}, active window: {active_window:?}");
            }
        }
    }

    #[tokio::test]
    async fn test_quiet_or_manual_window_resolves_dominant_run_overlap() {
        let file_size = 512 * 1024 * 1024;
        let l1_file =
            |start, end, sequence, size, range: Option<(bytes::Bytes, bytes::Bytes)>, num_rows| {
                let file = new_file_handle_with_size_sequence_and_primary_key_range(
                    FileId::random(),
                    start,
                    end,
                    1,
                    sequence,
                    size,
                    range.clone(),
                );
                FileHandle::new_with_primary_key_range(
                    FileMeta {
                        num_rows,
                        ..file.meta_ref().clone()
                    },
                    crate::test_util::new_noop_file_purger(),
                    range,
                )
            };

        // The window holds 9 rows. It is quiet once the region writes 9 more
        // rows after it, and only if its row counts are known. A manual
        // compaction does not wait, even for the active window.
        for (newer_sequence, num_rows, manual, resolves) in [
            (108, 1, false, false),
            (109, 1, false, true),
            (109, 0, false, false),
            (100, 1, true, true),
        ] {
            // A run of eight key-sliced files and one small file overlapping all
            // of them: resolving the overlap rewrites the run for its small peer.
            let mut window_files = (0..8)
                .map(|idx| {
                    let key = b'a' + idx as u8;
                    let range = pk_range(&[key, b'0'], &[key, b'9']);
                    l1_file(idx, 10_000, 100, file_size, range, num_rows)
                })
                .collect::<Vec<_>>();
            window_files.push(l1_file(
                1_000,
                1_999,
                100,
                file_size / 8,
                pk_range(b"a0", b"h9"),
                num_rows,
            ));
            let expected = window_files
                .iter()
                .map(FileHandle::file_id)
                .collect::<HashSet<_>>();
            let picker = TwcsPicker {
                trigger_file_num: 4,
                active_window_l1_merge_trigger: 4,
                inactive_window_trigger_file_num: 2,
                inactive_window_l1_merge_trigger: 16,
                time_window_seconds: Some(60),
                max_output_file_size: Some(file_size),
                append_mode: false,
                merge_mode: MergeMode::LastRow,
                max_background_tasks: Some(1),
                time_range: None,
                all_windows_inactive: manual,
                metrics: Default::default(),
            };
            let mut files = window_files;
            files.push(l1_file(
                120_000,
                120_999,
                newer_sequence,
                file_size,
                pk_range(b"a0", b"h9"),
                1,
            ));
            let outputs = picker
                .build_output_with_time_range(
                    RegionId::from_u64(0),
                    assign_to_windows(files.iter(), 60),
                    manual.then_some(0),
                    Some(60),
                )
                .await
                .unwrap();
            let actual = outputs
                .iter()
                .flat_map(|output| output.inputs.iter().map(FileHandle::file_id))
                .collect::<HashSet<_>>();
            let case = format!("sequence {newer_sequence}, rows {num_rows}, manual {manual}");
            if resolves {
                assert_eq!(expected, actual, "{case}");
            } else {
                assert!(actual.is_empty(), "{case}: {actual:?}");
            }
        }
    }

    #[test]
    fn test_count_first_prefers_guaranteed_reduction_over_pure_rewrite() {
        // A window mixing large and small files: the large triple on its own is a
        // pure rewrite, and every interval containing large files rewrites more
        // bytes for the same predicted reduction, so the small triple wins.
        let picked = pick_count_first(
            vec![SortedRun::from(vec![
                new_file(0, 9, 1, 600),
                new_file(10, 19, 2, 600),
                new_file(20, 29, 3, 600),
                new_file(30, 39, 4, 10),
                new_file(40, 49, 5, 10),
                new_file(50, 59, 6, 10),
            ])],
            Some(512),
        );

        assert_eq!(vec![(30, 39), (40, 49), (50, 59)], picked_ranges(&picked));
    }

    #[test]
    fn test_count_first_prefers_earlier_time_on_exact_tie() {
        let files = (0..=DEFAULT_MAX_INPUT_FILES)
            .map(|idx| {
                let start = idx as i64 * 20;
                new_file(start, start + 9, idx as u64 + 1, 10)
            })
            .collect::<Vec<_>>();

        let picked = pick_count_first(vec![SortedRun::from(files)], None);
        let ranges = picked_ranges(&picked);

        assert_eq!(DEFAULT_MAX_INPUT_FILES, ranges.len());
        assert_eq!(Some(&(0, 9)), ranges.first());
        let last_start = (DEFAULT_MAX_INPUT_FILES as i64 - 1) * 20;
        assert_eq!(Some(&(last_start, last_start + 9)), ranges.last());
    }

    #[test]
    fn test_count_first_keeps_each_interleaved_run_contiguous() {
        let picked = pick_count_first(
            vec![
                SortedRun::from(vec![
                    new_file(0, 9, 1, 1),
                    new_file(20, 29, 2, 1),
                    new_file(40, 49, 3, 1),
                ]),
                SortedRun::from(vec![new_file(10, 19, 4, 1), new_file(30, 39, 5, 1)]),
            ],
            None,
        );

        assert_eq!(
            vec![(0, 9), (10, 19), (20, 29), (30, 39), (40, 49)],
            picked_ranges(&picked)
        );
    }

    // TODO(hl): TTL tester that checks if get_expired_ssts function works as expected.

    // Reproduces the leftover-small-files deadlock observed in the compaction bench:
    // an "EngineFull" flush emits several SSTs sharing one sequence. They must be ordinary
    // candidates so a contiguous candidate can span them instead of treating them as an
    // indivisible block that fragments the timeline.
    #[test]
    fn test_count_first_candidate_spans_same_sequence_files() {
        let picked = pick_count_first(
            vec![SortedRun::from(vec![
                new_file(0, 9, 1, 10),
                // One EngineFull flush output: three SSTs with the same sequence.
                new_file(10, 19, 2, 10),
                new_file(10, 19, 2, 10),
                new_file(10, 19, 2, 10),
                new_file(20, 29, 3, 10),
            ])],
            None,
        );

        assert_eq!(
            vec![(0, 9), (10, 19), (10, 19), (10, 19), (20, 29)],
            picked_ranges(&picked)
        );
    }

    // End-to-end repro of the compaction-bench leftover files: a window where
    // EngineFull flush groups (multiple SSTs sharing a sequence) sit between
    // singleton flushes. The picker must be able to merge the whole contiguous
    // range; otherwise these files can never be compacted once the active window
    // moves on.
    #[tokio::test]
    async fn test_count_first_merges_window_with_interleaved_flush_groups() {
        let files = [
            // Singleton flush.
            new_file_handle_with_size_and_sequence(FileId::random(), 0, 9, 0, 1, 10),
            // EngineFull flush: three SSTs sharing sequence 2.
            new_file_handle_with_size_and_sequence(FileId::random(), 10, 19, 0, 2, 10),
            new_file_handle_with_size_and_sequence(FileId::random(), 10, 19, 0, 2, 10),
            new_file_handle_with_size_and_sequence(FileId::random(), 10, 19, 0, 2, 10),
            // Singleton flush.
            new_file_handle_with_size_and_sequence(FileId::random(), 20, 29, 0, 3, 10),
        ];
        let windows = assign_to_windows(files.iter(), 100);
        let picker = TwcsPicker {
            trigger_file_num: 3,
            active_window_l1_merge_trigger: 8,
            inactive_window_trigger_file_num: 3,
            inactive_window_l1_merge_trigger: 8,
            time_window_seconds: Some(100),
            max_output_file_size: None,
            append_mode: false,
            merge_mode: MergeMode::LastRow,
            max_background_tasks: None,
            time_range: None,
            all_windows_inactive: false,
            metrics: Default::default(),
        };

        let output = picker
            .build_output_with_time_range(RegionId::from_u64(1), windows, Some(0), None)
            .await
            .unwrap();

        assert_eq!(1, output.len());
        // All five SSTs of the window are merged in one compaction.
        assert_eq!(5, output[0].inputs.len());
    }
}