coordinode-lsm-tree 7.4.1

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

use crate::{
    InternalValue,
    comparator::SharedComparator,
    key::InternalKey,
    memtable::Memtable,
    merge_operator::MergeOperator,
    merge_source::{CoherentIterSource, CoherentMergeSource, IterItem, MergeSource},
    mvcc_stream::MvccStream,
    range_tombstone::RangeTombstone,
    range_tombstone_filter::RangeTombstoneFilter,
    reseek::{ReseekCtx, Reseekable},
    run_reader::RunReader,
    seeking_merger::SeekingMerger,
    skip_key::SkipKey,
    value::{SeqNo, UserKey},
    version::{Run, SuperVersion},
};
use alloc::sync::Arc;
#[cfg(not(feature = "std"))]
use alloc::{boxed::Box, vec::Vec};
use core::ops::{Bound, RangeBounds};

use self_cell::self_cell;

#[must_use]
pub fn seqno_filter(item_seqno: SeqNo, seqno: SeqNo) -> bool {
    item_seqno < seqno
}

/// One source of a scan: double-ended, and able to jump past a resolved key's
/// versions (see [`SkipKey`]).
trait ScanSource: DoubleEndedIterator<Item = crate::Result<InternalValue>> + SkipKey + Send {}

impl<T: DoubleEndedIterator<Item = crate::Result<InternalValue>> + SkipKey + Send> ScanSource
    for T
{
}

/// A boxed [`ScanSource`].
type ScanBox<'a> = Box<dyn ScanSource + 'a>;

/// Wrap a vector of boxed double-ended source iterators into a
/// `SeekingMerger`, the loser-tree-based merging iterator that
/// replaces the legacy `Merger` (sorted-vector heap) on the read
/// path. The `ScanBox`es become `MergeSource`s via the
/// `CoherentIterSource` adapter — its coherent-cursor contract
/// is satisfied by every memtable / run / table iter we feed in
/// here (they all share a single front/back cursor over the
/// remaining range).
///
/// The source type stays concrete (`CoherentIterSource<ScanBox<'a>>`)
/// rather than being re-boxed behind a `dyn MergeSource` trait
/// object: `ScanBox` is itself already a heap-allocated
/// trait object, so wrapping it in another `Box<dyn ...>` would
/// cost an extra allocation per source AND an extra vtable
/// dispatch on every `next`/`next_back`. With the concrete
/// adapter type, only the inner `ScanBox`'s vtable
/// remains — exactly one indirect call per per-step source pull.
///
/// The compaction merge path still uses the legacy `Merger`
/// because compaction `Scanner`s are forward-only (`Iterator`,
/// not `DoubleEndedIterator`) and `MergeSource` requires
/// `next_back`. That swap is a separate refactor.
fn build_seeking<'a>(
    iters: Vec<ScanBox<'a>>,
    comparator: SharedComparator,
) -> SeekingMerger<CoherentIterSource<ScanBox<'a>>, SharedComparator> {
    let sources: Vec<CoherentIterSource<ScanBox<'a>>> =
        iters.into_iter().map(CoherentIterSource::new).collect();
    SeekingMerger::new(sources, comparator)
}

/// The rows of a source the scan's snapshot sees: those below its seqno.
/// Errors pass through.
struct Visible<I> {
    inner: I,
    seqno: SeqNo,
}

impl<I: Iterator<Item = crate::Result<InternalValue>>> Iterator for Visible<I> {
    type Item = crate::Result<InternalValue>;

    fn next(&mut self) -> Option<Self::Item> {
        loop {
            match self.inner.next()? {
                Ok(item) if !seqno_filter(item.key.seqno, self.seqno) => {}
                other => return Some(other),
            }
        }
    }
}

impl<I: DoubleEndedIterator<Item = crate::Result<InternalValue>>> DoubleEndedIterator
    for Visible<I>
{
    fn next_back(&mut self) -> Option<Self::Item> {
        loop {
            match self.inner.next_back()? {
                Ok(item) if !seqno_filter(item.key.seqno, self.seqno) => {}
                other => return Some(other),
            }
        }
    }
}

impl<I: SkipKey> SkipKey for Visible<I> {
    fn skip_front_past(&mut self, key: &[u8]) -> crate::Result<bool> {
        self.inner.skip_front_past(key)
    }

    /// The newest version this source's snapshot sees, which is what the
    /// caller resolves the key from.
    fn skip_back_to_newest(&mut self, key: &[u8], seqno: SeqNo) -> crate::Result<bool> {
        self.inner.skip_back_to_newest(key, seqno.min(self.seqno))
    }
}

/// A memtable's rows over a range.
struct MemtableRows<'a>(crate::memtable::skiplist::Range<'a>);

impl Iterator for MemtableRows<'_> {
    type Item = crate::Result<InternalValue>;

    fn next(&mut self) -> Option<Self::Item> {
        self.0.next().map(|entry| {
            Ok(InternalValue {
                key: entry.key(),
                value: entry.value(),
            })
        })
    }
}

impl DoubleEndedIterator for MemtableRows<'_> {
    fn next_back(&mut self) -> Option<Self::Item> {
        self.0.next_back().map(|entry| {
            Ok(InternalValue {
                key: entry.key(),
                value: entry.value(),
            })
        })
    }
}

impl SkipKey for MemtableRows<'_> {
    fn skip_front_past(&mut self, key: &[u8]) -> crate::Result<bool> {
        self.0.skip_front_past(key);
        Ok(true)
    }

    fn skip_back_to_newest(&mut self, key: &[u8], seqno: SeqNo) -> crate::Result<bool> {
        Ok(self.0.skip_back_to_newest(key, seqno))
    }
}

/// The source over `memtable`'s rows in `range` that a snapshot at `seqno`
/// sees.
fn memtable_source(
    memtable: &Memtable,
    range: (Bound<InternalKey>, Bound<InternalKey>),
    seqno: SeqNo,
) -> ScanBox<'_> {
    Box::new(Visible {
        inner: MemtableRows(memtable.items.range(range)),
        seqno,
    })
}

/// Calculates the prefix's upper range.
///
/// # Panics
///
/// Panics if the prefix is empty.
pub(crate) fn prefix_upper_range(prefix: &[u8]) -> Bound<UserKey> {
    use core::ops::Bound::{Excluded, Unbounded};

    assert!(!prefix.is_empty(), "prefix may not be empty");

    let mut end = prefix.to_vec();
    let len = end.len();

    for (idx, byte) in end.iter_mut().rev().enumerate() {
        let idx = len - 1 - idx;

        if *byte < 255 {
            *byte += 1;
            end.truncate(idx + 1);
            return Excluded(end.into());
        }
    }

    Unbounded
}

/// Converts a prefix to range bounds.
#[must_use]
#[expect(clippy::module_name_repetitions)]
pub fn prefix_to_range(prefix: &[u8]) -> (Bound<UserKey>, Bound<UserKey>) {
    use core::ops::Bound::{Included, Unbounded};

    if prefix.is_empty() {
        return (Unbounded, Unbounded);
    }

    (Included(prefix.into()), prefix_upper_range(prefix))
}

/// The iter state references the memtables used while the range is open
///
/// Because of Rust rules, the state is referenced using `self_cell`, see below.
///
/// `Clone` is cheap (every field is an `Arc` / small `Copy` value) and is used
/// by the seekable range iterator to snapshot the version once; repositions then
/// reseek the leaf cursors in place against that same snapshot.
#[derive(Clone)]
pub struct IterState {
    pub(crate) version: SuperVersion,
    pub(crate) ephemeral: Option<(Arc<Memtable>, SeqNo)>,
    pub(crate) merge_operator: Option<Arc<dyn MergeOperator>>,

    /// Where a merge reads a base kept in the value log; set for a blob
    /// tree's index only.
    pub(crate) blob_source: Option<crate::blob_tree::BlobSource>,

    /// User key comparator for merge ordering.
    pub(crate) comparator: crate::comparator::SharedComparator,

    /// Optional prefix hash for prefix bloom filter skipping.
    ///
    /// When set, segments whose bloom filter reports no match for this
    /// hash will be skipped entirely during the scan.
    pub(crate) prefix_hash: Option<u64>,

    /// Optional metrics handle for recording prefix-related statistics (e.g. bloom skips).
    ///
    /// `None` when the caller does not wish to record metrics; this is
    /// independent of whether the iterator uses a prefix.
    #[cfg(feature = "metrics")]
    pub(crate) metrics: Option<Arc<crate::Metrics>>,
}

impl IterState {
    /// The value log a merge reads a separated base from: this state's blob
    /// source over the version it holds.
    fn value_log(&self) -> Option<crate::mvcc_stream::ValueLog<'_>> {
        self.blob_source
            .as_ref()
            .map(|source| crate::mvcc_stream::ValueLog {
                source,
                version: &self.version.version,
            })
    }

    /// What this read asks each table's filters: a scan asks by its prefix,
    /// never by one key, which a point read asks through its own path.
    fn filter_query(&self) -> FilterQuery<'_> {
        FilterQuery {
            prefix_hash: self.prefix_hash,
            key_hash: None,
            bloom_key: None,
            #[cfg(feature = "metrics")]
            metrics: self.metrics.as_ref(),
        }
    }
}

/// What a read asks each table's filters: the prefix it scans, or the key it
/// reads.
#[derive(Clone, Copy)]
pub(crate) struct FilterQuery<'a> {
    /// The hash of the prefix a scan reads, for prefix filters.
    pub(crate) prefix_hash: Option<u64>,
    /// The hash of the key a point read resolves, for key filters.
    pub(crate) key_hash: Option<u64>,
    /// That key, for a partitioned filter to seek to its partition; set only
    /// alongside `key_hash`.
    pub(crate) bloom_key: Option<&'a [u8]>,
    #[cfg(feature = "metrics")]
    pub(crate) metrics: Option<&'a Arc<crate::Metrics>>,
}

type BoxedMerge<'a> = Box<dyn DoubleEndedIterator<Item = crate::Result<InternalValue>> + Send + 'a>;

self_cell!(
    pub struct TreeIter {
        owner: IterState,

        #[covariant]
        dependent: BoxedMerge,
    }
);

impl Iterator for TreeIter {
    type Item = crate::Result<InternalValue>;

    fn next(&mut self) -> Option<Self::Item> {
        self.with_dependent_mut(|_, iter| iter.next())
    }
}

impl DoubleEndedIterator for TreeIter {
    fn next_back(&mut self) -> Option<Self::Item> {
        self.with_dependent_mut(|_, iter| iter.next_back())
    }
}

pub(crate) fn range_tombstone_overlaps_bounds(
    rt: &RangeTombstone,
    bounds: &(Bound<UserKey>, Bound<UserKey>),
    comparator: &dyn crate::comparator::UserComparator,
) -> bool {
    let overlaps_lo = match &bounds.0 {
        Bound::Included(key) | Bound::Excluded(key) => {
            comparator.compare(&rt.end, key) == core::cmp::Ordering::Greater
        }
        Bound::Unbounded => true,
    };

    let overlaps_hi = match &bounds.1 {
        Bound::Included(key) => comparator.compare(&rt.start, key) != core::cmp::Ordering::Greater,
        Bound::Excluded(key) => comparator.compare(&rt.start, key) == core::cmp::Ordering::Less,
        Bound::Unbounded => true,
    };

    overlaps_lo && overlaps_hi
}

/// What a table's filters said of a read's prefix and key.
#[derive(Clone, Copy, PartialEq, Eq)]
enum FilterAnswer {
    /// A filter rules the table out.
    Absent,
    /// The key's filter was asked and let the key through, a probe counted:
    /// a read of the table finding no version of the key is a false positive.
    KeyPassed,
    /// The prefix's filter was asked and let the prefix through, a probe
    /// counted: the scan is over exactly that prefix, so a table holding no
    /// key under it is a false positive.
    PrefixPassed,
    /// The table is read with no probe counted: no filter was asked, or none
    /// could be read.
    Unanswered,
}

impl FilterAnswer {
    /// A filter let the read through: a read of the table finding nothing is
    /// the false positive of the probe already counted.
    fn passed(self) -> bool {
        matches!(self, Self::KeyPassed | Self::PrefixPassed)
    }
}

/// Checks prefix and key bloom filters for a table.
///
/// Returns `true` if the table should be included (bloom says "maybe" or no
/// filter available), `false` if it can be safely skipped.
fn bloom_passes(state: &IterState, table: &crate::table::Table) -> bool {
    filter_answer(state.filter_query(), table) != FilterAnswer::Absent
}

/// Asks a table's prefix and key filters about the read, counting the probes
/// they answer (see [`FilterAnswer`]).
fn filter_answer(query: FilterQuery<'_>, table: &crate::table::Table) -> FilterAnswer {
    let mut answer = FilterAnswer::Unanswered;
    // A filter indexing another extractor's prefixes, or none, is not asked.
    if let Some(prefix_hash) = query.prefix_hash
        && table.has_prefix_filter()
    {
        // A prefix answer counts as a key's does: the filter holds the
        // prefix's hash beside the keys'.
        use crate::table::probe_stats::ProbeCounts;
        match table.maybe_contains_prefix(prefix_hash) {
            Ok(false) => {
                table.count_probes(ProbeCounts {
                    probes: 1,
                    negatives: 1,
                });
                #[cfg(feature = "metrics")]
                if let Some(m) = query.metrics {
                    m.prefix_bloom_skips
                        .fetch_add(1, core::sync::atomic::Ordering::Relaxed);
                }
                return FilterAnswer::Absent;
            }
            Ok(true) if table.key_check_consults_filter(false) => {
                table.count_probes(ProbeCounts {
                    probes: 1,
                    negatives: 0,
                });
                answer = FilterAnswer::PrefixPassed;
            }
            Err(e) => {
                log::debug!("prefix bloom check failed for table {:?}: {e}", table.id());
            }
            Ok(true) => {}
        }
    }

    // bloom_key without key_hash is meaningless — catch misuse early
    debug_assert!(
        query.bloom_key.is_none() || query.key_hash.is_some(),
        "bloom_key requires key_hash to be set"
    );

    if let Some(key_hash) = query.key_hash {
        use crate::table::KeyFilterAnswer;
        let result = if let Some(bloom_key) = query.bloom_key {
            table.key_filter_answer(bloom_key, key_hash)
        } else {
            table.bloom_may_contain_key_hash(key_hash).map(|may| {
                if may {
                    KeyFilterAnswer::MayContain
                } else {
                    KeyFilterAnswer::Absent
                }
            })
        };
        // The filter's answers count as a point read's do; a key passed for
        // want of a filter, or ruled out by the partition index alone, is no
        // probe of one.
        use crate::table::probe_stats::ProbeCounts;
        match result {
            Ok(KeyFilterAnswer::Absent) => {
                table.count_probes(ProbeCounts {
                    probes: 1,
                    negatives: 1,
                });
                return FilterAnswer::Absent;
            }
            Ok(KeyFilterAnswer::PastPartitions) => return FilterAnswer::Absent,
            Ok(KeyFilterAnswer::MayContain)
                if table.key_check_consults_filter(query.bloom_key.is_some()) =>
            {
                table.count_probes(ProbeCounts {
                    probes: 1,
                    negatives: 0,
                });
                return FilterAnswer::KeyPassed;
            }
            Err(e) => {
                log::debug!("key bloom check failed for table {:?}: {e}", table.id());
            }
            Ok(KeyFilterAnswer::MayContain) => {}
        }
    }

    answer
}

/// One table's versions of a point key or of the keys under a prefix, read
/// after its filter let the key or prefix through. A reader that ends having
/// found none shows the filter answered for what the table holds no version
/// of, and counts it as the point read counts a false positive. It reads
/// every version whatever the snapshot, so finding none is conclusive; a
/// reader dropped before it ends counts nothing.
struct FilterPassReader<I, T> {
    inner: I,
    table: T,
    /// A version was found, or the miss was already counted.
    settled: bool,
}

impl<I, T> FilterPassReader<I, T>
where
    I: DoubleEndedIterator<Item = crate::Result<InternalValue>>,
    T: core::borrow::Borrow<crate::table::Table>,
{
    fn settle(
        &mut self,
        item: Option<crate::Result<InternalValue>>,
    ) -> Option<crate::Result<InternalValue>> {
        if !self.settled {
            self.settled = true;
            if item.is_none() {
                self.table
                    .borrow()
                    .count_probes(crate::table::probe_stats::ProbeCounts {
                        probes: 0,
                        negatives: 1,
                    });
            }
        }
        item
    }
}

impl<I, T> Iterator for FilterPassReader<I, T>
where
    I: DoubleEndedIterator<Item = crate::Result<InternalValue>>,
    T: core::borrow::Borrow<crate::table::Table>,
{
    type Item = crate::Result<InternalValue>;

    fn next(&mut self) -> Option<Self::Item> {
        let item = self.inner.next();
        self.settle(item)
    }
}

impl<I, T> DoubleEndedIterator for FilterPassReader<I, T>
where
    I: DoubleEndedIterator<Item = crate::Result<InternalValue>>,
    T: core::borrow::Borrow<crate::table::Table>,
{
    fn next_back(&mut self) -> Option<Self::Item> {
        let item = self.inner.next_back();
        self.settle(item)
    }
}

impl<I: SkipKey, T> SkipKey for FilterPassReader<I, T> {
    /// A jump follows a version handed out, so the reader has settled: the
    /// versions it passes are no miss to count.
    fn skip_front_past(&mut self, key: &[u8]) -> crate::Result<bool> {
        self.inner.skip_front_past(key)
    }

    /// As [`Self::skip_front_past`]: the jump follows a version handed out.
    fn skip_back_to_newest(&mut self, key: &[u8], seqno: SeqNo) -> crate::Result<bool> {
        self.inner.skip_back_to_newest(key, seqno)
    }
}

/// The reader of `table` over `user_range` at `seqno`, which its filters did
/// not rule out: it counts a miss when `answer` says a filter was asked and
/// let the read through, the probe that miss belongs to.
fn table_reader<'a, T>(
    answer: FilterAnswer,
    table: T,
    user_range: (Bound<UserKey>, Bound<UserKey>),
    seqno: SeqNo,
) -> ScanBox<'a>
where
    T: core::borrow::Borrow<crate::table::Table> + Send + 'a,
{
    let reader = table.borrow().range_iter(user_range);
    let counts = answer.passed() && table.borrow().probe_stats().is_some();
    if counts {
        Box::new(Visible {
            inner: FilterPassReader {
                inner: reader,
                table,
                settled: false,
            },
            seqno,
        })
    } else {
        Box::new(Visible {
            inner: reader,
            seqno,
        })
    }
}

/// A source of a scan, a table's reader built only once the scan knows whether
/// the table is its only source and can be read without the indirection.
enum Source<'a> {
    Table(crate::table::Table, FilterAnswer),
    Boxed(ScanBox<'a>),
}

impl<'a> Source<'a> {
    fn into_boxed(self, user_range: &UserBounds, seqno: SeqNo) -> ScanBox<'a> {
        match self {
            Self::Table(table, answer) => table_reader(answer, table, user_range.clone(), seqno),
            Self::Boxed(iter) => iter,
        }
    }
}

/// The whole scan over a table that is its only source and holds one version
/// of each key: the rows its snapshot sees, but tombstones, in one pass over
/// the reader, then the range tombstones.
fn lone_table_scan<'a>(
    answer: FilterAnswer,
    table: crate::table::Table,
    user_range: UserBounds,
    seqno: SeqNo,
    range_tombstones: Vec<(RangeTombstone, SeqNo)>,
    comparator: SharedComparator,
) -> BoxedMerge<'a> {
    let reader = table.range(user_range);
    let counts = answer.passed() && table.probe_stats().is_some();
    let handed_out = move |item: &crate::Result<InternalValue>| match item {
        Ok(item) => seqno_filter(item.key.seqno, seqno) && !item.key.is_tombstone(),
        Err(_) => true,
    };
    if counts {
        finish_scan(
            FilterPassReader {
                inner: reader,
                table,
                settled: false,
            }
            .filter(handed_out),
            range_tombstones,
            comparator,
        )
    } else {
        finish_scan(reader.filter(handed_out), range_tombstones, comparator)
    }
}

impl TreeIter {
    #[expect(
        clippy::too_many_lines,
        reason = "create_range wires up multiple iterator sources, filters, and tombstone handling; splitting further would reduce clarity"
    )]
    pub fn create_range<K: AsRef<[u8]>, R: RangeBounds<K>>(
        guard: IterState,
        range: R,
        seqno: SeqNo,
    ) -> Self {
        Self::new(guard, |lock| {
            let user_range = (
                match range.start_bound() {
                    Bound::Included(key) => Bound::Included(UserKey::from(key.as_ref())),
                    Bound::Excluded(key) => Bound::Excluded(UserKey::from(key.as_ref())),
                    Bound::Unbounded => Bound::Unbounded,
                },
                match range.end_bound() {
                    Bound::Included(key) => Bound::Included(UserKey::from(key.as_ref())),
                    Bound::Excluded(key) => Bound::Excluded(UserKey::from(key.as_ref())),
                    Bound::Unbounded => Bound::Unbounded,
                },
            );

            let range = user_to_internal_bounds(&user_range);

            let mut iters: Vec<Source<'_>> = Vec::with_capacity(5);
            // Per source: whether it holds at most one version of any key,
            // which a lone source then needs no version grouping for.
            let mut one_version: Vec<bool> = Vec::with_capacity(5);
            // Each RT is paired with the per-source visibility cutoff so that
            // ephemeral memtable RTs use their own index_seqno instead of the
            // outer scan seqno (see issue #33).
            let mut all_range_tombstones: Vec<(RangeTombstone, SeqNo)> = Vec::new();
            // The tables of each run the range reaches, whose filters are
            // asked once the tombstones are known: a table a newer tombstone
            // wholly covers is not read, so its filter is not asked either.
            let mut candidates: Vec<Vec<crate::table::Table>> = Vec::new();
            let mut multi_runs = Vec::new();

            for run in lock
                .version
                .version
                .iter_levels()
                .flat_map(|lvl| lvl.iter())
            {
                match run.len() {
                    0 => {
                        // Do nothing
                    }
                    1 => {
                        #[expect(clippy::expect_used, reason = "we checked for length")]
                        let table = run.first().expect("should exist");

                        all_range_tombstones.extend(
                            table
                                .range_tombstones()
                                .iter()
                                .filter(|rt| {
                                    range_tombstone_overlaps_bounds(
                                        rt,
                                        &user_range,
                                        lock.comparator.as_ref(),
                                    )
                                })
                                .map(|rt| (rt.clone(), seqno)),
                        );

                        // Check key range overlap first (cheap metadata check) before
                        // running the O(rt_count) table-skip scan.
                        if table.check_key_range_overlap_cmp(
                            &(
                                user_range.0.as_ref().map(core::convert::AsRef::as_ref),
                                user_range.1.as_ref().map(core::convert::AsRef::as_ref),
                            ),
                            lock.comparator.as_ref(),
                        ) {
                            candidates.push(alloc::vec![table.clone()]);
                        }
                    }
                    _ => {
                        // Collect range tombstones from ALL tables in the run
                        // regardless of bloom filtering — they may affect keys
                        // in other tables/levels.
                        for table in run.iter() {
                            all_range_tombstones.extend(
                                table
                                    .range_tombstones()
                                    .iter()
                                    .filter(|rt| {
                                        range_tombstone_overlaps_bounds(
                                            rt,
                                            &user_range,
                                            lock.comparator.as_ref(),
                                        )
                                    })
                                    .map(|rt| (rt.clone(), seqno)),
                            );
                        }

                        // A prefix scan filters the tables of the run one by
                        // one by their prefix filters.
                        if lock.prefix_hash.is_some() {
                            let bounds = (
                                user_range.0.as_ref().map(core::convert::AsRef::as_ref),
                                user_range.1.as_ref().map(core::convert::AsRef::as_ref),
                            );

                            // Cheap key-range metadata check first; the
                            // filters are asked below.
                            candidates.push(
                                run.iter()
                                    .filter(|table| {
                                        table.check_key_range_overlap_cmp(
                                            &bounds,
                                            lock.comparator.as_ref(),
                                        )
                                    })
                                    .cloned()
                                    .collect(),
                            );
                        } else {
                            multi_runs.push(run.clone());
                        }
                    }
                }
            }

            // Sort SST-sourced RTs by start key for binary search in
            // table-skip below. This is intentionally a separate sort from
            // the full sort+dedup later: table-skip runs here (before memtable
            // RTs are collected), so only SST RTs are present. The later sort
            // covers the complete list. Both sorts are O(n log n) on their
            // respective subsets; the SST-only subset is typically small.
            all_range_tombstones
                .sort_unstable_by(|(a, _), (b, _)| lock.comparator.compare(&a.start, &b.start));

            // Table-skip: a table a range tombstone fully covers with a higher
            // seqno holds nothing visible, so it is not read (no I/O), and its
            // filter is not asked.
            //
            // Uses get_highest_kv_seqno() which excludes RT seqnos, so a
            // covering RT stored in the same table can trigger the skip.
            //
            // Binary search on sorted RT list: partition_point finds the first
            // RT with start > table_min; only the prefix [0..idx] can have
            // start <= table_min (required for fully_covers). key_range.max()
            // is inclusive; fully_covers checks max < rt.end (half-open), so
            // this is correct for inclusive upper bounds.
            let is_covered = |table: &crate::table::Table| {
                let table_min: &[u8] = table.metadata.key_range.min().as_ref();
                let table_max: &[u8] = table.metadata.key_range.max().as_ref();
                let table_kv_seqno = table.get_highest_kv_seqno();
                let candidate_end = all_range_tombstones.partition_point(|(rt, _)| {
                    lock.comparator.compare(&rt.start, table_min) != core::cmp::Ordering::Greater
                });
                all_range_tombstones
                    .iter()
                    .take(candidate_end)
                    .any(|(rt, cutoff)| {
                        rt.visible_at(*cutoff)
                            && rt.fully_covers_with(table_min, table_max, lock.comparator.as_ref())
                            && rt.seqno > table_kv_seqno
                    })
            };

            // A scan, counted once, and on every table it reads.
            lock.version.version.read_mix().note_scan();
            for run in candidates {
                let mut surviving: Vec<(crate::table::Table, FilterAnswer)> = run
                    .into_iter()
                    .filter(|table| !is_covered(table))
                    .filter_map(|table| {
                        let answer = filter_answer(lock.filter_query(), &table);
                        (answer != FilterAnswer::Absent).then_some((table, answer))
                    })
                    .collect();
                if surviving.is_empty() {
                    continue;
                }
                // Tables of a run are disjoint and in key order, and a table's
                // first key is a data key; only its last may be a range
                // tombstone's end a compaction widened it to. So a read range
                // reaching two or more holds keys of every one but perhaps the
                // first, which is read on its own to count a pass that finds
                // nothing; the others' passes are no false positives.
                let (first, answer) = surviving.remove(0);
                first.note_scan();
                one_version.push(holds_one_version_per_key(&first));
                iters.push(Source::Table(first, answer));
                match surviving.len() {
                    0 => {}
                    1 => {
                        if let Some((table, answer)) = surviving.pop() {
                            table.note_scan();
                            one_version.push(holds_one_version_per_key(&table));
                            iters.push(Source::Table(table, answer));
                        }
                    }
                    _ => {
                        let rest = surviving.into_iter().map(|(table, _)| table).collect();
                        // Two or more tables are left, so Run::new cannot
                        // return None (only empty vecs yield None).
                        #[expect(
                            clippy::expect_used,
                            reason = "Run::new returns None only for empty vecs"
                        )]
                        let new_run = Run::new(rest).expect("non-empty surviving tables");
                        multi_runs.push(Arc::new(new_run));
                    }
                }
            }

            for run in multi_runs {
                // The tables of a run hold disjoint keys, so the run holds one
                // version of a key when each of its tables does.
                let run_one_version = run.iter().all(holds_one_version_per_key);
                if let Some(reader) =
                    RunReader::new_cmp(run, user_range.clone(), lock.comparator.as_ref())
                        .map(RunReader::counting_scans)
                {
                    one_version.push(run_one_version);
                    iters.push(Source::Boxed(Box::new(Visible {
                        inner: reader,
                        seqno,
                    })));
                }
            }

            // Sealed memtables
            let table_source_count = iters.len();
            for memtable in lock.version.sealed_memtables.iter() {
                all_range_tombstones.extend(
                    memtable
                        .range_tombstones_sorted()
                        .into_iter()
                        .filter(|rt| {
                            range_tombstone_overlaps_bounds(
                                rt,
                                &user_range,
                                lock.comparator.as_ref(),
                            )
                        })
                        .map(|rt| (rt, seqno)),
                );

                // A memtable holding no row adds nothing to merge; its range
                // tombstones were collected above and still apply.
                if memtable.items.is_empty() {
                    continue;
                }

                one_version.push(false);
                iters.push(Source::Boxed(memtable_source(
                    memtable,
                    range.clone(),
                    seqno,
                )));
            }

            // Active memtable
            {
                all_range_tombstones.extend(
                    lock.version
                        .active_memtable
                        .range_tombstones_sorted()
                        .into_iter()
                        .filter(|rt| {
                            range_tombstone_overlaps_bounds(
                                rt,
                                &user_range,
                                lock.comparator.as_ref(),
                            )
                        })
                        .map(|rt| (rt, seqno)),
                );

                // Empty when the scan is built, it holds no row the snapshot
                // can see: a write is published as visible only once it is in
                // the memtable, so one below the scan's seqno is already there.
                // A write backdated below the seqno after the scan began is
                // one the live walk could miss anyway.
                if !lock.version.active_memtable.items.is_empty() {
                    one_version.push(false);
                    iters.push(Source::Boxed(memtable_source(
                        &lock.version.active_memtable,
                        range.clone(),
                        seqno,
                    )));
                }
            }

            if let Some((mt, eph_seqno)) = &lock.ephemeral {
                all_range_tombstones.extend(
                    mt.range_tombstones_sorted()
                        .into_iter()
                        .filter(|rt| {
                            range_tombstone_overlaps_bounds(
                                rt,
                                &user_range,
                                lock.comparator.as_ref(),
                            )
                        })
                        .map(|rt| (rt, *eph_seqno)),
                );

                if !mt.items.is_empty() {
                    one_version.push(false);
                    iters.push(Source::Boxed(memtable_source(mt, range, *eph_seqno)));
                }
            }

            // Tie-break order: the merger resolves identical (key, seqno)
            // entries to the LOWEST source index, so sources must run NEWEST
            // FIRST — the point read's order (ephemeral overlay, active
            // memtable, sealed newest first, then tables). The blocks above
            // push tables first and memtables oldest-first (their RT
            // collection deliberately trails the table-skip sort, which wants
            // the SST-only subset), so reverse the memtable segment and
            // rotate it to the front rather than reordering the collection.
            if let Some(memtable_segment) = iters.get_mut(table_source_count..) {
                memtable_segment.reverse();
            }
            iters.rotate_left(table_source_count);

            // A lone source needs no merge, and a lone source holding one
            // version of each key needs no version grouping either, unless a
            // merge operator has operands to resolve.
            let lone = (iters.len() == 1).then(|| iters.pop()).flatten();
            let lone_one_version = one_version.len() == 1 && one_version.first() == Some(&true);
            let groups_versions = !(lone_one_version && lock.merge_operator.is_none());

            // Clone needed: MvccStream uses the RT set for merge suppression,
            // while RangeTombstoneFilter below consumes it for post-merge
            // filtering. An Arc<[_]> could avoid the copy if RT sets grow large.
            let mvcc_tombstones = if lone.is_some() && !groups_versions {
                Vec::new()
            } else {
                all_range_tombstones.clone()
            };

            // Deduplicate: MultiWriter rotation copies the same RTs into each
            // output table, so collected tombstones can contain duplicates.
            // When the same RT appears from different sources with different
            // cutoffs (e.g., persisted SST + ephemeral), keep the max cutoff
            // so the RT stays visible if ANY source's snapshot includes it.
            all_range_tombstones
                .sort_by(|a, b| a.0.cmp_with_comparator(&b.0, lock.comparator.as_ref()));
            all_range_tombstones.dedup_by(|a, b| {
                if a.0 == b.0 {
                    // dedup_by passes (a=later, b=earlier); b survives, a is
                    // removed.  Merge a's cutoff into the surviving b.
                    b.1 = b.1.max(a.1);
                    true
                } else {
                    false
                }
            });

            let comparator = lock.comparator.clone();
            match lone {
                Some(Source::Table(table, answer)) if !groups_versions => lone_table_scan(
                    answer,
                    table,
                    user_range,
                    seqno,
                    all_range_tombstones,
                    comparator,
                ),
                Some(source) if !groups_versions => finish_scan(
                    source.into_boxed(&user_range, seqno).filter(is_live),
                    all_range_tombstones,
                    comparator,
                ),
                Some(source) => finish_scan(
                    MvccStream::new_with_comparator(
                        source.into_boxed(&user_range, seqno),
                        lock.merge_operator.clone(),
                        lock.comparator.clone(),
                    )
                    .jumping()
                    .with_value_log(lock.value_log())
                    .with_range_tombstones(mvcc_tombstones)
                    .filter(is_live),
                    all_range_tombstones,
                    comparator,
                ),
                None => finish_scan(
                    MvccStream::new_with_comparator(
                        build_seeking(
                            iters
                                .into_iter()
                                .map(|source| source.into_boxed(&user_range, seqno))
                                .collect(),
                            lock.comparator.clone(),
                        ),
                        lock.merge_operator.clone(),
                        lock.comparator.clone(),
                    )
                    .jumping()
                    .with_value_log(lock.value_log())
                    .with_range_tombstones(mvcc_tombstones)
                    .filter(is_live),
                    all_range_tombstones,
                    comparator,
                ),
            }
        })
    }
}

/// Whether a resolved row is handed out: a tombstone ends its key instead.
fn is_live(item: &crate::Result<InternalValue>) -> bool {
    match item {
        Ok(value) => !value.key.is_tombstone(),
        Err(_) => true,
    }
}

/// Whether `table` holds at most one version of any key: the distinct keys
/// its writer counted are all its rows.
fn holds_one_version_per_key(table: &crate::table::Table) -> bool {
    table.metadata.key_count == Some(table.metadata.item_count)
}

/// The scan's last stage: rows a range tombstone deletes are dropped, a pass
/// left out when no tombstone is visible at its per-source cutoff.
fn finish_scan<'a, I>(
    iter: I,
    range_tombstones: Vec<(RangeTombstone, SeqNo)>,
    comparator: SharedComparator,
) -> BoxedMerge<'a>
where
    I: DoubleEndedIterator<Item = crate::Result<InternalValue>> + Send + 'a,
{
    // Each RT carries the seqno of its originating source, so the check is
    // per-RT rather than global.
    if range_tombstones
        .iter()
        .all(|(rt, cutoff)| !rt.visible_at(*cutoff))
    {
        Box::new(iter)
    } else {
        Box::new(RangeTombstoneFilter::new_with_comparator(
            iter,
            range_tombstones,
            comparator,
        ))
    }
}

/// User-key bounds for a scan; the seekable iterator rebuilds its merge
/// pipeline for any sub-range of the union without recollecting sources.
type UserBounds = (Bound<UserKey>, Bound<UserKey>);

/// Translate user-key bounds into the internal-key bounds the per-source readers
/// expect. A bound is a search needle, not a stored key, so a bound longer than
/// any key can be is taken as it is.
pub(crate) fn user_to_internal_bounds(
    user: &UserBounds,
) -> (Bound<InternalKey>, Bound<InternalKey>) {
    internal_bounds(&user.0, &user.1)
}

/// [`user_to_internal_bounds`] of the bounds `lo` and `hi`: every version of a
/// bound key is inside an inclusive bound and outside an exclusive one.
pub(crate) fn internal_bounds(
    lo: &Bound<UserKey>,
    hi: &Bound<UserKey>,
) -> (Bound<InternalKey>, Bound<InternalKey>) {
    // Clone the `UserKey` (a reference-counted `Slice`: an `Arc` bump, no heap
    // copy) and hand it to `InternalKey::bound` by value — `UserKey:
    // Into<UserKey>` is the identity, so the key bytes are reused. Passing
    // `key.as_ref()` (`&[u8]`) instead would force a fresh heap copy of the
    // bytes on every reposition, which the in-place seek loop must avoid.
    let lo = match lo {
        // The first version of the key: versions sort by descending seqno.
        Bound::Included(key) => Bound::Included(InternalKey::bound(
            key.clone(),
            SeqNo::MAX,
            crate::ValueType::Tombstone,
        )),
        Bound::Excluded(key) => Bound::Excluded(InternalKey::bound(
            key.clone(),
            0,
            crate::ValueType::Tombstone,
        )),
        Bound::Unbounded => Bound::Unbounded,
    };
    // The upper end reaches the key's LAST version, the reverse of the lower:
    // for `..Excluded(abdef)` over `a@7, abc@5, abc@4, abc@3, abdef@6`, the
    // range is `a@7` through `abc@3`.
    let hi = match hi {
        Bound::Included(key) => {
            Bound::Included(InternalKey::bound(key.clone(), 0, crate::ValueType::Value))
        }
        Bound::Excluded(key) => Bound::Excluded(InternalKey::bound(
            key.clone(),
            SeqNo::MAX,
            crate::ValueType::Value,
        )),
        Bound::Unbounded => Bound::Unbounded,
    };
    (lo, hi)
}

/// SSTs / runs / range-tombstones overlapping the union range, collected once.
///
/// Rebuilding the merge pipeline for a sub-range (a seek, or one batch interval)
/// reuses these — Phase 1 (level iteration + bloom + RT collection) is the
/// ~100µs setup that [`SeekableTreeIter`] amortizes across repositions.
#[derive(Clone)]
struct CollectedSources {
    single_tables: Vec<crate::table::Table>,
    multi_runs: Vec<Arc<Run<crate::table::Table>>>,
    range_tombstones: Vec<(RangeTombstone, SeqNo)>,
    union: UserBounds,
}

/// Phase 1: collect every source overlapping `union` once. The table-skip
/// optimization and the prefix/key-hash sub-filtering used by
/// [`TreeIter::create_range`] are intentionally omitted: this is a plain range
/// scan (no prefix/point pipeline), and skipping table-skip only ever reads a
/// table a tombstone fully covers — correct, just marginally less optimal.
fn collect_sources(state: &IterState, union: UserBounds, seqno: SeqNo) -> CollectedSources {
    let mut single_tables: Vec<crate::table::Table> = Vec::new();
    let mut multi_runs: Vec<Arc<Run<crate::table::Table>>> = Vec::new();
    let mut rts: Vec<(RangeTombstone, SeqNo)> = Vec::new();

    let bounds_ref = (
        union.0.as_ref().map(core::convert::AsRef::as_ref),
        union.1.as_ref().map(core::convert::AsRef::as_ref),
    );

    for run in state
        .version
        .version
        .iter_levels()
        .flat_map(|lvl| lvl.iter())
    {
        match run.len() {
            0 => {}
            1 => {
                #[expect(clippy::expect_used, reason = "len checked")]
                let table = run.first().expect("len == 1");
                rts.extend(
                    table
                        .range_tombstones()
                        .iter()
                        .filter(|rt| {
                            range_tombstone_overlaps_bounds(rt, &union, state.comparator.as_ref())
                        })
                        .map(|rt| (rt.clone(), seqno)),
                );
                if table.check_key_range_overlap_cmp(&bounds_ref, state.comparator.as_ref())
                    && bloom_passes(state, table)
                {
                    single_tables.push(table.clone());
                }
            }
            _ => {
                for table in run.iter() {
                    rts.extend(
                        table
                            .range_tombstones()
                            .iter()
                            .filter(|rt| {
                                range_tombstone_overlaps_bounds(
                                    rt,
                                    &union,
                                    state.comparator.as_ref(),
                                )
                            })
                            .map(|rt| (rt.clone(), seqno)),
                    );
                }
                multi_runs.push(run.clone());
            }
        }
    }

    let mut collect_mt_rts = |iter: alloc::vec::Vec<RangeTombstone>, cutoff: SeqNo| {
        rts.extend(
            iter.into_iter()
                .filter(|rt| range_tombstone_overlaps_bounds(rt, &union, state.comparator.as_ref()))
                .map(|rt| (rt, cutoff)),
        );
    };
    for memtable in state.version.sealed_memtables.iter() {
        collect_mt_rts(memtable.range_tombstones_sorted(), seqno);
    }
    collect_mt_rts(
        state.version.active_memtable.range_tombstones_sorted(),
        seqno,
    );
    if let Some((mt, eph_seqno)) = &state.ephemeral {
        collect_mt_rts(mt.range_tombstones_sorted(), *eph_seqno);
    }

    rts.sort_by(|a, b| a.0.cmp_with_comparator(&b.0, state.comparator.as_ref()));
    rts.dedup_by(|a, b| {
        if a.0 == b.0 {
            b.1 = b.1.max(a.1);
            true
        } else {
            false
        }
    });

    CollectedSources {
        single_tables,
        multi_runs,
        range_tombstones: rts,
        union,
    }
}

/// Named tombstone-skip wrapper. The non-seekable read path uses an inline
/// `.filter(|x| !is_tombstone)` closure (unnameable), but the seekable pipeline
/// must stay a concrete, [`Reseekable`] type, so the same drop-resolved-
/// tombstones step is expressed as this struct. Errors pass through unchanged;
/// the reposition is stateless, so it just forwards to the inner layer.
struct TombstoneSkip<I> {
    inner: I,
}

impl<I: Iterator<Item = crate::Result<InternalValue>>> Iterator for TombstoneSkip<I> {
    type Item = crate::Result<InternalValue>;

    fn next(&mut self) -> Option<Self::Item> {
        loop {
            match self.inner.next()? {
                Ok(value) if value.key.is_tombstone() => {}
                other => return Some(other),
            }
        }
    }
}

impl<I: DoubleEndedIterator<Item = crate::Result<InternalValue>>> DoubleEndedIterator
    for TombstoneSkip<I>
{
    fn next_back(&mut self) -> Option<Self::Item> {
        loop {
            match self.inner.next_back()? {
                Ok(value) if value.key.is_tombstone() => {}
                other => return Some(other),
            }
        }
    }
}

impl<I: Reseekable> Reseekable for TombstoneSkip<I> {
    fn reseek(&mut self, ctx: &ReseekCtx) {
        self.inner.reseek(ctx);
    }
}

/// Leaf source over a single SST, re-positioned in place by re-seeking the
/// reader's owned index iterator (no new reader, no `Arc` re-clone).
struct TableLeaf {
    table: crate::table::Table,
    iter: crate::table::iter::Iter,
    seqno: SeqNo,
    /// Whether the table was counted as reached: on the first row read from
    /// it, seen by the snapshot or not, since a source opened over the whole
    /// keyspace and repositioned to intervals may never be read at all.
    noted: bool,
}

impl TableLeaf {
    fn new(table: crate::table::Table, user_range: UserBounds, seqno: SeqNo) -> Self {
        let iter = table.range_iter(user_range);
        Self {
            table,
            iter,
            seqno,
            noted: false,
        }
    }

    /// Counts the scan on the table the first time a row is read from it.
    fn note_read(&mut self) {
        if !self.noted {
            self.table.note_scan();
            self.noted = true;
        }
    }

    fn next_filtered(&mut self) -> Option<IterItem> {
        loop {
            match self.iter.next()? {
                Ok(value) => {
                    self.note_read();
                    if seqno_filter(value.key.seqno, self.seqno) {
                        return Some(Ok(value));
                    }
                }
                Err(e) => return Some(Err(e)),
            }
        }
    }

    fn next_back_filtered(&mut self) -> Option<IterItem> {
        loop {
            match self.iter.next_back()? {
                Ok(value) => {
                    self.note_read();
                    if seqno_filter(value.key.seqno, self.seqno) {
                        return Some(Ok(value));
                    }
                }
                Err(e) => return Some(Err(e)),
            }
        }
    }
}

/// Leaf source over a multi-table run. Re-positioned by recomputing the run's
/// overlapping-table window; the boundary readers re-open lazily on the next
/// pull (run-internal residual allocation, unlike the alloc-free table / memtable
/// leaves — multi-table runs are L1+ and less common on the seekable path).
struct RunLeaf {
    reader: RunReader,
    comparator: SharedComparator,
    seqno: SeqNo,
}

impl RunLeaf {
    fn new(
        run: Arc<Run<crate::table::Table>>,
        user_range: UserBounds,
        seqno: SeqNo,
        comparator: SharedComparator,
    ) -> Option<Self> {
        // Repositioned over every range of a batch, which counts as one scan.
        let reader =
            RunReader::new_cmp(run, user_range, comparator.as_ref())?.counting_each_table_once();
        Some(Self {
            reader,
            comparator,
            seqno,
        })
    }

    fn next_filtered(&mut self) -> Option<IterItem> {
        loop {
            match self.reader.next()? {
                Ok(value) if seqno_filter(value.key.seqno, self.seqno) => return Some(Ok(value)),
                Ok(_) => {}
                Err(e) => return Some(Err(e)),
            }
        }
    }

    fn next_back_filtered(&mut self) -> Option<IterItem> {
        loop {
            match self.reader.next_back()? {
                Ok(value) if seqno_filter(value.key.seqno, self.seqno) => return Some(Ok(value)),
                Ok(_) => {}
                Err(e) => return Some(Err(e)),
            }
        }
    }
}

/// Leaf source over one memtable (active / sealed / ephemeral). Re-positioned by
/// recreating the skiplist range cursor (a `seek_ge` pointer-walk, no heap
/// allocation). Borrows the memtable from the owning [`IterState`] for `'a`.
struct MemtableLeaf<'a> {
    mt: &'a Memtable,
    range: crate::memtable::skiplist::Range<'a>,
    seqno: SeqNo,
}

impl<'a> MemtableLeaf<'a> {
    fn new(
        mt: &'a Memtable,
        internal: (Bound<InternalKey>, Bound<InternalKey>),
        seqno: SeqNo,
    ) -> Self {
        let range = mt.items.range(internal);
        Self { mt, range, seqno }
    }

    fn next_filtered(&mut self) -> Option<IterItem> {
        loop {
            let entry = self.range.next()?;
            let value = InternalValue {
                key: entry.key(),
                value: entry.value(),
            };
            if seqno_filter(value.key.seqno, self.seqno) {
                return Some(Ok(value));
            }
        }
    }

    fn next_back_filtered(&mut self) -> Option<IterItem> {
        loop {
            let entry = self.range.next_back()?;
            let value = InternalValue {
                key: entry.key(),
                value: entry.value(),
            };
            if seqno_filter(value.key.seqno, self.seqno) {
                return Some(Ok(value));
            }
        }
    }
}

/// One reseekable leaf of the seekable merge pipeline.
///
/// All three variants self-coordinate their forward/back cursors over a single
/// shrinking window (SST index span, run table window, skiplist node range), so
/// the enum is a [`CoherentMergeSource`]: mixed forward/backward consumption
/// never yields an item twice. The [`MergeSource::seek`] hook is a no-op — the
/// seekable path repositions via [`Reseekable`] (which carries both new bounds),
/// not the single-target merge-source seek.
enum SeekableLeaf<'a> {
    // The SST reader and run reader are large (owned index iterator / per-table
    // readers); box them so the enum (held one-per-source in a `Vec`) is sized to
    // the small memtable variant. This is a one-time per-leaf allocation at build,
    // not a per-seek cost.
    Table(Box<TableLeaf>),
    Run(Box<RunLeaf>),
    Memtable(MemtableLeaf<'a>),
}

impl MergeSource for SeekableLeaf<'_> {
    fn next(&mut self) -> Option<IterItem> {
        match self {
            SeekableLeaf::Table(l) => l.next_filtered(),
            SeekableLeaf::Run(l) => l.next_filtered(),
            SeekableLeaf::Memtable(l) => l.next_filtered(),
        }
    }

    fn next_back(&mut self) -> Option<IterItem> {
        match self {
            SeekableLeaf::Table(l) => l.next_back_filtered(),
            SeekableLeaf::Run(l) => l.next_back_filtered(),
            SeekableLeaf::Memtable(l) => l.next_back_filtered(),
        }
    }

    fn seek(&mut self, _target: &InternalKey) -> crate::Result<()> {
        // No-op: the seekable pipeline repositions via `Reseekable::reseek`
        // (which carries both the new lower AND upper bound). The single-target
        // `MergeSource::seek` is never invoked on this path; the leaf cursors
        // self-coordinate direction switches without it.
        Ok(())
    }
}

impl CoherentMergeSource for SeekableLeaf<'_> {}

impl SkipKey for SeekableLeaf<'_> {
    fn skip_front_past(&mut self, key: &[u8]) -> crate::Result<bool> {
        match self {
            SeekableLeaf::Table(l) => l.iter.skip_past(key).map(|_| true),
            SeekableLeaf::Run(l) => l.reader.skip_front_past(key),
            SeekableLeaf::Memtable(l) => {
                l.range.skip_front_past(key);
                Ok(true)
            }
        }
    }

    /// The newest version the leaf's snapshot sees, as its filter passes.
    fn skip_back_to_newest(&mut self, key: &[u8], seqno: SeqNo) -> crate::Result<bool> {
        match self {
            SeekableLeaf::Table(l) => l.iter.skip_back_to_newest(key, seqno.min(l.seqno)),
            SeekableLeaf::Run(l) => l.reader.skip_back_to_newest(key, seqno.min(l.seqno)),
            SeekableLeaf::Memtable(l) => Ok(l.range.skip_back_to_newest(key, seqno.min(l.seqno))),
        }
    }
}

impl Reseekable for SeekableLeaf<'_> {
    fn reseek(&mut self, ctx: &ReseekCtx) {
        match self {
            SeekableLeaf::Table(l) => l.table.reseek_range(&mut l.iter, ctx.user.clone()),
            SeekableLeaf::Run(l) => l.reader.reseek(ctx.user.clone(), l.comparator.as_ref()),
            SeekableLeaf::Memtable(l) => {
                l.range = l.mt.items.range(ctx.internal.clone());
            }
        }
    }
}

/// The concrete, [`Reseekable`] seekable merge pipeline: loser-tree merger ->
/// MVCC resolution -> drop-resolved-tombstones -> range-tombstone suppression.
/// Built once over the union range; every reposition reseeks it in place.
type SeekPipeline<'a> = RangeTombstoneFilter<
    TombstoneSkip<
        MvccStream<
            SeekingMerger<SeekableLeaf<'a>, SharedComparator>,
            Option<crate::mvcc_stream::ValueLog<'a>>,
            crate::mvcc_stream::Jump,
        >,
    >,
>;

/// Phase 2: build the [`SeekPipeline`] for the sub-range `[lower, upper)` from
/// already-collected sources. Runs ONCE per iterator (at construction); every
/// later reposition reseeks the returned pipeline in place rather than rebuilding
/// it. The per-source readers reuse the tested seek-to-start path (no block I/O
/// until the first `next`).
fn build_seek_pipeline<'a>(
    state: &'a IterState,
    collected: &CollectedSources,
    lower: Bound<UserKey>,
    upper: Bound<UserKey>,
    seqno: SeqNo,
) -> SeekPipeline<'a> {
    let user_range: UserBounds = (lower, upper);
    let internal = user_to_internal_bounds(&user_range);

    let mut sources: Vec<SeekableLeaf<'a>> =
        Vec::with_capacity(collected.single_tables.len() + collected.multi_runs.len() + 3);

    // Tie-break order: the merger resolves identical (key, seqno) entries to
    // the LOWEST source index, so sources go NEWEST FIRST — the point read's
    // order (ephemeral overlay, active memtable, sealed newest first, then
    // tables). A tied entry otherwise resolves to a source the point read
    // shadows, and a scan serves a value `get` does not.
    if let Some((mt, eph_seqno)) = &state.ephemeral {
        sources.push(SeekableLeaf::Memtable(MemtableLeaf::new(
            mt,
            internal.clone(),
            *eph_seqno,
        )));
    }
    sources.push(SeekableLeaf::Memtable(MemtableLeaf::new(
        &state.version.active_memtable,
        internal.clone(),
        seqno,
    )));
    for memtable in state.version.sealed_memtables.iter().rev() {
        sources.push(SeekableLeaf::Memtable(MemtableLeaf::new(
            memtable,
            internal.clone(),
            seqno,
        )));
    }
    // Built once per iterator: the scan is counted here, once.
    state.version.version.read_mix().note_scan();
    for table in &collected.single_tables {
        sources.push(SeekableLeaf::Table(Box::new(TableLeaf::new(
            table.clone(),
            user_range.clone(),
            seqno,
        ))));
    }
    for run in &collected.multi_runs {
        if let Some(leaf) = RunLeaf::new(
            run.clone(),
            user_range.clone(),
            seqno,
            state.comparator.clone(),
        ) {
            sources.push(SeekableLeaf::Run(Box::new(leaf)));
        }
    }

    let merged = SeekingMerger::new(sources, state.comparator.clone());
    let mvcc = MvccStream::new_with_comparator(
        merged,
        state.merge_operator.clone(),
        state.comparator.clone(),
    )
    .jumping()
    .with_value_log(state.value_log())
    .with_range_tombstones(collected.range_tombstones.clone());
    let skip = TombstoneSkip { inner: mvcc };
    // Always wrap in the range-tombstone filter: with an empty or all-invisible
    // tombstone set it suppresses nothing (a per-item visibility check), so the
    // result matches the non-seekable path's fast-path-or-filter branch while
    // staying a single concrete reseekable type.
    RangeTombstoneFilter::new_with_comparator(
        skip,
        collected.range_tombstones.clone(),
        state.comparator.clone(),
    )
}

/// Owner half of [`SeekableTreeIter`]'s self-cell.
///
/// Holds the version snapshot, the collected sources (shared via `Arc` so a
/// reposition is a cheap refcount bump), and the current sub-range bounds the
/// dependent pipeline was built for.
pub struct SeekableOwner {
    state: IterState,
    collected: Arc<CollectedSources>,
    seqno: SeqNo,
    lower: Bound<UserKey>,
    upper: Bound<UserKey>,
}

self_cell!(
    /// Internal self-referential merge cell: owns the collected sources and
    /// borrows them into the Phase-2 merge pipeline. Wrapped by
    /// [`SeekableTreeIter`], which adds the seek API and a lookahead buffer.
    ///
    /// Built ONCE; [`Self::reseek`] moves the leaf cursors in place, so the
    /// merge pipeline is never reconstructed across repositions.
    struct SeekableCell {
        owner: SeekableOwner,

        #[covariant]
        dependent: SeekPipeline,
    }
);

impl SeekableCell {
    fn build(owner: SeekableOwner) -> Self {
        Self::new(owner, |o| {
            build_seek_pipeline(
                &o.state,
                &o.collected,
                o.lower.clone(),
                o.upper.clone(),
                o.seqno,
            )
        })
    }

    /// Re-position the borrowed pipeline in place to the bounds in `ctx`,
    /// without rebuilding it. The owner (version snapshot + collected sources)
    /// is untouched.
    fn reseek(&mut self, ctx: &ReseekCtx) {
        self.with_dependent_mut(|_, pipeline| pipeline.reseek(ctx));
    }
}

impl Iterator for SeekableCell {
    type Item = crate::Result<InternalValue>;

    fn next(&mut self) -> Option<Self::Item> {
        self.with_dependent_mut(|_, iter| iter.next())
    }
}

impl DoubleEndedIterator for SeekableCell {
    fn next_back(&mut self) -> Option<Self::Item> {
        self.with_dependent_mut(|_, iter| iter.next_back())
    }
}

/// A range iterator that can reposition (seek) in place without reopening
/// per-SST readers or rebuilding the merge stack.
///
/// Built once over a union range; [`Self::seek_to`], [`Self::seek_to_for_prev`],
/// and `Self::reposition` move the leaf cursors in place (SST index re-seek /
/// skiplist `seek_ge` / run-window recompute) while reusing the loser-tree
/// merger, MVCC stream, and tombstone filter, so a tight seek loop does not
/// reconstruct the pipeline.
///
/// A one-item lookahead buffer backs [`Self::peek_key`], which reads the next
/// key without consuming it (a leapfrog join takes the max of several
/// iterators' current keys before deciding where to seek next).
pub struct SeekableTreeIter {
    cell: SeekableCell,
    /// `None` = nothing peeked; `Some(None)` = peeked past the end;
    /// `Some(Some(item))` = buffered front item (the std `Peekable` shape).
    #[expect(
        clippy::option_option,
        reason = "std `Peekable` buffer shape: outer None = not yet peeked, \
                  inner None = peeked past the end, Some(Some) = buffered item"
    )]
    peeked: Option<Option<crate::Result<InternalValue>>>,
}

impl SeekableTreeIter {
    /// Open a seekable iterator over `[union_lower, union_upper)`. The source
    /// collection (Phase 1) runs once here; subsequent repositions reuse it.
    #[must_use]
    pub fn create(
        state: IterState,
        union_lower: Bound<UserKey>,
        union_upper: Bound<UserKey>,
        seqno: SeqNo,
    ) -> Self {
        let collected = Arc::new(collect_sources(
            &state,
            (union_lower.clone(), union_upper.clone()),
            seqno,
        ));
        Self {
            cell: SeekableCell::build(SeekableOwner {
                state,
                collected,
                seqno,
                lower: union_lower,
                upper: union_upper,
            }),
            peeked: None,
        }
    }

    /// Re-position the merge pipeline to the sub-range `[lower, upper)` IN PLACE:
    /// the loser-tree merger, MVCC stream, and range-tombstone filter are reused;
    /// only the leaf cursors move (an SST index re-seek / skiplist `seek_ge` /
    /// run-window recompute). No merge-stack reconstruction, no source re-collection
    /// (no block I/O until the next `next`/`next_back`).
    pub(crate) fn reposition(&mut self, lower: Bound<UserKey>, upper: Bound<UserKey>) {
        let user_range: UserBounds = (lower, upper);
        let internal = user_to_internal_bounds(&user_range);
        let ctx = ReseekCtx {
            user: user_range,
            internal,
        };
        self.cell.reseek(&ctx);
        // The lookahead came from the old position; drop it.
        self.peeked = None;
    }

    /// Reposition so the next [`Iterator::next`] yields the first entry with
    /// user key `>= key` (`RocksDB` `Seek`).
    ///
    /// The new lower bound is clamped to the iterator's collected window: Phase 1
    /// only gathered sources overlapping the original range, so seeking below the
    /// window lower would scan outside it (leaking rows below the window, or
    /// missing rows from sources that were skipped during collection). A `key`
    /// below the window lower positions at the window start; the upper bound
    /// stays the window upper.
    pub fn seek_to(&mut self, key: &[u8]) {
        let (lower, upper) = {
            let union = &self.cell.borrow_owner().collected.union;
            let lower = match &union.0 {
                Bound::Included(floor) if floor.as_ref() > key => union.0.clone(),
                Bound::Excluded(floor) if floor.as_ref() >= key => union.0.clone(),
                _ => Bound::Included(UserKey::from(key)),
            };
            (lower, union.1.clone())
        };
        self.reposition(lower, upper);
    }

    /// Reposition so the next [`DoubleEndedIterator::next_back`] yields the last
    /// entry with user key `<= key` (`RocksDB` `SeekForPrev`).
    ///
    /// The new upper bound is clamped to the iterator's collected window (see
    /// [`seek_to`](Self::seek_to) for why): a `key` above the window upper
    /// positions at the window end; the lower bound stays the window lower.
    pub fn seek_to_for_prev(&mut self, key: &[u8]) {
        let (lower, upper) = {
            let union = &self.cell.borrow_owner().collected.union;
            let upper = match &union.1 {
                Bound::Included(ceil) if ceil.as_ref() < key => union.1.clone(),
                Bound::Excluded(ceil) if ceil.as_ref() <= key => union.1.clone(),
                _ => Bound::Included(UserKey::from(key)),
            };
            (union.0.clone(), upper)
        };
        self.reposition(lower, upper);
    }

    /// The version snapshot this iterator reads from. Used by KV-separated trees
    /// to resolve blob handles against the same snapshot the keys came from.
    #[must_use]
    pub fn version(&self) -> crate::version::Version {
        self.cell.borrow_owner().state.version.version.clone()
    }

    /// Return the user key the next [`Iterator::next`] would yield, without
    /// consuming it. Buffers one item; cleared by any reposition / seek.
    ///
    /// A successful key is cloned (cheap: `UserKey` is reference-counted) and
    /// the item stays buffered for `next`, so `peek_key()` followed by `next()`
    /// observes the SAME entry.
    ///
    /// # Error is a consuming peek
    ///
    /// `crate::Error` is not `Clone`, so a `peek_key()` that returns `Err(...)`
    /// CONSUMES the failing position: the error is moved out of the buffer and
    /// the iterator is logically advanced past it. A following `next()` yields
    /// the entry AFTER the error, NOT the error again. Treat a peeked `Err` as
    /// you would a consumed one (surface / propagate it; do not also expect
    /// `next()` to re-report it). The leapfrog / zig-zag join callers this
    /// method is built for read `peek_key()`, propagate any `Err`, and never
    /// fall through to a redundant `next()` on the failing iterator, so the
    /// distinction does not bite them; it is documented here for any caller that
    /// treats `peek_key` as a pure lookahead.
    pub fn peek_key(&mut self) -> Option<crate::Result<UserKey>> {
        if self.peeked.is_none() {
            self.peeked = Some(self.cell.next());
        }
        if matches!(self.peeked, Some(Some(Err(_)))) {
            return match self.peeked.take() {
                Some(Some(Err(e))) => Some(Err(e)),
                _ => None,
            };
        }
        match &self.peeked {
            Some(Some(Ok(item))) => Some(Ok(item.key.user_key.clone())),
            _ => None,
        }
    }
}

impl Iterator for SeekableTreeIter {
    type Item = crate::Result<InternalValue>;

    fn next(&mut self) -> Option<Self::Item> {
        match self.peeked.take() {
            Some(buffered) => buffered,
            None => self.cell.next(),
        }
    }
}

impl DoubleEndedIterator for SeekableTreeIter {
    fn next_back(&mut self) -> Option<Self::Item> {
        // The buffer holds the FRONT item, so pull from the back first and only
        // fall back to the buffered front once the back is exhausted.
        match self.cell.next_back() {
            Some(item) => Some(item),
            None => self.peeked.take().flatten(),
        }
    }
}

/// Forward scan over a (possibly lazily-produced) sequence of disjoint,
/// ascending sub-ranges, reusing a single [`SeekableTreeIter`].
///
/// Each interval is served by repositioning the shared pipeline — the per-SST
/// setup is paid once (when the `SeekableTreeIter` is created over the union)
/// and amortized across every interval, rather than reopening readers per
/// interval.
///
/// The interval source is pulled on demand, so intervals may be generated
/// dynamically (the next interval can depend on rows already returned).
pub struct BatchRangeScan<I> {
    iter: SeekableTreeIter,
    intervals: I,
    /// Whether the shared iterator is currently positioned on an interval that
    /// may still yield rows.
    primed: bool,
}

impl<I: Iterator<Item = (Bound<UserKey>, Bound<UserKey>)>> BatchRangeScan<I> {
    /// Build a batch scan from a seekable iterator opened over the union range
    /// and a (lazy) sequence of `[lower, upper)` sub-ranges to visit in order.
    pub fn new(iter: SeekableTreeIter, intervals: I) -> Self {
        Self {
            iter,
            intervals,
            primed: false,
        }
    }
}

impl<I: Iterator<Item = (Bound<UserKey>, Bound<UserKey>)>> Iterator for BatchRangeScan<I> {
    type Item = crate::Result<InternalValue>;

    fn next(&mut self) -> Option<Self::Item> {
        loop {
            if self.primed {
                if let Some(item) = self.iter.next() {
                    return Some(item);
                }
                self.primed = false;
            }
            let (lower, upper) = self.intervals.next()?;
            self.iter.reposition(lower, upper);
            self.primed = true;
        }
    }
}

mod point_merge;
pub(crate) use point_merge::{MergeRead, MergeTable, MergeWalk, TableFilter, resolve_point_merge};

#[cfg(test)]
mod tests;