coordinode-lsm-tree 7.4.2

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

#[cfg(feature = "zstd")]
use super::KeyedBlockHandle;
use super::{BlockOffset, DataBlock, GlobalTableId, data_block::Iter as DataBlockIter};
use crate::{
    Cache, CompressionType, InternalValue, SeqNo, UserKey,
    comparator::SharedComparator,
    encryption::EncryptionProvider,
    file_accessor::FileAccessor,
    table::{
        BlockHandle,
        block::ParsedItem,
        block_index::{BlockIndexIter, BlockIndexIterImpl},
        util::ReadCharge,
    },
};
use alloc::sync::Arc;
// `Vec` is not in the `no_std` prelude; the columnar block iterator below needs
// it explicitly (under `std` this just re-imports the prelude type).
#[cfg(feature = "columnar")]
use alloc::vec::Vec;

use crate::path::PathBuf;
use self_cell::self_cell;

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

type InnerIter<'a> = DataBlockIter<'a>;

/// Whether the cold-block partial-tier read path is engaged (opt-in, OFF by
/// default). Set `LSM_PARTIAL_DECODE` to `1` / `on` / `true` / `yes` to enable.
///
/// On a large cold zstd block, a bounded range decodes only the inner blocks up
/// to its upper bound and keeps just that decompressed prefix resident; growing
/// the extent RESUMES from a cached entropy/window snapshot (decoding only the
/// new tail blocks) rather than re-decompressing from block 0. Repeatedly read
/// or mostly-decoded blocks promote to a full resident block. Off by default
/// while the path is being validated against full decode + block cache; flip the
/// env to compare.
// no-std: feature-gate behind `std`; a no-std build has no env, default OFF.
#[cfg(feature = "zstd")]
fn partial_decode_enabled() -> bool {
    use std::sync::OnceLock;
    #[cfg(test)]
    if FORCE_PARTIAL_DECODE.with(core::cell::Cell::get) {
        return true;
    }
    static ENABLED: OnceLock<bool> = OnceLock::new();
    *ENABLED.get_or_init(|| {
        std::env::var("LSM_PARTIAL_DECODE")
            .is_ok_and(|v| matches!(v.trim(), "1" | "on" | "true" | "yes"))
    })
}

#[cfg(all(test, feature = "zstd"))]
std::thread_local! {
    /// Engages the partial tier on the current thread regardless of the env
    /// switch, so a unit test neither mutates the process environment nor
    /// depends on which test first initialized the cached switch.
    static FORCE_PARTIAL_DECODE: core::cell::Cell<bool> = const { core::cell::Cell::new(false) };
}

/// Engages the partial-decode tier for reads on the calling thread.
#[cfg(all(test, feature = "zstd"))]
pub(crate) fn force_partial_decode_on_this_thread() {
    FORCE_PARTIAL_DECODE.with(|f| f.set(true));
}

/// Minimum decompressed block size for the partial tier to engage. Below this a
/// block decodes fully: the decode it would save is cheap relative to the
/// per-call setup (decoder reset, window prime, trailer synthesis), so partial
/// would cost more than it saves. Only genuinely large cold blocks qualify.
#[cfg(feature = "zstd")]
const PARTIAL_MIN_BLOCK_BYTES: usize = 256 * 1024;

/// Promote a partially-decoded cold block to a full resident block once it has
/// been served this many times: a repeatedly-read block earns its full memory.
#[cfg(feature = "zstd")]
const PARTIAL_PROMOTE_HITS: u32 = 4;

/// Promote once covering a query would decode at least this percentage of the
/// block's inner zstd blocks: past this point the partial saves little memory
/// and a single full decode beats re-growing the extent.
#[cfg(feature = "zstd")]
const PARTIAL_PROMOTE_FRACTION_PCT: u32 = 75;

/// Whether `covered` of `total` inner blocks reaches the promotion fraction.
/// `total == 0` never promotes (no layout → partial tier does not apply).
#[cfg(feature = "zstd")]
fn promote_by_fraction(covered: u32, total: u32) -> bool {
    total > 0
        && u64::from(covered) * 100 >= u64::from(total) * u64::from(PARTIAL_PROMOTE_FRACTION_PCT)
}

pub enum Bound {
    Included(UserKey),
    Excluded(UserKey),
}
type Bounds = (Option<Bound>, Option<Bound>);

self_cell!(
    pub struct OwnedDataBlockIter {
        owner: DataBlock,

        #[covariant]
        dependent: InnerIter,
    }
);

impl OwnedDataBlockIter {
    fn seek_lower_inclusive(&mut self, needle: &[u8], seqno: SeqNo) -> bool {
        self.with_dependent_mut(|_, m| m.seek(needle, seqno))
    }

    fn seek_upper_inclusive(&mut self, needle: &[u8], seqno: SeqNo) -> bool {
        self.with_dependent_mut(|_, m| m.seek_upper(needle, seqno))
    }

    fn seek_lower_exclusive(&mut self, needle: &[u8], seqno: SeqNo) -> bool {
        self.with_dependent_mut(|_, m| m.seek_exclusive(needle, seqno))
    }

    fn seek_upper_exclusive(&mut self, needle: &[u8], seqno: SeqNo) -> bool {
        self.with_dependent_mut(|_, m| m.seek_upper_exclusive(needle, seqno))
    }

    pub(crate) fn seek_lower_bound(&mut self, bound: &Bound, seqno: SeqNo) -> bool {
        match bound {
            Bound::Included(key) => self.seek_lower_inclusive(key, seqno),
            Bound::Excluded(key) => self.seek_lower_exclusive(key, seqno),
        }
    }

    pub(crate) fn seek_upper_bound(&mut self, bound: &Bound, seqno: SeqNo) -> bool {
        match bound {
            Bound::Included(key) => self.seek_upper_inclusive(key, seqno),
            Bound::Excluded(key) => self.seek_upper_exclusive(key, seqno),
        }
    }

    /// See [`crate::table::data_block::Iter::skip_past`].
    fn skip_past(&mut self, key: &[u8]) -> bool {
        self.with_dependent_mut(|_, m| m.skip_past(key))
    }

    /// See [`crate::table::data_block::Iter::trim_back_above`].
    fn trim_back_above(&mut self, key: &[u8], seqno: SeqNo) -> bool {
        self.with_dependent_mut(|_, m| m.trim_back_above(key, seqno))
    }

    /// The next row, its key rebuilt in `keys` when it has to be rebuilt.
    pub(crate) fn next_in(&mut self, keys: &mut crate::slice::KeyChunks) -> Option<InternalValue> {
        self.with_dependent_mut(|block, iter| {
            iter.next()
                .map(|item| item.materialize_in(&block.inner.data, keys))
        })
    }
}

impl Iterator for OwnedDataBlockIter {
    type Item = InternalValue;

    fn next(&mut self) -> Option<Self::Item> {
        self.with_dependent_mut(|block, iter| {
            iter.next().map(|item| item.materialize(&block.inner.data))
        })
    }
}

impl DoubleEndedIterator for OwnedDataBlockIter {
    fn next_back(&mut self) -> Option<Self::Item> {
        self.with_dependent_mut(|block, iter| {
            iter.next_back()
                .map(|item| item.materialize(&block.inner.data))
        })
    }
}

/// In-memory cursor over a columnar block's reconstructed entries. Lets the scan
/// path iterate a columnar block directly instead of re-encoding it into a
/// row-major block and re-parsing it. Entries are in block-index order
/// (`user_key` ASC, `seqno` DESC), so the bound seeks are binary searches; MVCC
/// version selection happens in the merge layer above, so `seqno` is unused here.
#[cfg(feature = "columnar")]
struct ColumnarBlockIter {
    entries: Vec<InternalValue>,
    /// Forward cursor: next index to yield. Window is `[lo, hi)`.
    lo: usize,
    /// Exclusive back cursor: `next_back` yields `hi - 1`.
    hi: usize,
    comparator: SharedComparator,
}

#[cfg(feature = "columnar")]
impl ColumnarBlockIter {
    fn new(entries: Vec<InternalValue>, comparator: SharedComparator) -> Self {
        let hi = entries.len();
        Self {
            entries,
            lo: 0,
            hi,
            comparator,
        }
    }

    /// Narrows `lo` to the first entry at/after the lower bound.
    fn seek_lower_bound(&mut self, bound: &Bound, _seqno: SeqNo) -> bool {
        let (key, skip_equal) = match bound {
            Bound::Included(k) => (k, false),
            Bound::Excluded(k) => (k, true),
        };
        let cmp = self.comparator.as_ref();
        let pos = self.entries.get(self.lo..self.hi).map_or(0, |window| {
            window.partition_point(
                |e| match cmp.compare(e.key.user_key.as_ref(), key.as_ref()) {
                    core::cmp::Ordering::Less => true,
                    core::cmp::Ordering::Equal => skip_equal,
                    core::cmp::Ordering::Greater => false,
                },
            )
        });
        self.lo += pos;
        self.lo < self.hi
    }

    /// Narrows `hi` to just past the last entry at/before the upper bound.
    fn seek_upper_bound(&mut self, bound: &Bound, _seqno: SeqNo) -> bool {
        let (key, keep_equal) = match bound {
            Bound::Included(k) => (k, true),
            Bound::Excluded(k) => (k, false),
        };
        let cmp = self.comparator.as_ref();
        let pos = self.entries.get(self.lo..self.hi).map_or(0, |window| {
            window.partition_point(
                |e| match cmp.compare(e.key.user_key.as_ref(), key.as_ref()) {
                    core::cmp::Ordering::Less => true,
                    core::cmp::Ordering::Equal => keep_equal,
                    core::cmp::Ordering::Greater => false,
                },
            )
        });
        self.hi = self.lo + pos;
        self.lo < self.hi
    }

    /// Moves `lo` past every entry whose key is `key`; whether one of a later
    /// key is left.
    fn skip_past(&mut self, key: &[u8]) -> bool {
        let cmp = self.comparator.as_ref();
        while self.lo < self.hi {
            let Some(entry) = self.entries.get(self.lo) else {
                return false;
            };
            if cmp.compare(entry.key.user_key.as_ref(), key) == core::cmp::Ordering::Greater {
                return true;
            }
            self.lo += 1;
        }
        false
    }

    /// Moves `hi` down to just past the lowest remaining entry above
    /// `(key, seqno)` in internal-key order; whether one is left.
    fn trim_back_above(&mut self, key: &[u8], seqno: SeqNo) -> bool {
        let cmp = self.comparator.as_ref();
        let Some(window) = self.entries.get(self.lo..self.hi) else {
            return false;
        };
        let first_above =
            window.partition_point(|e| match cmp.compare(e.key.user_key.as_ref(), key) {
                core::cmp::Ordering::Greater => false,
                core::cmp::Ordering::Equal => e.key.seqno >= seqno,
                core::cmp::Ordering::Less => true,
            });
        if first_above == window.len() {
            return false;
        }
        self.hi = self.lo + first_above + 1;
        true
    }
}

#[cfg(feature = "columnar")]
impl Iterator for ColumnarBlockIter {
    type Item = InternalValue;

    fn next(&mut self) -> Option<InternalValue> {
        if self.lo < self.hi {
            let item = self.entries.get(self.lo).cloned();
            self.lo += 1;
            item
        } else {
            None
        }
    }
}

#[cfg(feature = "columnar")]
impl DoubleEndedIterator for ColumnarBlockIter {
    fn next_back(&mut self) -> Option<InternalValue> {
        if self.lo < self.hi {
            self.hi -= 1;
            self.entries.get(self.hi).cloned()
        } else {
            None
        }
    }
}

/// A loaded data block ready to iterate: a row-major byte block (every row SST,
/// and the partial-decode fast path) or a columnar block's reconstructed entries
/// (iterated directly, skipping the re-encode round-trip).
enum BlockSource {
    Row(DataBlock),
    #[cfg(feature = "columnar")]
    Columnar(Vec<InternalValue>),
}

/// Iterator over a loaded block, dispatching to the byte-based row reader or the
/// in-memory columnar reader. Exposes the surface the scan loop drives.
enum BlockReader {
    Row(OwnedDataBlockIter),
    #[cfg(feature = "columnar")]
    Columnar(ColumnarBlockIter),
}

impl BlockReader {
    fn seek_lower_bound(&mut self, bound: &Bound, seqno: SeqNo) -> bool {
        match self {
            Self::Row(r) => r.seek_lower_bound(bound, seqno),
            #[cfg(feature = "columnar")]
            Self::Columnar(c) => c.seek_lower_bound(bound, seqno),
        }
    }

    fn seek_upper_bound(&mut self, bound: &Bound, seqno: SeqNo) -> bool {
        match self {
            Self::Row(r) => r.seek_upper_bound(bound, seqno),
            #[cfg(feature = "columnar")]
            Self::Columnar(c) => c.seek_upper_bound(bound, seqno),
        }
    }

    /// Moves the front past every entry whose key is `key`; whether one of a
    /// later key is left.
    fn skip_past(&mut self, key: &[u8]) -> bool {
        match self {
            Self::Row(r) => r.skip_past(key),
            #[cfg(feature = "columnar")]
            Self::Columnar(c) => c.skip_past(key),
        }
    }

    /// Moves the back down to the lowest remaining entry above `(key,
    /// seqno)`; whether one is left.
    fn trim_back_above(&mut self, key: &[u8], seqno: SeqNo) -> bool {
        match self {
            Self::Row(r) => r.trim_back_above(key, seqno),
            #[cfg(feature = "columnar")]
            Self::Columnar(c) => c.trim_back_above(key, seqno),
        }
    }
}

impl Iterator for BlockReader {
    type Item = InternalValue;

    fn next(&mut self) -> Option<InternalValue> {
        match self {
            Self::Row(r) => r.next(),
            #[cfg(feature = "columnar")]
            Self::Columnar(c) => c.next(),
        }
    }
}

impl DoubleEndedIterator for BlockReader {
    fn next_back(&mut self) -> Option<InternalValue> {
        match self {
            Self::Row(r) => r.next_back(),
            #[cfg(feature = "columnar")]
            Self::Columnar(c) => c.next_back(),
        }
    }
}

fn create_data_block_reader(
    source: BlockSource,
    comparator: SharedComparator,
) -> crate::Result<BlockReader> {
    match source {
        BlockSource::Row(block) => Ok(BlockReader::Row(OwnedDataBlockIter::try_new(block, |b| {
            b.try_iter(comparator)
        })?)),
        #[cfg(feature = "columnar")]
        BlockSource::Columnar(entries) => Ok(BlockReader::Columnar(ColumnarBlockIter::new(
            entries, comparator,
        ))),
    }
}

/// Per-SST positional delete state for a columnar iterator: the delete-bitmap
/// plus each data block's first global row position (block-index order), so a
/// reconstructed block can drop its deleted rows by position.
#[cfg_attr(
    not(feature = "columnar"),
    allow(
        dead_code,
        reason = "read only on the columnar delete-bitmap reconstruction path"
    )
)]
pub struct DeleteMask {
    pub(crate) bitmap: Arc<crate::table::delete_bitmap::DeleteBitmap>,
    pub(crate) block_start_rows: Arc<crate::HashMap<u64, u32>>,
}

#[expect(
    clippy::struct_excessive_bools,
    reason = "each bool is an independent per-SST read-path flag (kv footer, columnar, init, poisoned), not a state enum"
)]
pub struct Iter {
    table_id: GlobalTableId,
    path: Arc<PathBuf>,

    global_seqno: SeqNo,

    #[expect(clippy::struct_field_names)]
    index_iter: BlockIndexIterImpl,

    file_accessor: FileAccessor,
    cache: Arc<Cache>,
    compression: CompressionType,
    encryption: Option<Arc<dyn EncryptionProvider>>,
    /// Per-SST Page-ECC scheme from table metadata; the block reader needs
    /// it to size + recover the parity trailer. `None` = no parity.
    ecc: Option<crate::table::block::EccParams>,
    /// Tree-wide ECC heal sink. A scan that recovers a data block from parity
    /// records this SST here (on confirmed persistence) for a healing
    /// recompaction. `None` before the table is tree-owned.
    heal_hints: Option<Arc<crate::heal_hints::HealHints>>,
    /// Per-SST per-KV-footer flag from table metadata
    /// (`kv_checksum_algo.is_some()`); data blocks omit the `block_flags` byte,
    /// so `from_loaded` is told here whether to strip a footer.
    has_kv_footer: bool,
    #[cfg(zstd_any)]
    zstd_dictionary: Option<Arc<crate::compression::ZstdDictionary>>,
    comparator: SharedComparator,

    /// Per-table inner-block layout (clone). Drives the partial-decode path:
    /// when a data block has a recorded layout and the query upper bound falls
    /// within it, only the inner zstd blocks covering `[start, upper]` are
    /// decoded (trailing blocks skipped).
    #[cfg(feature = "zstd")]
    block_layout: crate::table::block_layout::BlockLayoutMap,
    /// Data-block restart interval, to rebuild a positional index when
    /// synthesizing a partial block's trailer.
    #[cfg(feature = "zstd")]
    data_block_restart_interval: u8,

    /// Whether this SST's data entries are columnar (PAX) row groups; when set,
    /// each entry is read as a row group and reconstructed into row entries.
    columnar: bool,

    /// Positional delete mask for a columnar SST with materialized deletes;
    /// `None` when the segment has no deletes. Applied during block
    /// reconstruction so deleted rows never reach the reader.
    #[cfg_attr(
        not(feature = "columnar"),
        allow(
            dead_code,
            reason = "set + read only on the columnar delete-bitmap reconstruction path"
        )
    )]
    delete_mask: Option<DeleteMask>,

    index_initialized: bool,

    lo_offset: BlockOffset,
    lo_data_block: Option<BlockReader>,

    hi_offset: BlockOffset,
    hi_data_block: Option<BlockReader>,

    /// A block handle taken from the back of the index but not yet read: the
    /// block right before the high block, which a backward jump looked at
    /// to learn where the version it seeks starts. Either end reads it
    /// before going on to the index.
    back_held: Option<crate::table::KeyedBlockHandle>,

    range: Bounds,

    /// Set on unrecoverable block-init error so subsequent `next()` /
    /// `next_back()` calls return `None` instead of skipping past the
    /// corrupt block.
    poisoned: bool,

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

    /// Whose read this iteration is.
    charge: ReadCharge,

    /// How the iteration's columnar reads fetch their pages.
    read_budget: crate::config::ReadBudget,

    /// Told about every block this iteration reads from the file, as the
    /// index walk inside it is.
    pace: Option<crate::table::util::Pacer>,

    /// The table file this iteration's blocks, and the index partitions it
    /// walks, are read through.
    fd: crate::table::util::ScanFd,
}

impl Iter {
    #[expect(
        clippy::too_many_arguments,
        reason = "encryption, comparator and metrics add extra parameters to the constructor"
    )]
    pub fn new(
        table_id: GlobalTableId,
        global_seqno: SeqNo,
        path: Arc<PathBuf>,
        index_iter: BlockIndexIterImpl,
        file_accessor: FileAccessor,
        cache: Arc<Cache>,
        compression: CompressionType,
        encryption: Option<Arc<dyn EncryptionProvider>>,
        ecc: Option<crate::table::block::EccParams>,
        heal_hints: Option<Arc<crate::heal_hints::HealHints>>,
        has_kv_footer: bool,
        columnar: bool,
        delete_mask: Option<DeleteMask>,
        #[cfg(zstd_any)] zstd_dictionary: Option<Arc<crate::compression::ZstdDictionary>>,
        comparator: SharedComparator,
        #[cfg(feature = "zstd")] block_layout: crate::table::block_layout::BlockLayoutMap,
        #[cfg(feature = "zstd")] data_block_restart_interval: u8,
        #[cfg(feature = "metrics")] metrics: Arc<Metrics>,
    ) -> Self {
        Self {
            table_id,
            path,

            global_seqno,

            index_iter,
            file_accessor,
            cache,
            compression,
            encryption,
            ecc,
            heal_hints,
            has_kv_footer,
            columnar,
            delete_mask,
            #[cfg(zstd_any)]
            zstd_dictionary,
            comparator,

            #[cfg(feature = "zstd")]
            block_layout,
            #[cfg(feature = "zstd")]
            data_block_restart_interval,

            index_initialized: false,

            lo_offset: BlockOffset(0),
            lo_data_block: None,

            hi_offset: BlockOffset(u64::MAX),
            hi_data_block: None,

            back_held: None,

            range: (None, None),
            poisoned: false,

            #[cfg(feature = "metrics")]
            metrics,
            charge: ReadCharge::Foreground,
            read_budget: crate::config::ReadBudget::default(),
            pace: None,
            fd: crate::table::util::ScanFd::default(),
        }
    }

    /// Tells `pace` about every block this iteration, and the index walk
    /// inside it, reads from the file.
    #[must_use]
    #[cfg_attr(
        not(feature = "std"),
        expect(dead_code, reason = "its compaction consumer is std-gated")
    )]
    pub(crate) fn with_pace(mut self, pace: crate::table::util::Pacer) -> Self {
        self.index_iter = self.index_iter.with_pace(alloc::sync::Arc::clone(&pace));
        self.pace = Some(pace);
        self
    }

    /// Reads this iteration's columnar row groups under `budget`.
    #[must_use]
    pub(crate) fn with_read_budget(mut self, budget: crate::config::ReadBudget) -> Self {
        self.read_budget = budget;
        self
    }

    /// Marks this iterator, and the index walk inside it, as maintenance such
    /// as a sub-compaction's input: not a read a caller made, so it stays out
    /// of the read counters while its Page-ECC recoveries are still counted.
    #[must_use]
    #[cfg_attr(
        not(feature = "std"),
        expect(dead_code, reason = "its compaction consumer is std-gated")
    )]
    pub(crate) fn for_maintenance(self) -> Self {
        self.with_charge(ReadCharge::Maintenance)
    }

    /// Marks this iterator, and the index walk inside it, as a read that
    /// leaves no trace, such as verification during a scrub: uncounted, and
    /// it neither fills the block cache nor promotes what is cached.
    #[must_use]
    #[cfg(feature = "std")]
    pub(crate) fn untraced(self) -> Self {
        self.with_charge(ReadCharge::Untraced)
    }

    fn with_charge(mut self, charge: ReadCharge) -> Self {
        self.charge = charge;
        self.index_iter = self.index_iter.with_charge(charge);
        self
    }

    /// Loads, reconstructs (if columnar), and masks a data block by handle,
    /// dispatching on the SST layout: a columnar SST's entry is a row group,
    /// read and decoded into its row entries; a row SST's is a data block,
    /// loaded directly. Returns `Ok(None)` when a columnar group is wholly
    /// deleted by the positional mask, so the caller skips it.
    fn load_and_resolve(&self, handle: &BlockHandle) -> crate::Result<Option<BlockSource>> {
        let read = self.fd.lookup();
        let block = self.load_and_resolve_through(handle, &read)?;
        self.fd.keep(read);
        Ok(block)
    }

    /// [`Self::load_and_resolve`], the block read from the file through `fd`.
    fn load_and_resolve_through(
        &self,
        handle: &BlockHandle,
        fd: &crate::table::util::LookupFd,
    ) -> crate::Result<Option<BlockSource>> {
        if self.columnar {
            #[cfg(feature = "columnar")]
            {
                let group = crate::table::row_group::GroupRead {
                    table_id: self.table_id,
                    path: &self.path,
                    file_accessor: &self.file_accessor,
                    cache: &self.cache,
                    group: handle,
                    compression: self.compression,
                    encryption: self.encryption.as_deref(),
                    ecc: self.ecc,
                    #[cfg(zstd_any)]
                    zstd_dict: self.zstd_dictionary.as_deref(),
                    heal_hints: self.heal_hints.as_ref().map(AsRef::as_ref),
                    #[cfg(feature = "metrics")]
                    metrics: &self.metrics,
                    charge: self.charge,
                    budget: self.read_budget,
                    pace: self.pace.as_deref(),
                    fd,
                }
                .load(&crate::table::row_group::PageWant::ALL)?;
                // What decoding the pages copied out of them, plus the values
                // rebuilt from sub-columns below. Keys, and values of a single
                // bytes column, are views into the decoded columns and cost
                // nothing. Charged before the result is judged: a group
                // refused after a gather still did it.
                let mut gathered = 0usize;
                let batch = group.to_row_pages(&mut gathered).map(|pages| pages.batches);
                // Mask only when the segment has deletes AND this block's start row
                // is known. The start-row map is built at open from the zone map
                // (which covers every block), so an unmapped block is unreachable;
                // it falls through to reconstructing the whole block (no mask),
                // sharing the no-deletes path rather than defaulting the start to 0
                // and masking against the wrong physical positions. This matches the
                // direct-load path (`Table::load_columnar_data_block`).
                //
                // The reconstructed entries are iterated directly (no re-encode to
                // a row-major block + re-parse), so no restart interval is needed.
                let masked = self.delete_mask.as_ref().and_then(|mask| {
                    mask.block_start_rows
                        .get(&handle.offset().0)
                        .map(|&start| (mask, start))
                });
                let entries = batch.and_then(|batch| match masked {
                    Some((mask, start)) => DataBlock::column_batch_entries_masked(
                        batch,
                        &mask.bitmap,
                        start,
                        &mut gathered,
                    ),
                    None => DataBlock::column_batch_entries(batch, &mut gathered).map(Some),
                });
                #[cfg(feature = "metrics")]
                if self.charge.is_counted() {
                    self.metrics.record_gather(gathered);
                }
                #[cfg(not(feature = "metrics"))]
                let _ = gathered;
                return Ok(entries?.map(BlockSource::Columnar));
            }
            #[cfg(not(feature = "columnar"))]
            {
                return Err(crate::Error::FeatureUnsupported("columnar"));
            }
        }
        let raw = crate::table::util::load_block_paced(
            self.table_id,
            &self.path,
            &self.file_accessor,
            &self.cache,
            handle,
            crate::table::block::BlockType::Data,
            self.compression,
            self.encryption.as_deref(),
            self.ecc,
            #[cfg(zstd_any)]
            self.zstd_dictionary.as_deref(),
            self.heal_hints.as_ref().map(AsRef::as_ref),
            #[cfg(feature = "metrics")]
            &self.metrics,
            self.charge,
            self.pace.as_deref(),
            fd,
        )?;
        DataBlock::from_loaded(raw, self.has_kv_footer).map(|db| Some(BlockSource::Row(db)))
    }

    pub fn set_lower_bound(&mut self, bound: Bound) {
        self.range.0 = Some(bound);
    }

    pub fn set_upper_bound(&mut self, bound: Bound) {
        self.range.1 = Some(bound);
    }

    /// Reset this iterator to an un-positioned state so it can be re-seeked to a
    /// fresh range WITHOUT rebuilding the index iterator or re-cloning the
    /// table's `Arc` handles.
    ///
    /// Clears the bound pair, the materialized low/high data blocks, the offset
    /// cursors, a block handle a backward jump held, and the poison flag, and rewinds the retained `index_iter` to
    /// the table start. The rewind is load-bearing: lazy init only calls
    /// `seek_lower` when the new range HAS a lower bound, so a re-seek to a
    /// range with an UNBOUNDED lower would otherwise inherit the previous pass's
    /// front cursor and skip earlier blocks. After this call the next
    /// [`Iterator::next`] / [`DoubleEndedIterator::next_back`] re-positions from
    /// scratch against the bounds set via [`set_lower_bound`](Self::set_lower_bound)
    /// / [`set_upper_bound`](Self::set_upper_bound).
    pub fn reset_for_reseek(&mut self) {
        self.range = (None, None);
        self.index_initialized = false;
        self.lo_offset = BlockOffset(0);
        self.lo_data_block = None;
        self.hi_offset = BlockOffset(u64::MAX);
        self.hi_data_block = None;
        // A handle a backward jump held belongs to the previous pass: either
        // end would read it before the index and replay a block out of range.
        self.back_held = None;
        // A fresh `table.range()` would produce a non-poisoned iterator; match
        // that so a re-seek past a previously-corrupt block can make progress
        // again (the block is re-read, and re-poisons only if still corrupt).
        self.poisoned = false;
        // Rewind the front (and back) index cursor to the first block. An empty
        // needle is `<=` every key, so `seek_lower` positions at the start and
        // resets both caches; an unbounded-lower re-seek then starts fresh, and
        // a bounded-lower one re-seeks over this in init.
        self.index_iter.seek_lower(&[], u64::MAX);
    }

    /// Adaptive partial-tier read for `handle`: keep a cold block only partially
    /// decoded (the touched fraction) instead of materializing the whole block,
    /// and promote it to a full resident block once access justifies the memory.
    /// Returns `None` (caller does a full load + caches the whole block) when the
    /// partial tier does not apply or the block should be promoted:
    ///
    /// - not plain zstd, encrypted, no recorded multi-inner-block layout, or no
    ///   query upper bound (an unbounded scan reads the whole block anyway);
    /// - the block sits entirely below `upper` (fully in range → full decode);
    /// - the block is already resident as a full block (let the cache serve it);
    /// - PROMOTION: the cached partial has been served [`PARTIAL_PROMOTE_HITS`]
    ///   times, or covering the query would decode at least
    ///   [`PARTIAL_PROMOTE_FRACTION_PCT`]% of the inner blocks — at that point a
    ///   single full decode + block cache beats re-growing the partial extent
    ///   (which, lacking a resumable decoder, re-decodes from inner block 0).
    ///
    /// Otherwise it returns a synthesized covering block and caches it with the
    /// extent + access stats for the next read.
    #[cfg(feature = "zstd")]
    fn try_partial_block(&self, handle: &KeyedBlockHandle) -> crate::Result<Option<DataBlock>> {
        if !partial_decode_enabled()
            || !matches!(self.compression, CompressionType::Zstd(_))
            || self.encryption.is_some()
            // Columnar blocks have a different on-disk shape and are
            // reconstructed whole, so the inner-block partial decode never
            // applies to them.
            || self.columnar
            // The tier lives in the block cache (inserts, hit bumps,
            // promotion), so a read that must leave no trace there skips it.
            || !self.charge.touches_cache()
        {
            return Ok(None);
        }
        let Some(Bound::Included(upper) | Bound::Excluded(upper)) = &self.range.1 else {
            return Ok(None);
        };
        if self.comparator.compare(upper, handle.end_key()) == std::cmp::Ordering::Greater {
            return Ok(None);
        }
        let Some(ends) = self.block_layout.ends_for(*handle.offset()) else {
            return Ok(None);
        };
        let offset = BlockOffset(*handle.offset());
        // Cold-block gate: only engage on genuinely large blocks (the cold tier
        // runs multi-MB blocks) where decoding a fraction saves far more than the
        // per-call setup. The total decompressed size is the last cumulative end.
        let total_bytes = ends.last().copied().unwrap_or(0) as usize;
        if total_bytes < PARTIAL_MIN_BLOCK_BYTES {
            return Ok(None);
        }
        #[expect(
            clippy::cast_possible_truncation,
            reason = "inner-block count is bounded well within u32"
        )]
        let total_blocks = ends.len() as u32;
        let ends = ends.to_vec();

        // Looks at the cached prefix, and again after waiting on another
        // read's growth of it, which may have covered this query already.
        let mut waited = false;
        let (carried_hits, carried_resume, _growth) = loop {
            // Already promoted to a full resident block → let the normal
            // cached load path serve it (no point re-synthesizing a partial
            // alongside it).
            if self.cache.has_block(self.table_id, offset) {
                return Ok(None);
            }
            // A full load of the block is in flight (a point read, an
            // unbounded scan, a promotion): the full load path waits for it
            // and takes the whole block, which covers any query, rather than
            // reading the frame again for a prefix.
            if self.cache.block_load_in_flight(self.table_id, offset) {
                return Ok(None);
            }
            // Existing partial entry: serve from it (bumping hits) when it
            // already covers the query, else decide whether to grow it or
            // promote. The fall-through carries the hit count + the resume
            // payload to continue from.
            let (carried_hits, carried_resume) =
                match self.cache.peek_partial_block(self.table_id, offset) {
                    Some(entry) => {
                        let covered = self.comparator.compare(&entry.covered_upper, upper)
                            != std::cmp::Ordering::Less;
                        if covered {
                            let hits = entry.hits + 1;
                            if hits >= PARTIAL_PROMOTE_HITS {
                                // Hot enough: promote to a full resident block.
                                self.cache.evict_partial_block(self.table_id, offset);
                                return Ok(None);
                            }
                            // Synthesize the served block from the cached decompressed
                            // prefix (only the touched fraction is held in memory).
                            let block = crate::table::lazy_block::synthesize_data_block(
                                &entry.resume.window_prime,
                                self.data_block_restart_interval,
                            )?;
                            // The block is a gather of the cached prefix's entries.
                            #[cfg(feature = "metrics")]
                            if self.charge.is_counted() {
                                self.metrics.record_gather(block.inner.data.len());
                            }
                            self.cache.insert_partial_block(
                                self.table_id,
                                offset,
                                crate::cache::PartialBlockEntry { hits, ..entry },
                            );
                            return Ok(Some(block));
                        }
                        // Coverage miss → grow the extent by resuming from the snapshot.
                        // If most of the block is already decoded, a full decode + block
                        // cache beats holding a near-complete partial, so promote.
                        if promote_by_fraction(entry.resume.decoded_blocks, total_blocks) {
                            self.cache.evict_partial_block(self.table_id, offset);
                            return Ok(None);
                        }
                        (entry.hits, Some(entry.resume))
                    }
                    // The growth this read waited on left no prefix: it
                    // promoted the block and is loading it whole, or it
                    // failed. Either way the full load path serves this read,
                    // taking the block from that load rather than reading the
                    // frame again for a prefix.
                    None if waited => return Ok(None),
                    None => (0, None),
                };
            match self.cache.claim_partial_growth(self.table_id, offset) {
                crate::cache::GrowthClaim::Lead(lead) => {
                    // A growth that ended between the look above and this
                    // claim left a longer prefix: look at it instead of
                    // decoding from the older one.
                    let decoded = |resume: Option<&crate::table::lazy_block::PartialResume>| {
                        resume.map_or(0, |resume| resume.decoded_blocks)
                    };
                    let grown = self
                        .cache
                        .peek_partial_block(self.table_id, offset)
                        .is_some_and(|entry| {
                            decoded(Some(&entry.resume)) > decoded(carried_resume.as_ref())
                        });
                    if grown
                        || self.cache.has_block(self.table_id, offset)
                        || self.cache.block_load_in_flight(self.table_id, offset)
                    {
                        drop(lead);
                        continue;
                    }
                    break (carried_hits, carried_resume, Some(lead));
                }
                crate::cache::GrowthClaim::Alone => break (carried_hits, carried_resume, None),
                crate::cache::GrowthClaim::Waited => waited = true,
            }
        };

        let timer = crate::cache::Cache::time_load();
        let read = self.fd.lookup();
        let fd = read.resolve(
            &self.file_accessor,
            &self.table_id,
            &self.path,
            self.charge,
            #[cfg(feature = "metrics")]
            &self.metrics,
        )?;
        let transform = crate::table::block::BlockTransform::from_parts(
            self.compression,
            None,
            #[cfg(zstd_any)]
            None,
        )?;
        let transform = match self.ecc {
            Some(ecc) => transform.with_ecc(ecc),
            None => transform,
        };
        let block_handle = BlockHandle::new(handle.offset(), handle.size());
        let (_header, frame, recovery) = crate::table::block::Block::read_data_frame(
            fd,
            block_handle,
            crate::table::block::BlockIdentity {
                table_id: self.table_id.table_id(),
                block_type: crate::table::block::BlockType::Data,
                dict_id: self.compression.dict_id(),
                window_log: 0,
            },
            &transform,
            // The whole frame is asked of the filesystem on every partial
            // read, including a resume-grow; charged as issued, like
            // `load_block`.
            || {
                #[cfg(feature = "metrics")]
                if self.charge.is_counted() {
                    crate::table::util::record_block_read(
                        &self.metrics,
                        crate::table::block::BlockType::Data,
                        handle.size().into(),
                    );
                }
            },
        )?;
        self.fd.keep(read);
        // The partial path bypasses `load_block`, so schedule auto-heal here too:
        // an ECC-corrected frame read flags the SST for a healing rewrite (the
        // partial path is non-encrypted by the guard above). This is a primary
        // read, so count the recovery by mechanism as well.
        if let Some(kind) = recovery {
            #[cfg(feature = "metrics")]
            self.metrics.record_ecc_recovery(kind);
            #[cfg(not(feature = "metrics"))]
            let _ = kind;
            crate::table::util::maybe_record_persistent_heal(
                self.table_id,
                &self.path,
                &self.file_accessor,
                &block_handle,
                crate::table::block::BlockType::Data,
                self.compression,
                None,
                self.ecc,
                #[cfg(zstd_any)]
                self.zstd_dictionary.as_deref(),
                self.heal_hints.as_ref().map(AsRef::as_ref),
                #[cfg(feature = "metrics")]
                &self.metrics,
                self.charge,
            );
        }
        // Cold first touch (carried_resume None) or resume-grow from the cached
        // snapshot — either way only the new tail blocks are decompressed.
        let mut work = crate::table::lazy_block::PartialWork::default();
        let built = crate::table::lazy_block::partial_data_block(
            frame,
            ends,
            self.data_block_restart_interval,
            &self.comparator,
            upper,
            carried_resume,
            &mut work,
        );
        // Charged before the result is judged: a read that fails at a later
        // inner block still decoded and copied what came before it. Every
        // prefix copy and the synthesized block are gathers, so a range that
        // grows across reads re-copies a longer prefix each time; only the
        // tail this read decoded counts, a resumed prefix was charged by the
        // read that decoded it.
        #[cfg(feature = "metrics")]
        if self.charge.is_counted() {
            self.metrics.record_gather(work.copied);
            self.metrics
                .block_bytes_decoded
                .fetch_add(work.decoded as u64, core::sync::atomic::Ordering::Relaxed);
        }
        #[cfg(not(feature = "metrics"))]
        let _ = work;
        let crate::table::lazy_block::PartialBlock {
            block,
            covered_upper,
            payload,
        } = built?;
        // Timed into the shard its growths coordinate in, which no other
        // load of the block feeds.
        self.cache
            .record_partial_growth(self.table_id, offset, timer);
        // Covering this query decoded most of the block → promote: drop the
        // partial and let the caller cache the whole block.
        if promote_by_fraction(payload.decoded_blocks, total_blocks) {
            self.cache.evict_partial_block(self.table_id, offset);
            return Ok(None);
        }
        // Cache the resume payload tagged with the extent it covers + carried
        // access stats, so the next read serves or resume-grows it.
        if let Some(covered) = covered_upper {
            self.cache.insert_partial_block(
                self.table_id,
                offset,
                crate::cache::PartialBlockEntry {
                    resume: payload,
                    covered_upper: covered,
                    total_blocks,
                    hits: carried_hits,
                },
            );
        }
        Ok(Some(block))
    }
}

impl Iter {
    /// Marks the iterator as permanently failed so subsequent `next()` /
    /// `next_back()` calls return `None` instead of skipping past the error.
    ///
    /// Returns `Some(Err(...))` so callers can `return self.poison(e)` directly
    /// inside `Iterator::next`.
    #[expect(
        clippy::unnecessary_wraps,
        reason = "matches Iterator::next return type"
    )]
    fn poison<E: Into<crate::Error>>(&mut self, err: E) -> Option<crate::Result<InternalValue>> {
        self.poisoned = true;
        Some(Err(err.into()))
    }

    /// Positions the index on the blocks the range reaches, on the first pull
    /// rather than in `new`, so bounds set after construction cost no seek or
    /// I/O. Bound exclusivity is enforced in the data blocks; the index only
    /// narrows the span. `false` when no block overlaps the range.
    fn init_index(&mut self) -> bool {
        let mut ok = if let Some(bound) = &self.range.0 {
            // The first block whose end key is >= the lower bound.
            let key = match bound {
                Bound::Included(k) | Bound::Excluded(k) => k,
            };
            self.index_iter.seek_lower(key, u64::MAX)
        } else {
            true
        };

        if ok && let Some(bound) = &self.range.1 {
            let key = match bound {
                Bound::Included(k) | Bound::Excluded(k) => k,
            };
            ok = self.index_iter.seek_upper(key, u64::MAX);
        }

        self.index_initialized = true;

        if !ok {
            self.lo_data_block = None;
            self.hi_data_block = None;
        }
        ok
    }

    /// The next block handle the front reads: from the index, then the one a
    /// backward jump holds, which lies after everything the index has left.
    fn next_front_handle(&mut self) -> Option<crate::Result<crate::table::KeyedBlockHandle>> {
        let read = self.fd.lookup();
        let handle = self.index_iter.next_with(&read);
        self.fd.keep(read);
        handle.or_else(|| self.back_held.take().map(Ok))
    }

    /// The next block handle the back reads: the one a backward jump holds,
    /// then the index.
    fn next_back_handle(&mut self) -> Option<crate::Result<crate::table::KeyedBlockHandle>> {
        if let Some(handle) = self.back_held.take() {
            return Some(Ok(handle));
        }
        let read = self.fd.lookup();
        let handle = self.index_iter.next_back_with(&read);
        self.fd.keep(read);
        handle
    }

    /// The reader of the data block `handle` names, clamped to the range;
    /// `None` when a delete mask leaves nothing of it.
    fn open_front_block(
        &self,
        handle: &crate::table::KeyedBlockHandle,
    ) -> crate::Result<Option<BlockReader>> {
        // Partial-decode fast path: for a large multi-inner-block zstd block
        // whose recorded layout lets us skip the inner blocks past the query
        // upper bound, decode only the covering prefix. Falls back to a full
        // load (which checks the block cache) otherwise. Positional masking
        // needs the whole block, so a masked table never takes it.
        #[cfg(feature = "zstd")]
        let partial = if self.delete_mask.is_some() {
            None
        } else {
            self.try_partial_block(handle)?
        };
        #[cfg(not(feature = "zstd"))]
        let partial: Option<DataBlock> = None;

        let block = if let Some(db) = partial {
            BlockSource::Row(db)
        } else {
            match self.load_and_resolve(handle.as_ref())? {
                Some(b) => b,
                None => return Ok(None),
            }
        };

        let mut reader = create_data_block_reader(block, self.comparator.clone())?;

        // The low side first, so no entry below the lower bound is returned,
        // then the high side, which keeps the block in [low, high] with exact
        // inclusivity.
        if let Some(bound) = &self.range.0 {
            reader.seek_lower_bound(bound, SeqNo::MAX);
        }
        if let Some(bound) = &self.range.1 {
            reader.seek_upper_bound(bound, SeqNo::MAX);
        }
        Ok(Some(reader))
    }

    /// The reader of the data block `handle` names for the back, clamped to
    /// the range; `None` when a delete mask leaves nothing of it.
    fn open_back_block(
        &self,
        handle: &crate::table::KeyedBlockHandle,
    ) -> crate::Result<Option<BlockReader>> {
        // Partial-decode fast path: for a large multi-inner-block zstd block
        // whose recorded layout lets us skip the inner blocks past the query
        // upper bound, decode only the covering prefix. Falls back to a full
        // load (which checks the block cache) otherwise. Positional masking
        // needs the whole block, so a masked table never takes it.
        #[cfg(feature = "zstd")]
        let partial = if self.delete_mask.is_some() {
            None
        } else {
            self.try_partial_block(handle)?
        };
        #[cfg(not(feature = "zstd"))]
        let partial: Option<DataBlock> = None;

        let block = if let Some(db) = partial {
            BlockSource::Row(db)
        } else {
            match self.load_and_resolve(handle.as_ref())? {
                Some(b) => b,
                None => return Ok(None),
            }
        };

        let mut reader = create_data_block_reader(block, self.comparator.clone())?;

        // The high side first, so `next_back` never yields an entry above the
        // upper bound, then the low side, so it never steps below the lower.
        if let Some(bound) = &self.range.1 {
            reader.seek_upper_bound(bound, SeqNo::MAX);
        }
        if let Some(bound) = &self.range.0 {
            reader.seek_lower_bound(bound, SeqNo::MAX);
        }
        Ok(Some(reader))
    }

    /// Moves the back up to the newest version of `key` visible below
    /// `seqno` (see [`crate::skip_key::SkipKey::skip_back_to_newest`]).
    ///
    /// The index's handles name each data block's last entry. A block whose
    /// predecessor ends above the newest visible version's position holds only
    /// older versions of `key` and is passed on the index alone, never loaded;
    /// the handle that showed it is held for the next read from either end.
    /// The block the version is in is then trimmed in place.
    ///
    /// # Errors
    ///
    /// An error reading the index or a block, which fails the iterator as
    /// `next_back()` would.
    pub(crate) fn skip_back_to_newest(&mut self, key: &[u8], seqno: SeqNo) -> crate::Result<bool> {
        // A masked table's handles may name a deleted last row, which says
        // nothing of the rows that survive.
        if self.poisoned || self.delete_mask.is_some() {
            return Ok(false);
        }
        // The table stores seqnos less its global seqno, so a stored seqno is
        // visible below `seqno - global_seqno`; below a snapshot at or under
        // the global seqno nothing is.
        let Some(seqno) = seqno.checked_sub(self.global_seqno) else {
            return Ok(false);
        };
        if !self.index_initialized && !self.init_index() {
            return Ok(false);
        }
        // An entry `above` lies past `(key, seqno)` in internal-key order: a
        // later key, or a version of `key` the snapshot sees. The version
        // sought is the lowest such entry. A cold path: the comparator is
        // cloned so the closure does not hold `self`.
        let comparator = self.comparator.clone();
        let above = |end_key: &[u8], end_seqno: SeqNo| match comparator.compare(end_key, key) {
            core::cmp::Ordering::Greater => true,
            core::cmp::Ordering::Equal => end_seqno < seqno,
            core::cmp::Ordering::Less => false,
        };

        // The block the back reads next: the high block while it is open,
        // or a handle not yet read.
        let mut pending: Option<crate::table::KeyedBlockHandle> = None;
        loop {
            let Some(handle) = self.next_back_handle() else {
                break;
            };
            let handle = match handle {
                Ok(h) => h,
                Err(e) => {
                    self.poisoned = true;
                    return Err(e);
                }
            };
            if above(handle.end_key(), handle.seqno()) {
                // Every entry after that block lies above its last one, itself
                // above `(key, seqno)`: the back's block holds only versions
                // older than the one sought.
                self.hi_data_block = None;
                pending = Some(handle);
            } else {
                // The version sought lies after that block.
                self.back_held = Some(handle);
                break;
            }
        }

        if let Some(handle) = pending {
            match self.open_back_block(&handle) {
                Ok(reader) => {
                    self.hi_offset = handle.offset();
                    self.hi_data_block = reader;
                }
                Err(e) => {
                    self.poisoned = true;
                    return Err(e);
                }
            }
        }

        // The back reads the high block, or the low block once the index is
        // used up and no high block is open.
        let block = match &mut self.hi_data_block {
            Some(block) => block,
            None if self.back_held.is_none() => match &mut self.lo_data_block {
                Some(block) => block,
                None => return Ok(false),
            },
            None => return Ok(false),
        };
        Ok(block.trim_back_above(key, seqno))
    }

    /// Moves the front past every entry whose user key is `key` (see
    /// [`crate::skip_key::SkipKey`]); whether an entry of a later key is left.
    /// A data block whose last key is `key` holds nothing else and is passed
    /// on the index alone, never loaded.
    ///
    /// # Errors
    ///
    /// An error reading the index or a block, which fails the iterator as
    /// `next()` would.
    pub(crate) fn skip_past(&mut self, key: &[u8]) -> crate::Result<bool> {
        if self.poisoned {
            return Ok(false);
        }
        if let Some(block) = &mut self.lo_data_block
            && block.skip_past(key)
        {
            return Ok(true);
        }
        if !self.index_initialized && !self.init_index() {
            return Ok(false);
        }

        loop {
            let Some(handle) = self.next_front_handle() else {
                // The blocks left were read from the back: the front goes on
                // in the last of them.
                return Ok(self
                    .hi_data_block
                    .as_mut()
                    .is_some_and(|block| block.skip_past(key)));
            };
            let handle = match handle {
                Ok(h) => h,
                Err(e) => {
                    self.poisoned = true;
                    return Err(e);
                }
            };
            if self.comparator.compare(handle.end_key(), key) != core::cmp::Ordering::Greater {
                continue;
            }
            let mut reader = match self.open_front_block(&handle) {
                Ok(Some(reader)) => reader,
                Ok(None) => continue,
                Err(e) => {
                    self.poisoned = true;
                    return Err(e);
                }
            };
            let more = reader.skip_past(key);
            self.lo_offset = handle.offset();
            self.lo_data_block = Some(reader);
            if more {
                return Ok(true);
            }
        }
    }
}

impl crate::skip_key::SkipKey for Iter {
    fn skip_front_past(&mut self, key: &[u8]) -> crate::Result<bool> {
        self.skip_past(key).map(|_| true)
    }

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

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

    fn next(&mut self) -> Option<Self::Item> {
        if self.poisoned {
            return None;
        }

        // Always try to keep iterating inside the already-materialized low data block first; this
        // lets callers consume multiple entries without touching the index or cache again.
        if let Some(block) = &mut self.lo_data_block
            && let Some(item) = block
                .next()
                .map(|mut v| {
                    v.key.seqno += self.global_seqno;
                    v
                })
                .map(Ok)
        {
            return Some(item);
        }

        if !self.index_initialized && !self.init_index() {
            return None;
        }

        loop {
            let Some(handle) = self.next_front_handle() else {
                // No more block handles coming from the index.  Flush any pending items buffered on
                // the high side (used by reverse iteration) before signalling completion.
                if let Some(block) = &mut self.hi_data_block
                    && let Some(item) = block
                        .next()
                        .map(|mut v| {
                            v.key.seqno += self.global_seqno;
                            v
                        })
                        .map(Ok)
                {
                    return Some(item);
                }

                // Nothing left to serve; drop both buffers so the iterator can be reused safely.
                self.lo_data_block = None;
                self.hi_data_block = None;
                return None;
            };
            let handle = match handle {
                Ok(h) => h,
                Err(e) => return self.poison(e),
            };

            let mut reader = match self.open_front_block(&handle) {
                Ok(Some(reader)) => reader,
                // The block was wholly deleted by the mask; skip to the next.
                Ok(None) => continue,
                Err(e) => return self.poison(e),
            };

            let item = reader.next();

            self.lo_offset = handle.offset();
            self.lo_data_block = Some(reader);

            if let Some(mut item) = item {
                item.key.seqno += self.global_seqno;

                // Serving the first item immediately avoids stashing it in a temporary buffer and
                // keeps block iteration semantics identical to the simple case at the top.
                return Some(Ok(item));
            }
        }
    }
}

impl DoubleEndedIterator for Iter {
    fn next_back(&mut self) -> Option<Self::Item> {
        if self.poisoned {
            return None;
        }

        // Mirror the forward iterator: prefer consuming buffered items from the high data block to
        // avoid touching the index once a block has been materialized.
        if let Some(block) = &mut self.hi_data_block
            && let Some(item) = block
                .next_back()
                .map(|mut v| {
                    v.key.seqno += self.global_seqno;
                    v
                })
                .map(Ok)
        {
            return Some(item);
        }

        if !self.index_initialized {
            // Mirror forward iteration: initialize lazily so bounds can be applied up-front. The
            // index only restricts which blocks we consider; tight bound enforcement happens in
            // the data block readers below.
            let mut ok = if let Some(bound) = &self.range.0 {
                let key = match bound {
                    Bound::Included(k) | Bound::Excluded(k) => k,
                };
                self.index_iter.seek_lower(key, u64::MAX)
            } else {
                true
            };

            if ok && let Some(bound) = &self.range.1 {
                let key = match bound {
                    Bound::Included(k) | Bound::Excluded(k) => k,
                };
                ok = self.index_iter.seek_upper(key, u64::MAX);
            }

            self.index_initialized = true;

            if !ok {
                // No index span overlaps the requested window; clear both buffers and finish early.
                self.lo_data_block = None;
                self.hi_data_block = None;
                return None;
            }
        }

        loop {
            let Some(handle) = self.next_back_handle() else {
                // Once we exhaust the index in reverse order, flush any items that were buffered on
                // the low side (set when iterating forward first) before signalling completion.
                if let Some(block) = &mut self.lo_data_block
                    && let Some(item) = block
                        .next_back()
                        .map(|mut v| {
                            v.key.seqno += self.global_seqno;
                            v
                        })
                        .map(Ok)
                {
                    return Some(item);
                }

                // Nothing left to produce; reset both buffers to keep the iterator reusable.
                self.lo_data_block = None;
                self.hi_data_block = None;
                return None;
            };
            let handle = match handle {
                Ok(h) => h,
                Err(e) => return self.poison(e),
            };

            let mut reader = match self.open_back_block(&handle) {
                Ok(Some(r)) => r,
                // The block was wholly deleted by the mask; skip to the next.
                Ok(None) => continue,
                Err(e) => return self.poison(e),
            };

            let item = reader.next_back();

            self.hi_offset = handle.offset();
            self.hi_data_block = Some(reader);

            if let Some(mut item) = item {
                item.key.seqno += self.global_seqno;

                // Emit the first materialized entry immediately to match the forward path and avoid
                // storing it in a temporary buffer.
                return Some(Ok(item));
            }
        }
    }
}

#[cfg(all(test, feature = "zstd"))]
mod promote_tests;