coordinode-lsm-tree 7.4.2

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

pub mod field_row;
mod gc;
pub mod handle;
pub mod ingest;
#[cfg(feature = "std")]
pub(crate) mod links;
pub(crate) mod overwritten;
pub(crate) mod released;

#[doc(hidden)]
pub use gc::{FragmentationEntry, FragmentationMap};

use crate::path::PathBuf;
use crate::tree::inner::{FlushGuard, VersionsWriteGuard};
use crate::{
    Config, Memtable, ScanSinceEvent, SeqNo, TableId, TreeId, UserKey, UserValue,
    abstract_tree::{AbstractTree, RangeItem},
    coding::Decode,
    iter_guard::{IterGuard, IterGuardImpl},
    table::Table,
    tree::inner::MemtableId,
    value::InternalValue,
    version::Version,
    vlog::{Accessor, BlobFile, BlobFileWriter},
};
use alloc::sync::Arc;
#[cfg(not(feature = "std"))]
use alloc::{boxed::Box, string::ToString, vec::Vec};
use core::ops::RangeBounds;
use handle::BlobIndirection;

/// Iterator value guard
pub struct Guard {
    tree: crate::BlobTree,
    version: Version,
    /// Whether resolving the value caches it, as the tree said when the scan
    /// began.
    fill: crate::config::BlobCacheFill,
    /// The values of this item the scan read ahead for it, when it does not
    /// cache them: resolved from here, not read again.
    prefetched: Vec<(crate::vlog::ValueHandle, UserValue)>,
    kv: crate::Result<InternalValue>,
    /// Present when this guard came from a scan that can read ahead; resolving
    /// a value through it is what arms that read-ahead.
    prefetch: Option<Arc<ScanPrefetch>>,
    /// Counts `prefetched` against its scan's read-ahead bound until this
    /// guard is resolved or dropped: held for its drop alone.
    _held: Option<HeldReadAhead>,
}

/// Bytes of read-ahead values a guard holds, released with the guard: what
/// lets a scan see that the guards it handed out still hold a read's worth.
struct HeldReadAhead {
    state: Arc<ScanPrefetch>,
    bytes: usize,
}

impl Drop for HeldReadAhead {
    fn drop(&mut self) {
        self.state
            .held
            .fetch_sub(self.bytes, core::sync::atomic::Ordering::Relaxed);
    }
}

impl IterGuard for Guard {
    fn into_inner_if(
        self,
        pred: impl Fn(&UserKey) -> bool,
    ) -> crate::Result<(UserKey, Option<UserValue>)> {
        let kv = self.kv?;

        // Deliberately does NOT arm the scan's read-ahead, even on a match.
        // This method exists so a caller can decide per key whether the value
        // is worth reading, and a sparse predicate is the case it exists for.
        // Arming on the first match would read ahead for every later item
        // before its own predicate could reject it, which is exactly the I/O
        // the caller asked to avoid: one match in a thousand would turn a
        // selective scan into a full read of the value log. A caller that
        // wants every value uses `into_inner`, and that is what arms it.
        if pred(&kv.key.user_key) {
            resolve_value_handle(
                self.tree.id(),
                &self.tree.index.config.cache,
                self.fill,
                &self.prefetched,
                &self.version,
                #[cfg(feature = "metrics")]
                self.tree.metrics(),
                kv,
            )
            .map(|(k, v)| (k, Some(v)))
        } else {
            Ok((kv.key.user_key, None))
        }
    }

    fn key(self) -> crate::Result<UserKey> {
        self.kv.map(|kv| kv.key.user_key)
    }

    fn size(self) -> crate::Result<u32> {
        let kv = self.kv?;

        if kv.key.value_type.is_indirection() {
            let mut cursor = crate::io::Cursor::new(kv.value);
            Ok(BlobIndirection::decode_from(&mut cursor)?.size)
        } else if kv.key.value_type.is_cell_row() {
            field_row::logical_len(&kv.value)
        } else {
            #[expect(clippy::cast_possible_truncation, reason = "values are u32 max length")]
            Ok(kv.value.len() as u32)
        }
    }

    fn into_inner(self) -> crate::Result<(UserKey, UserValue)> {
        if let Some(p) = &self.prefetch {
            p.note_resolved();
        }
        resolve_value_handle(
            self.tree.id(),
            &self.tree.index.config.cache,
            self.fill,
            &self.prefetched,
            &self.version,
            #[cfg(feature = "metrics")]
            self.tree.metrics(),
            self.kv?,
        )
    }
}

/// The value `prefetched` holds for `vhandle`: read ahead for this very item,
/// under its key, so its place and declared size are what identify it.
fn prefetched_value(
    prefetched: &[(crate::vlog::ValueHandle, UserValue)],
    vhandle: &crate::vlog::ValueHandle,
) -> Option<UserValue> {
    prefetched
        .iter()
        .find(|(held, _)| {
            held.blob_file_id == vhandle.blob_file_id
                && held.offset == vhandle.offset
                && held.on_disk_size == vhandle.on_disk_size
        })
        .map(|(_, value)| value.clone())
}

pub(crate) fn resolve_value_handle(
    tree_id: TreeId,
    cache: &crate::Cache,
    fill: crate::config::BlobCacheFill,
    prefetched: &[(crate::vlog::ValueHandle, UserValue)],
    version: &Version,
    #[cfg(feature = "metrics")] metrics: &crate::metrics::Metrics,
    item: InternalValue,
) -> RangeItem {
    if item.key.value_type.is_cell_row() {
        let accessor = Accessor::new(
            &version.blob_files,
            fill,
            #[cfg(feature = "metrics")]
            Some(metrics),
        );
        let value = field_row::resolve_row(&item.value, |vptr| {
            if let Some(value) = prefetched_value(prefetched, &vptr.vhandle) {
                return Ok(value);
            }
            // A reference to an object the version no longer holds is damage,
            // reported as such rather than read as an empty field.
            accessor
                .get(tree_id, &item.key.user_key, &vptr.vhandle, cache)?
                .ok_or(crate::Error::InvalidHeader(
                    "field row: a reference names a blob file the version does not hold",
                ))
        })?;
        return Ok((item.key.user_key, UserValue::from(value)));
    }
    if item.key.value_type.is_indirection() {
        let mut cursor = crate::io::Cursor::new(item.value);
        let vptr = BlobIndirection::decode_from(&mut cursor)?;
        if let Some(value) = prefetched_value(prefetched, &vptr.vhandle) {
            return Ok((item.key.user_key, value));
        }

        // Resolve indirection using value log. No dictionary passed: each blob
        // file carries the one its own descriptor names, pinned for as long as
        // this version keeps the file alive.
        let accessor = Accessor::new(
            &version.blob_files,
            fill,
            #[cfg(feature = "metrics")]
            Some(metrics),
        );

        // An indirection naming a blob file the version does not hold is
        // damage, reported as such, the way a cell row's dangling reference is.
        let value = accessor
            .get(tree_id, &item.key.user_key, &vptr.vhandle, cache)?
            .ok_or(crate::Error::InvalidHeader(
                "blob tree: an indirection names a blob file the version does not hold",
            ))?;
        Ok((item.key.user_key, value))
    } else {
        let k = item.key.user_key;
        let v = item.value;
        Ok((k, v))
    }
}

/// How a batched read gets one separated value: see [`BlobSource::plan_value`].
pub(crate) enum ValuePlan<'v> {
    /// The value, from the cache.
    Ready(UserValue),
    /// One record to read.
    Read(ValueRead<'v>),
    /// Read on the value's own path.
    Alone,
}

/// One blob record a batched read reads: `len` bytes of `blob_file` at the
/// handle's offset.
pub(crate) struct ValueRead<'v> {
    pub(crate) blob_file: &'v BlobFile,
    pub(crate) vhandle: crate::vlog::ValueHandle,
    pub(crate) len: usize,
}

/// Where a blob tree's values kept in the value log are read from: the tree
/// they belong to and the cache their blocks go through.
#[derive(Clone)]
pub(crate) struct BlobSource {
    pub(crate) tree_id: TreeId,
    pub(crate) cache: Arc<crate::Cache>,
    /// Whether the values read through it go into the cache, as the tree's
    /// runtime config said when the read began.
    pub(crate) fill: crate::config::BlobCacheFill,
    #[cfg(feature = "metrics")]
    pub(crate) metrics: Arc<crate::metrics::Metrics>,
}

impl BlobSource {
    /// The value `item` stands for in `version`: read from the value log when
    /// it is an indirection, its own bytes otherwise.
    pub(crate) fn value(&self, version: &Version, item: InternalValue) -> crate::Result<UserValue> {
        resolve_value_handle(
            self.tree_id,
            &self.cache,
            self.fill,
            &[],
            version,
            #[cfg(feature = "metrics")]
            &self.metrics,
            item,
        )
        .map(|(_, value)| value)
    }

    /// How a batched read gets the value the indirection `item` stands for in
    /// `version`: from the cache, by reading one record of a blob file, or on
    /// its own path through [`Self::value`] (a cell row, a reference the
    /// version does not hold, a malformed one), which also reports what is
    /// wrong with it. Nothing here blocks on I/O.
    pub(crate) fn plan_value<'v>(
        &self,
        version: &'v Version,
        item: &InternalValue,
    ) -> ValuePlan<'v> {
        if !item.key.value_type.is_indirection() {
            return ValuePlan::Alone;
        }
        let mut cursor = crate::io::Cursor::new(&item.value[..]);
        let Ok(indirection) = BlobIndirection::decode_from(&mut cursor) else {
            return ValuePlan::Alone;
        };
        let vhandle = indirection.vhandle;
        if let Some(value) = self
            .cache
            .get_blob(self.tree_id, &vhandle, &item.key.user_key)
        {
            return ValuePlan::Ready(value);
        }
        let (Some(blob_file), Ok(len)) = (
            version.blob_files.get(vhandle.blob_file_id),
            crate::vlog::blob_file::reader::record_len(item.key.user_key.len(), &vhandle),
        ) else {
            return ValuePlan::Alone;
        };
        ValuePlan::Read(ValueRead {
            blob_file,
            vhandle,
            len,
        })
    }

    /// `blob_file` open, from the descriptor cache or opened now: it may
    /// block, so a batched read calls it from a job.
    pub(crate) fn open_blob_file(
        &self,
        blob_file: &BlobFile,
    ) -> crate::Result<Arc<dyn crate::fs::FsFile>> {
        let id = crate::GlobalTableId::from((self.tree_id, blob_file.id()));
        Ok(blob_file
            .file_accessor()
            .get_or_open_blob_file(&id, &blob_file.0.path)?
            .0)
    }

    /// The value `read` planned, from the record's `bytes` as read: checked
    /// against `key` and the handle as a single read checks it, counted, and
    /// cached when the read caches what it reads.
    ///
    /// # Errors
    ///
    /// The record fails its checks; the caller then reads the value on its
    /// own path, which reports the failure as a single read would.
    pub(crate) fn finish_value(
        &self,
        version: &Version,
        key: &[u8],
        read: &ValueRead<'_>,
        file: &dyn crate::fs::FsFile,
        bytes: &crate::Slice,
    ) -> crate::Result<UserValue> {
        let accessor = Accessor::new(
            &version.blob_files,
            self.fill,
            #[cfg(feature = "metrics")]
            Some(&self.metrics),
        );
        let reader = crate::vlog::blob_file::reader::Reader::new(read.blob_file, file);
        let mut decoded = 0;
        let parsed = reader.parse_record(key, &read.vhandle, bytes, &mut decoded);
        accessor.count_decoded(decoded);
        let value = parsed?;
        if self.fill.fills() {
            let detached = self
                .cache
                .insert_blob(self.tree_id, &read.vhandle, key, value.clone());
            accessor.count_gather(detached);
        }
        Ok(value)
    }

    /// Counts the read of `read`'s record as issued.
    pub(crate) fn count_issued(&self, version: &Version, read: &ValueRead<'_>) {
        Accessor::new(
            &version.blob_files,
            self.fill,
            #[cfg(feature = "metrics")]
            Some(&self.metrics),
        )
        .count_request(read.len);
    }

    /// The object `indirection` names under `key`, read from `version`'s blob
    /// files.
    ///
    /// # Errors
    ///
    /// Returns an error if the object cannot be read, or `version` holds no
    /// blob file of that id: a reference that names nothing is damage, never
    /// an empty value.
    pub(crate) fn object(
        &self,
        version: &Version,
        key: &[u8],
        indirection: &BlobIndirection,
    ) -> crate::Result<UserValue> {
        Accessor::new(
            &version.blob_files,
            self.fill,
            #[cfg(feature = "metrics")]
            Some(&self.metrics),
        )
        .get(self.tree_id, key, &indirection.vhandle, &self.cache)?
        .ok_or(crate::Error::InvalidHeader(
            "field row: a reference names a blob file the version does not hold",
        ))
    }

    /// [`Self::object`], taken from `local` when [`Self::prefetch`] read it
    /// ahead there.
    #[cfg(feature = "columnar")]
    pub(crate) fn object_prefetched(
        &self,
        version: &Version,
        key: &[u8],
        indirection: &BlobIndirection,
        local: &mut crate::vlog::Prefetched,
    ) -> crate::Result<UserValue> {
        match local.take(key, &indirection.vhandle) {
            Some(object) => Ok(object),
            None => self.object(version, key, indirection),
        }
    }

    /// Reads the objects `items` name (each its key, handle and a slot the
    /// read fills) in as few requests as their places in the blob files
    /// allow, for [`Self::object_prefetched`] to find them: into the cache
    /// when the read caches what it reads, else into `local`. A run of objects
    /// close together in one file is one read. Best effort: an object it does
    /// not read is read on its own when asked for.
    #[cfg(feature = "columnar")]
    pub(crate) fn prefetch(
        &self,
        version: &Version,
        items: &mut Vec<(&[u8], crate::vlog::ValueHandle, usize)>,
        local: &mut crate::vlog::Prefetched,
    ) {
        let into = if self.fill.fills() {
            crate::vlog::PrefetchInto::Cache
        } else {
            crate::vlog::PrefetchInto::Local(local)
        };
        Accessor::new(
            &version.blob_files,
            self.fill,
            #[cfg(feature = "metrics")]
            Some(&self.metrics),
        )
        .prefetch(
            self.tree_id,
            items,
            &self.cache,
            PREFETCH_MAX_GAP,
            PREFETCH_MAX_READ,
            into,
        );
    }
}

/// A key-value-separated log-structured merge tree
///
/// This tree is a composite structure, consisting of an
/// index tree (LSM-tree) and a log-structured value log
/// to reduce write amplification.
///
/// Closed like a [`Tree`](crate::Tree): once the [`TreeClose`](crate::TreeClose)
/// of [`close`](Self::close) fires, the tree writes nothing into its directory,
/// its blob folder included.
#[derive(Clone)]
pub struct BlobTree {
    /// Index tree that holds value handles or small inline values
    #[doc(hidden)]
    pub index: crate::Tree,

    pub(crate) blobs_folder: Arc<PathBuf>,
}

impl BlobTree {
    /// A handle that does not keep the tree open; see [`WeakTree`](crate::WeakTree).
    #[must_use]
    pub fn downgrade(&self) -> crate::WeakTree {
        crate::WeakTree::of(&self.index, Some(Arc::clone(&self.blobs_folder)))
    }

    /// Gives up this handle and returns the signal of the tree being closed;
    /// see [`Tree::close`](crate::Tree::close). A blob file an iterator or a
    /// snapshot still holds keeps the tree open, as a table does.
    pub fn close(self) -> crate::TreeClose {
        crate::close::close(self.index)
    }

    /// Physical footprint of the stale blob files a full compaction would
    /// relocate: the linked, stale, non-dead subset across every live table (the
    /// SAME set the merge gate budgets via `pick_blob_files_to_rewrite`), summed
    /// by on-disk size. Zero when KV separation is unconfigured.
    fn full_compaction_blob_need(&self, version: &crate::version::Version) -> crate::Result<u64> {
        let Some(blob_opts) = &self.index.config.kv_separation_opts else {
            return Ok(0);
        };
        let all_tables: crate::HashSet<TableId> = version.iter_tables().map(Table::id).collect();
        crate::compaction::worker::pick_blob_files_to_rewrite(&all_tables, version, blob_opts)?
            .iter()
            .try_fold(0u64, |acc, bf| bf.physical_size().map(|size| acc + size))
    }

    /// Opens the tree, coming by the directory lock as `lock` says; see
    /// [`Tree::open`](crate::Tree::open), and when this open made it, what
    /// takes it back.
    pub(crate) fn open(
        config: Config,
        #[cfg(feature = "std")] lock: crate::config::DirectoryLock,
        names: crate::tree::NewFolderNames,
    ) -> crate::Result<(Self, Option<crate::tree::Creation>)> {
        use crate::file::BLOBS_FOLDER;

        let (index, creation) = crate::Tree::open(
            config,
            #[cfg(feature = "std")]
            lock,
            names,
        )?;

        // The folder's name is an entry of the tree folder, and a blob file in
        // it must not be named by the manifest before that entry is durable.
        // A new tree made the folder before its first snapshot, whose
        // directory sync covered the name, and an opened one syncs the tree
        // folder at its first edit append (the log found at open is not
        // trusted to have a durable name), which is the only way a blob file
        // comes to be named. A name found here may not be durable yet (a
        // process that died before its sync leaves one the page cache still
        // shows), and that first append covers it. A folder this open makes,
        // though, may come after the first append: the open itself installs
        // an edit when it registers a dictionary. Its name is synced here.
        let blobs_folder = index.config.path.join(BLOBS_FOLDER);
        let folder = (|| -> crate::Result<()> {
            let made = !(*index.config.fs).exists(&blobs_folder)?;
            (*index.config.fs).create_dir_all(&blobs_folder)?;
            if made {
                crate::file::fsync_directory(
                    &index.config.path,
                    &*index.config.fs,
                    index.config.sync_mode,
                )?;
            }
            Ok(())
        })();
        // A tree this open made, whose open then fails, is taken back: the
        // caller is told it was not opened, and a retry makes it anew.
        if let Err(e) = folder {
            if let Some(creation) = creation {
                creation.take_back(index);
            }
            return Err(e);
        }

        let blob_file_id_to_continue_with = index
            .current_version()
            .blob_files
            .list_ids()
            .max()
            .map(|x| x + 1)
            .unwrap_or_default();

        index
            .0
            .blob_file_id_counter
            .set(blob_file_id_to_continue_with);

        Ok((
            Self {
                index,
                blobs_folder: Arc::new(blobs_folder),
            },
            creation,
        ))
    }

    /// Resolves a single key against a pre-acquired [`SuperVersion`](crate::version::SuperVersion),
    /// reading its value through `source`, taken as the read began, so every
    /// key of one read caches as the tree said then.
    fn resolve_key(
        &self,
        super_version: &crate::version::SuperVersion,
        source: &BlobSource,
        key: &[u8],
        seqno: SeqNo,
    ) -> crate::Result<Option<UserValue>> {
        let Some(item) = crate::Tree::get_internal_entry_from_version(
            super_version,
            key,
            seqno,
            self.index.config.comparator.as_ref(),
        )?
        else {
            return Ok(None);
        };
        // The newest version is an operand: the key's value is its operands
        // merged onto its base, as a batch of keys resolves it. Without a
        // merge operator the operand is the value.
        if item.key.value_type.is_merge_operand()
            && let Some(merge_operator) = &self.index.config.merge_operator
        {
            return crate::Tree::resolve_point_merge(
                super_version,
                key,
                seqno,
                merge_operator,
                Some(source),
            );
        }

        source.value(&super_version.version, item).map(Some)
    }

    /// Iterate change events with `seqno >= target_seqno`, resolving
    /// KV-separated values.
    ///
    /// Same change-data-capture contract as [`Tree::scan_since_seqno`](crate::Tree::scan_since_seqno),
    /// but a blob-indirected value is resolved from its blob file into a
    /// [`ScanSinceEvent::Insert`] carrying the real value, so a downstream
    /// consumer can replicate without access to the source's blob files.
    /// Block-skip, seqno ordering, and tombstone handling are identical to the
    /// standard-tree path (the same shared aggregation backs both).
    ///
    /// # Panics
    ///
    /// Panics if the internal version-history lock is poisoned.
    ///
    /// # Errors
    ///
    /// Returns `Err` if reading the index, a data block, or a referenced blob
    /// fails.
    pub fn scan_since_seqno(
        &self,
        target_seqno: SeqNo,
    ) -> crate::Result<impl Iterator<Item = ScanSinceEvent> + use<>> {
        let fill = self.blob_cache_fill();
        self.index
            .scan_since_seqno_with(target_seqno, true, |version, entry| {
                let seqno = entry.key.seqno;
                let (key, value) = resolve_value_handle(
                    self.id(),
                    &self.index.config.cache,
                    fill,
                    &[],
                    version,
                    #[cfg(feature = "metrics")]
                    self.metrics(),
                    entry,
                )?;
                Ok(ScanSinceEvent::Insert { key, value, seqno })
            })
    }

    /// Starts a multi-get of `keys` at `seqno` that its caller drives, as
    /// [`Tree::start_multi_get`](crate::Tree::start_multi_get) does; a value
    /// kept in the value log is read by a job the read hands out.
    ///
    /// # Errors
    ///
    /// Returns [`crate::Error::SnapshotBelowRetention`] when `seqno` is below
    /// what the tree retains.
    pub fn start_multi_get<K: Into<UserKey>>(
        &self,
        keys: impl IntoIterator<Item = K>,
        seqno: SeqNo,
    ) -> crate::Result<crate::resumable::Step> {
        let snapshot = self.index.snapshot_for_read(seqno)?;
        let seqno = crate::Tree::read_ceiling(&snapshot, seqno);
        let keys: Vec<UserKey> = keys.into_iter().map(Into::into).collect();
        // Each key a point lookup, counted once as the read starts, not on
        // every resume.
        snapshot
            .version
            .read_mix()
            .note_points(u64::try_from(keys.len()).unwrap_or(u64::MAX));
        // One source for the whole read: every key caches as the tree said
        // when the read began.
        let source = self.blob_source();
        Ok(crate::resumable::ResumableMultiGet::start(
            crate::tree::read_job::ReadCtx {
                super_version: crate::version::SnapshotRef::Owned((*snapshot).clone()),
                keys,
                seqno,
                comparator: Arc::clone(&self.index.config.comparator),
                merge_operator: self.index.config.merge_operator.clone(),
                merge_base: Some(source.clone()),
                values: crate::tree::read_job::Values::Blob(source),
                metadata_budget: self.index.config.multi_get_metadata_budget,
            },
        ))
    }

    /// Whether a read beginning now caches the values it reads from blob
    /// files.
    fn blob_cache_fill(&self) -> crate::config::BlobCacheFill {
        self.index.runtime_config.load().blob_cache_fill
    }

    /// Where this tree's values kept in the value log are read from.
    fn blob_source(&self) -> BlobSource {
        BlobSource {
            tree_id: self.id(),
            cache: Arc::clone(&self.index.config.cache),
            fill: self.blob_cache_fill(),
            #[cfg(feature = "metrics")]
            metrics: Arc::clone(self.metrics()),
        }
    }

    /// Range-scoped variant of [`Self::scan_since_seqno`], with the same
    /// contract as
    /// [`Tree::scan_since_seqno_in_range`](crate::Tree::scan_since_seqno_in_range):
    /// point events only for keys inside `range`, range deletions when their
    /// span overlaps it, and index SSTs outside the bounds skipped unread.
    ///
    /// # Panics
    ///
    /// Panics if the internal version-history lock is poisoned.
    ///
    /// # Errors
    ///
    /// Same as [`Self::scan_since_seqno`].
    pub fn scan_since_seqno_in_range<K: AsRef<[u8]>, R: core::ops::RangeBounds<K>>(
        &self,
        target_seqno: SeqNo,
        range: R,
    ) -> crate::Result<impl Iterator<Item = ScanSinceEvent> + use<K, R>> {
        let bounds = crate::tree::range_to_user_bounds(&range);
        let fill = self.blob_cache_fill();
        self.index.scan_since_seqno_scoped(
            target_seqno,
            true,
            |version, entry| {
                let seqno = entry.key.seqno;
                let (key, value) = resolve_value_handle(
                    self.id(),
                    &self.index.config.cache,
                    fill,
                    &[],
                    version,
                    #[cfg(feature = "metrics")]
                    self.metrics(),
                    entry,
                )?;
                Ok(ScanSinceEvent::Insert { key, value, seqno })
            },
            Some(&bounds),
        )
    }

    /// Writes `key` as a row of `fields` at `seqno`.
    ///
    /// Each [`Field`](field_row::Field) names its column and physical type,
    /// and both are stored with the row, so a stored row always reads with
    /// the layout it was written with. A column id is a field id of the
    /// columnar format (from [`FIRST_FIELD_COLUMN`](field_row::FIRST_FIELD_COLUMN)
    /// up to, not including, [`RESERVED_COLUMNS`](field_row::RESERVED_COLUMNS)),
    /// so a columnar table can store the
    /// field as that column. The row keeps its fields in ascending column
    /// order, whatever order they are given in. Plain reads of the key return
    /// the fields in that order, framed as the columnar format frames a row's value
    /// sub-columns (a variable-width field as a little-endian `u32` length and
    /// its bytes, a fixed-width one as its bytes), with every
    /// [`Cell::Ref`](field_row::Cell::Ref) replaced by its object. A
    /// [`Cell::Value`](field_row::Cell::Value) at or above its column's
    /// threshold
    /// ([`KvSeparationOptions::cell_separation_threshold`](crate::KvSeparationOptions::cell_separation_threshold),
    /// else the tree's separation threshold) is stored in a blob file when the
    /// row is flushed.
    ///
    /// A [`Cell::Ref`](field_row::Cell::Ref) keeps an object the key already
    /// holds without rewriting it, which is how a metadata-only update leaves
    /// a large field in place. Take the references from the key's latest
    /// version and write this row above it.
    ///
    /// Returns the added item's size and the new memtable size, as
    /// [`AbstractTree::insert`] does.
    ///
    /// # Errors
    ///
    /// Returns [`crate::Error::BlobRef`] if a reference was read from another
    /// key or tree, two fields name one object, or a reference is stale (its
    /// object was moved or let go of since the read: read the key again);
    /// [`crate::Error::CellRow`] if a column is not a field id, two fields
    /// share a column or a fixed-width field is not its type's width; and an
    /// error if the row has
    /// more than `u16::MAX` fields or a cell exceeds `u32::MAX` bytes. Nothing
    /// is written then.
    ///
    /// # Examples
    ///
    /// ```
    /// # use lsm_tree::{AbstractTree, Config, KvSeparationOptions, SequenceNumberCounter};
    /// # use lsm_tree::blob_tree::field_row::Field;
    /// # let folder = lsm_tree::get_tmp_folder();
    /// let tree = Config::new(folder.path(), SequenceNumberCounter::default(), SequenceNumberCounter::default())
    ///     .with_kv_separation(Some(KvSeparationOptions::default()))
    ///     .open()?;
    /// # let lsm_tree::AnyTree::Blob(tree) = tree else { unreachable!() };
    /// // Given out of order, stored and read in column order.
    /// tree.insert_cells("doc", &[Field::bytes(4, b"body"), Field::bytes(3, b"draft")], 0)?;
    /// let value = tree.get("doc", 1)?.expect("written");
    /// assert_eq!(&*value, b"\x05\0\0\0draft\x04\0\0\0body");
    /// # Ok::<(), lsm_tree::Error>(())
    /// ```
    pub fn insert_cells<K: Into<UserKey>>(
        &self,
        key: K,
        fields: &[field_row::Field<'_>],
        seqno: SeqNo,
    ) -> crate::Result<(u64, u64)> {
        let key = key.into();
        let mut row = Vec::with_capacity(fields.len());
        for field in fields {
            let cell = match &field.cell {
                field_row::Cell::Value(bytes) => field_row::RowCell::Value(bytes),
                field_row::Cell::Ref(reference) => {
                    if reference.tree != self.id() {
                        return Err(crate::Error::BlobRef("reference read from another tree"));
                    }
                    if !crate::comparator::same_user_key(reference.key, &key) {
                        return Err(crate::Error::BlobRef("reference read from another key"));
                    }
                    // Written back by a caller, a reference never owns its
                    // object: the version that wrote the object, or the one a
                    // compaction moved ownership to, still does.
                    field_row::RowCell::Ref {
                        indirection: reference.indirection,
                        owner: false,
                    }
                }
            };
            row.push(field_row::RowField {
                column: field.column,
                tag: field.tag,
                cell,
            });
        }
        field_row::order_fields(&mut row)?;
        let refs = fields.iter().filter_map(|field| match &field.cell {
            field_row::Cell::Ref(reference) => Some(reference),
            field_row::Cell::Value(_) => None,
        });
        for (i, a) in refs.clone().enumerate() {
            if refs.clone().skip(i + 1).any(|b| a == b) {
                return Err(crate::Error::BlobRef("one object in two fields"));
            }
        }
        let value = field_row::encode_row(&row)?;
        let sourced: Vec<(BlobIndirection, u64)> = refs
            .map(|reference| (reference.indirection, reference.source))
            .collect();
        // The caller's bytes, as an ingested row counts them: the key and the
        // value fields, not the row's encoding; a reference brings none.
        #[cfg(feature = "metrics")]
        let payload = key.len()
            + fields
                .iter()
                .map(|field| match &field.cell {
                    field_row::Cell::Value(bytes) => bytes.len(),
                    field_row::Cell::Ref(_) => 0,
                })
                .sum::<usize>();
        let written = self.index.append_cell_row(
            InternalValue::from_components(key, value, seqno, crate::ValueType::CellRow),
            &sourced,
        )?;
        // Charged once the row is in: a row refused for a stale reference is
        // written again, and counted, by the retry.
        #[cfg(feature = "metrics")]
        self.index.0.metrics.accepted_value(payload);
        Ok(written)
    }

    /// The fields of `key` as of `seqno`, references included as references:
    /// the read a metadata-only update starts from. No blob object is read.
    ///
    /// Returns `None` when the key is absent at `seqno`. References to write
    /// back must come from the key's latest version; a reference read at an
    /// older `seqno` may name an object a newer version already let go of.
    ///
    /// # Errors
    ///
    /// Returns [`crate::Error::SnapshotBelowRetention`] when `seqno` is below
    /// what the tree retains, and [`crate::Error::BlobRef`] when the key's
    /// version at `seqno` was not written as cells.
    ///
    /// # Examples
    ///
    /// ```
    /// # use lsm_tree::{AbstractTree, Config, KvSeparationOptions, SequenceNumberCounter};
    /// # use lsm_tree::blob_tree::field_row::{Cell, Field};
    /// # let folder = lsm_tree::get_tmp_folder();
    /// # let tree = Config::new(folder.path(), SequenceNumberCounter::default(), SequenceNumberCounter::default())
    /// #     .with_kv_separation(Some(KvSeparationOptions::default().separation_threshold(16)))
    /// #     .open()?;
    /// # let lsm_tree::AnyTree::Blob(tree) = tree else { unreachable!() };
    /// let body = vec![b'x'; 1_000];
    /// tree.insert_cells("doc", &[Field::bytes(3, b"draft"), Field::bytes(4, &body)], 0)?;
    /// tree.flush_active_memtable(0)?;
    ///
    /// // Change the status, keep the body where it is.
    /// let row = tree.get_cells("doc", 1)?.expect("written");
    /// let mut fields = row.fields()?;
    /// assert!(matches!(fields[1].cell, Cell::Ref(_)));
    /// fields[0].cell = Cell::Value(b"final");
    /// tree.insert_cells("doc", &fields, 1)?;
    /// # Ok::<(), lsm_tree::Error>(())
    /// ```
    pub fn get_cells<K: AsRef<[u8]>>(
        &self,
        key: K,
        seqno: SeqNo,
    ) -> crate::Result<Option<field_row::RowCells>> {
        let read = self.cell_read(seqno)?;
        read.version.version.read_mix().note_points(1);
        let Some(item) = crate::Tree::get_internal_entry_from_version(
            &read.version,
            key.as_ref(),
            seqno,
            self.index.config.comparator.as_ref(),
        )?
        else {
            return Ok(None);
        };
        if !item.key.value_type.is_cell_row() {
            return Err(crate::Error::BlobRef("the key's version is not a cell row"));
        }
        Ok(Some(field_row::RowCells {
            key: item.key.user_key,
            row: item.value,
            read: Arc::new(read),
        }))
    }

    /// The rows written as cells in `range` as of `seqno`, in key order, each
    /// with its references as references, as [`Self::get_cells`] reads one.
    ///
    /// A key whose version at `seqno` is stored whole rather than as cells is
    /// passed over: read it with a plain range read. Every row comes from one
    /// read, which keeps its objects readable and its references writable as
    /// [`RowCells`](field_row::RowCells) describes, for as long as any of its
    /// rows is held.
    ///
    /// # Errors
    ///
    /// Returns [`crate::Error::SnapshotBelowRetention`] when `seqno` is below
    /// what the tree retains; the iterator yields an error for a row it
    /// cannot read.
    ///
    /// # Examples
    ///
    /// ```
    /// # use lsm_tree::{AbstractTree, Config, KvSeparationOptions, SeqNo, SequenceNumberCounter};
    /// # use lsm_tree::blob_tree::field_row::{Cell, Field};
    /// # let folder = lsm_tree::get_tmp_folder();
    /// # let tree = Config::new(folder.path(), SequenceNumberCounter::default(), SequenceNumberCounter::default())
    /// #     .with_kv_separation(Some(KvSeparationOptions::default().separation_threshold(16)))
    /// #     .open()?;
    /// # let lsm_tree::AnyTree::Blob(tree) = tree else { unreachable!() };
    /// let body = vec![b'x'; 1_000];
    /// for key in ["a", "b"] {
    ///     tree.insert_cells(key, &[Field::bytes(3, b"draft"), Field::bytes(4, &body)], 0)?;
    /// }
    /// tree.insert("plain", "value", 0);
    /// tree.flush_active_memtable(0)?;
    ///
    /// // Publish every draft, keeping each body where it is.
    /// let rows: Vec<_> = tree.range_cells::<&str, _>(.., 1)?.collect::<lsm_tree::Result<_>>()?;
    /// assert_eq!(rows.len(), 2);
    /// for row in &rows {
    ///     let mut fields = row.fields()?;
    ///     fields[0].cell = Cell::Value(b"final");
    ///     tree.insert_cells(row.key().clone(), &fields, 1)?;
    /// }
    /// # Ok::<(), lsm_tree::Error>(())
    /// ```
    pub fn range_cells<K: AsRef<[u8]>, R: RangeBounds<K>>(
        &self,
        range: R,
        seqno: SeqNo,
    ) -> crate::Result<
        Box<dyn DoubleEndedIterator<Item = crate::Result<field_row::RowCells>> + Send + 'static>,
    > {
        let read = Arc::new(self.cell_read(seqno)?);
        let rows = crate::Tree::create_internal_range_with_prefix_hash(
            read.version.clone(),
            &range,
            seqno,
            None,
            self.index.config.merge_operator.clone(),
            self.index.config.comparator.clone(),
            None,
            Some(self.blob_source()),
        );
        Ok(Box::new(rows.filter_map(move |item| match item {
            Ok(item) if item.key.value_type.is_cell_row() => Some(Ok(field_row::RowCells {
                key: item.key.user_key,
                row: item.value,
                read: Arc::clone(&read),
            })),
            Ok(_) => None,
            Err(e) => Some(Err(e)),
        })))
    }

    /// A reference-aware read as of `seqno`: the latest version and the
    /// read's registration, taken together under the history's read lock. An
    /// install records its releases under the write lock, so none newer than
    /// this version can go unrecorded.
    fn cell_read(&self, seqno: SeqNo) -> crate::Result<field_row::CellRead> {
        let history = self.index.version_history.read();
        let version = history.latest_version_ref().clone();
        version.check_serves(seqno)?;
        let token = history.released().register(version.version.id());
        drop(history);
        Ok(field_row::CellRead {
            tree: self.id(),
            version,
            source: self.blob_source(),
            token,
        })
    }

    /// Runs a projected columnar scan across the tree, as
    /// [`Tree::columnar_scan`](crate::Tree::columnar_scan) does, reading each
    /// row's declared fields as late as it can.
    ///
    /// The rows are decided first on their keys, seqnos and value types; only
    /// the rows the scan returns have their fields read. A field of a row
    /// written as cells is read out of the row (or, in a columnar table, out
    /// of the field's own column), and its object is read from its blob file
    /// only when the field is projected: a projection of compact fields reads
    /// no blob at all. A row stored whole has its fields read through the
    /// projection's [`ValueProjector`](crate::ValueProjector), its value read
    /// from the value log first when it is kept there. When the predicate
    /// runs on a projected field, that field is read first and the objects of
    /// the other fields are read only for the rows it keeps.
    ///
    /// A field is projected as a declared field, with its type and what its
    /// absence means; the intrinsic key, seqno and value-type columns may be
    /// projected by id. A field holds its value, not how the tree stores it,
    /// so a row's value type reads as a value.
    ///
    /// # Errors
    ///
    /// Returns [`crate::Error::Projection`] when a value column is projected
    /// by id alone or the predicate runs on one no field declares, and
    /// otherwise as [`Tree::columnar_scan`](crate::Tree::columnar_scan), and,
    /// lazily, when a field is stored under another type than declared or an
    /// object cannot be read.
    ///
    /// # Examples
    ///
    /// ```
    /// # use lsm_tree::{AbstractTree, Config, KvSeparationOptions, SeqNo, SequenceNumberCounter};
    /// use lsm_tree::blob_tree::field_row::{Field, TypeTag};
    /// use lsm_tree::{Absent, ProjectedField, Projection};
    /// # let folder = lsm_tree::get_tmp_folder();
    /// # let tree = Config::new(folder.path(), SequenceNumberCounter::default(), SequenceNumberCounter::default())
    /// #     .with_kv_separation(Some(KvSeparationOptions::default().separation_threshold(16)))
    /// #     .open()?;
    /// # let lsm_tree::AnyTree::Blob(tree) = tree else { unreachable!() };
    /// let body = vec![b'x'; 1_000];
    /// tree.insert_cells("doc", &[Field::bytes(3, b"draft"), Field::bytes(4, &body)], 0)?;
    /// tree.flush_active_memtable(0)?;
    ///
    /// // The status alone: the body stays in its blob file, unread.
    /// let status = Projection::new().field(ProjectedField::new(3, TypeTag::Bytes, Absent::Null)?);
    /// let mut batches = tree.columnar_scan(status, None, SeqNo::MAX, ..)?;
    /// let batch = batches.next().expect("one batch")?;
    /// assert_eq!(batch.row_count, 1);
    /// # Ok::<(), lsm_tree::Error>(())
    /// ```
    #[cfg(feature = "columnar")]
    pub fn columnar_scan<R: core::ops::RangeBounds<UserKey>>(
        &self,
        projection: impl Into<crate::Projection>,
        predicate: Option<&crate::table::columnar_predicate::ColumnRangePredicate>,
        seqno: SeqNo,
        range: R,
    ) -> crate::Result<crate::tree::columnar_scan::ColumnarScan> {
        self.index
            .columnar_scan(projection, predicate, seqno, range)
    }

    /// Updates the live [`RuntimeConfig`](crate::runtime_config::RuntimeConfig),
    /// as [`Tree::update_runtime_config`](crate::Tree::update_runtime_config)
    /// does for the index tree.
    ///
    /// The blob files share the snapshot: a change of
    /// [`blob_compression`](crate::runtime_config::RuntimeConfig::blob_compression)
    /// applies to the next blob file written, and every earlier one keeps the
    /// codec it recorded.
    ///
    /// # Errors
    ///
    /// Same as [`Tree::update_runtime_config`](crate::Tree::update_runtime_config),
    /// including a blob compression that names a dictionary the tree does not
    /// hold.
    pub fn update_runtime_config<F>(&self, mutator: F) -> crate::Result<()>
    where
        F: FnOnce(&mut crate::runtime_config::RuntimeConfig),
    {
        self.index.update_runtime_config(mutator)
    }

    /// Snapshot of the current runtime config, shared by the index and the
    /// blob files.
    #[must_use]
    pub fn runtime_config(&self) -> Arc<crate::runtime_config::RuntimeConfig> {
        self.index.runtime_config()
    }

    /// Stores `dict` in the tree, as
    /// [`Tree::register_zstd_dictionary`](crate::Tree::register_zstd_dictionary)
    /// does. The step before a blob compression that names it: an update
    /// naming a dictionary the tree does not hold is refused.
    ///
    /// # Errors
    ///
    /// Same as [`Tree::register_zstd_dictionary`](crate::Tree::register_zstd_dictionary).
    #[cfg(zstd_any)]
    pub fn register_zstd_dictionary(
        &self,
        dict: Arc<crate::compression::ZstdDictionary>,
    ) -> crate::Result<()> {
        self.index.register_zstd_dictionary(dict)
    }
}

impl crate::abstract_tree::sealed::Sealed for BlobTree {}

/// Wasted bytes between two blob records worth reading through to merge their
/// reads. One record's header plus a short key is under a hundred bytes, so a
/// gap this size means the records are effectively adjacent: the skipped bytes
/// cost less than the second read's setup would.
const PREFETCH_MAX_GAP: u64 = 4 * 1_024;

/// Cap on a single coalesced prefetch read. Bounds the transient buffer a scan
/// holds regardless of how large the configured window is or how big the values
/// in it turn out to be.
const PREFETCH_MAX_READ: usize = 4 * 1_024 * 1_024;

impl BlobTree {
    /// Read-ahead state for one scan, seeded from the tree's configuration,
    /// caching under `fill` as the scan's source does.
    fn scan_prefetch_state(&self, fill: crate::config::BlobCacheFill) -> Arc<ScanPrefetch> {
        let window = self
            .index
            .config
            .kv_separation_opts
            .as_ref()
            .map_or(0, |o| o.scan_prefetch as usize);

        Arc::new(ScanPrefetch {
            resolving: core::sync::atomic::AtomicBool::new(false),
            held: core::sync::atomic::AtomicUsize::new(0),
            fill,
            window,
            max_gap: PREFETCH_MAX_GAP,
            max_read: PREFETCH_MAX_READ,
        })
    }
}

/// The values `kv` reads from blob files that a scan read ahead into `local`,
/// taken out of it for `kv`'s guard: its own value, or its cells' objects.
fn take_prefetched(
    local: &mut crate::vlog::Prefetched,
    kv: &InternalValue,
) -> Vec<(crate::vlog::ValueHandle, UserValue)> {
    let key = &kv.key.user_key;
    if kv.key.value_type.is_cell_row() {
        let Ok(refs) = field_row::row_refs(&kv.value) else {
            return Vec::new();
        };
        return refs
            .into_iter()
            .filter_map(|(vptr, _)| {
                local
                    .take(key, &vptr.vhandle)
                    .map(|value| (vptr.vhandle, value))
            })
            .collect();
    }
    if !kv.key.value_type.is_indirection() {
        return Vec::new();
    }
    let mut cursor = crate::io::Cursor::new(&kv.value[..]);
    let Ok(vptr) = BlobIndirection::decode_from(&mut cursor) else {
        return Vec::new();
    };
    local
        .take(key, &vptr.vhandle)
        .map(|value| alloc::vec![(vptr.vhandle, value)])
        .unwrap_or_default()
}

/// Maps a raw merge-pipeline item into a KV-separated iterator guard that
/// resolves the blob handle lazily against `version`.
fn blob_guard(
    tree: &crate::BlobTree,
    version: &Version,
    fill: crate::config::BlobCacheFill,
    item: crate::Result<InternalValue>,
) -> IterGuardImpl {
    IterGuardImpl::Blob(Guard {
        tree: tree.clone(),
        version: version.clone(),
        fill,
        prefetched: Vec::new(),
        kv: item,
        prefetch: None,
        _held: None,
    })
}

/// Whether a scan's caller is actually resolving separated values, shared
/// between the scan iterator and the guards it has handed out.
///
/// A guard resolves its value only if the caller asks: `key()` never touches
/// the blob file, and `into_inner_if` can reject an item on its key first.
/// Reading ahead unconditionally would therefore charge a key-only scan for
/// values nobody wants. So the guards report the first resolution, and only
/// then does the iterator start reading ahead: a scan that never asks for a
/// value never pays, and one that asks for every value pays a solo read just
/// for the first.
pub(crate) struct ScanPrefetch {
    resolving: core::sync::atomic::AtomicBool,
    /// Bytes of values read ahead into the scan's own store that guards it
    /// handed out still hold, unresolved. Within two reads, so a `usize`
    /// holds it on every target, 32-bit ones without 64-bit atomics too.
    held: core::sync::atomic::AtomicUsize,
    /// Whether the scan caches the values it reads, as the tree said when the
    /// scan began; when it does not, it keeps what it reads ahead itself.
    fill: crate::config::BlobCacheFill,
    /// How many upcoming items to gather per read-ahead.
    window: usize,
    /// Wasted bytes between two records worth swallowing to merge their reads.
    max_gap: u64,
    /// Cap on one coalesced read.
    max_read: usize,
}

impl ScanPrefetch {
    /// Records that the caller resolved a value, arming the read-ahead.
    pub(crate) fn note_resolved(&self) {
        self.resolving
            .store(true, core::sync::atomic::Ordering::Relaxed);
    }

    /// Whether the next item should come from a read-ahead window. A zero
    /// window is checked here so a disabled knob costs one load per item
    /// instead of an empty refill.
    fn armed(&self) -> bool {
        self.window > 0 && self.resolving.load(core::sync::atomic::Ordering::Relaxed)
    }
}

/// A KV-separated scan that reads its upcoming values ahead in coalesced runs.
///
/// Sits between the merge pipeline and the guards: it buffers a window of
/// upcoming items, warms their blobs through
/// [`Accessor::prefetch`](crate::vlog::Accessor::prefetch), and hands them out
/// one at a time. Guards resolve exactly as they always did, and hit the cache
/// instead of the disk.
struct PrefetchScan<I> {
    inner: I,
    /// Items pulled from the FRONT of `inner` and not yet handed out.
    buf: alloc::collections::VecDeque<crate::Result<InternalValue>>,
    /// Set once `inner` has yielded its last item from the back, which is when
    /// reverse iteration has to start draining `buf` instead.
    inner_back_done: bool,
    /// Latched copy of [`ScanPrefetch::armed`]. The shared flag only ever goes
    /// false to true, so once this iterator has seen it there is no reason to
    /// keep loading the atomic for every item it yields.
    armed: bool,
    state: Arc<ScanPrefetch>,
    tree: crate::BlobTree,
    version: Version,
    /// The values of the window read ahead when the scan does not cache
    /// them, until the guards of their items take them.
    local: crate::vlog::Prefetched,
}

impl<I: Iterator<Item = crate::Result<InternalValue>>> PrefetchScan<I> {
    /// Pulls the next window into `buf` and warms the values it points at.
    fn fill(&mut self) {
        // What the window before left untaken (an item whose guard was
        // dropped, a value it held twice) is no use to this one.
        self.local.clear();
        // Sized here rather than at construction: a scan that never resolves a
        // value never refills, and should not pay for a window it will not use.
        // After the first refill the capacity persists, so this is a no-op.
        self.buf.reserve(self.state.window);
        for _ in 0..self.state.window {
            match self.inner.next() {
                Some(item) => self.buf.push_back(item),
                None => break,
            }
        }

        // Decode the handles of the items that actually carry one. A window is
        // usually all-indirection or all-inline (separation is by value size),
        // so this is either the whole window or nothing. Sized up front: the
        // window is the only allocation a refill makes, and `prefetch` filters
        // and sorts it in place rather than building its own copy.
        //
        // The indirection is decoded from a BORROWED view of the value. Cloning
        // it would bump a refcount per item on the scan's hottest loop, for a
        // cursor that is dropped three lines later.
        let mut handles: Vec<(&[u8], crate::vlog::ValueHandle, usize)> =
            Vec::with_capacity(self.buf.len());
        for item in &self.buf {
            let Ok(kv) = item else { continue };
            if kv.key.value_type.is_cell_row() {
                // A cell row's references are read when the row resolves, as
                // an indirection is; warming them is the same read-ahead.
                if let Ok(refs) = field_row::row_refs(&kv.value) {
                    for (vptr, _) in refs {
                        handles.push((&kv.key.user_key, vptr.vhandle, 0));
                    }
                }
                continue;
            }
            if !kv.key.value_type.is_indirection() {
                continue;
            }
            let mut cursor = crate::io::Cursor::new(&kv.value[..]);
            let Ok(vptr) = BlobIndirection::decode_from(&mut cursor) else {
                continue;
            };
            handles.push((&kv.key.user_key, vptr.vhandle, 0));
        }

        if handles.len() < 2 {
            return;
        }

        let accessor = Accessor::new(
            &self.version.blob_files,
            self.state.fill,
            #[cfg(feature = "metrics")]
            Some(self.tree.metrics()),
        );

        // Into the cache when the scan caches what it reads, else into its own
        // store, where the guards of the window's items take their values.
        // While the guards handed out still hold a read's worth unresolved (a
        // caller keeping them, say), the scan reads nothing more ahead: its
        // memory stays bounded by two reads, not by the scan, and the guards
        // that follow read their values when resolved.
        let into = if self.state.fill.fills() {
            crate::vlog::PrefetchInto::Cache
        } else if self.state.held.load(core::sync::atomic::Ordering::Relaxed) >= self.state.max_read
        {
            return;
        } else {
            crate::vlog::PrefetchInto::Local(&mut self.local)
        };
        accessor.prefetch(
            self.tree.id(),
            &mut handles,
            &self.tree.index.config.cache,
            self.state.max_gap,
            self.state.max_read,
            into,
        );
    }

    fn guard(&mut self, item: crate::Result<InternalValue>) -> IterGuardImpl {
        let prefetched = match &item {
            Ok(kv) if !self.local.is_empty() => take_prefetched(&mut self.local, kv),
            _ => Vec::new(),
        };
        let held = (!prefetched.is_empty()).then(|| {
            let bytes = prefetched.iter().map(|(_, value)| value.len()).sum();
            self.state
                .held
                .fetch_add(bytes, core::sync::atomic::Ordering::Relaxed);
            HeldReadAhead {
                state: Arc::clone(&self.state),
                bytes,
            }
        });
        IterGuardImpl::Blob(Guard {
            tree: self.tree.clone(),
            version: self.version.clone(),
            fill: self.state.fill,
            prefetched,
            _held: held,
            kv: item,
            // Only until the read-ahead is armed, and never with the window
            // disabled. The guards exist to report the FIRST resolution; once
            // that has happened (or can change nothing, because a zero window
            // ignores it), handing every guard a counted reference would be an
            // atomic bump per item of the scan for no effect, plus an atomic
            // store when the guard resolves.
            prefetch: if self.armed || self.state.window == 0 {
                None
            } else {
                Some(self.state.clone())
            },
        })
    }
}

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

    fn next(&mut self) -> Option<Self::Item> {
        if self.buf.is_empty() {
            if !self.armed {
                self.armed = self.state.armed();
            }
            if self.armed {
                self.fill();
            }
        }

        let item = match self.buf.pop_front() {
            Some(item) => item,
            None => self.inner.next()?,
        };
        Some(self.guard(item))
    }
}

impl<I: DoubleEndedIterator<Item = crate::Result<InternalValue>>> DoubleEndedIterator
    for PrefetchScan<I>
{
    fn next_back(&mut self) -> Option<Self::Item> {
        // Latch the shared armed state here too: read-ahead stays forward-only
        // (no fill from this end), but once a reverse guard has resolved there
        // is nothing left for later guards to report, and without the latch
        // every one of them would carry a counted reference (an atomic bump on
        // emit, an atomic store on resolve) for no effect.
        if !self.armed {
            self.armed = self.state.armed();
        }
        // `buf` holds the EARLIEST remaining items (they were pulled off the
        // front), so the back of the scan is still inside `inner` until `inner`
        // runs out; only then does the buffer's own back become the last item.
        // Read-ahead stays a forward-only optimization: a reverse scan walks
        // `inner` directly and reads its values one at a time, as before.
        if !self.inner_back_done {
            if let Some(item) = self.inner.next_back() {
                return Some(self.guard(item));
            }
            self.inner_back_done = true;
        }

        let item = self.buf.pop_back()?;
        Some(self.guard(item))
    }
}

/// Wraps a [`SeekableTreeIter`](crate::range::SeekableTreeIter) over the index
/// tree so a KV-separated tree can expose it as a [`SeekableGuardIter`](crate::iter_guard::SeekableGuardIter).
struct BlobSeekable {
    inner: crate::range::SeekableTreeIter,
    tree: crate::BlobTree,
    version: Version,
    /// Whether its values are cached, as the tree said when it was opened.
    fill: crate::config::BlobCacheFill,
}

impl Iterator for BlobSeekable {
    type Item = IterGuardImpl;

    fn next(&mut self) -> Option<Self::Item> {
        self.inner
            .next()
            .map(|item| blob_guard(&self.tree, &self.version, self.fill, item))
    }
}

impl DoubleEndedIterator for BlobSeekable {
    fn next_back(&mut self) -> Option<Self::Item> {
        self.inner
            .next_back()
            .map(|item| blob_guard(&self.tree, &self.version, self.fill, item))
    }
}

impl crate::iter_guard::SeekableGuardIter for BlobSeekable {
    fn seek_to(&mut self, key: &[u8]) {
        self.inner.seek_to(key);
    }

    fn seek_to_for_prev(&mut self, key: &[u8]) {
        self.inner.seek_to_for_prev(key);
    }

    fn peek_key(&mut self) -> Option<crate::Result<crate::UserKey>> {
        // The key lives in the index tree (blob separation only moves the value),
        // so the inner seekable's peek already yields the right user key.
        self.inner.peek_key()
    }
}

impl AbstractTree for BlobTree {
    #[cfg(feature = "std")]
    fn create_checkpoint(
        &self,
        target_path: &crate::path::Path,
    ) -> crate::Result<crate::CheckpointInfo> {
        crate::checkpoint::run_checkpoint(
            self,
            &crate::checkpoint::CheckpointParams {
                target_root: target_path,
                target_fs: &self.index.config.fs,
                src_root: &self.index.config.path,
                src_fs: &self.index.config.fs,
                deletion_pause: &self.index.deletion_pause,
                visible_seqno: &self.index.config.visible_seqno,
                include_blobs: true,
                runtime_config: self.index.0.runtime_config.load_full(),
                encryption: self.index.0.config.encryption.clone(),
                source: &self.index.0,
            },
        )
    }

    #[cfg(feature = "std")]
    fn pin(&self) -> crate::PinnedVersion {
        crate::PinnedVersion::take(&self.index)
    }

    #[cfg(feature = "std")]
    fn acknowledge_repair_report(&self, id: u128) -> crate::Result<bool> {
        self.index.acknowledge_repair_report(id)
    }

    fn print_trace(&self, key: &[u8]) -> crate::Result<()> {
        self.index.print_trace(key)
    }

    fn table_file_cache_size(&self) -> usize {
        self.index.table_file_cache_size()
    }

    fn get_version_history_lock(&self) -> VersionsWriteGuard<'_> {
        self.index.get_version_history_lock()
    }

    fn persist_flushes(&self) -> crate::Result<()> {
        self.index.persist_flushes()
    }

    fn next_table_id(&self) -> TableId {
        self.index.next_table_id()
    }

    fn id(&self) -> crate::TreeId {
        self.index.id()
    }

    fn get_internal_entry(&self, key: &[u8], seqno: SeqNo) -> crate::Result<Option<InternalValue>> {
        self.index.get_internal_entry(key, seqno)
    }

    fn current_version(&self) -> Version {
        self.index.current_version()
    }

    #[cfg(feature = "std")]
    fn refresh_table_checksum(
        &self,
        table_id: crate::TableId,
        checksum: crate::checksum::Checksum,
        expected_restriction: Option<&crate::UserKey>,
    ) -> crate::Result<crate::abstract_tree::ChecksumRefreshOutcome> {
        // Tables live in the index tree's version; blob files carry no
        // manifest digest.
        self.index
            .refresh_table_checksum(table_id, checksum, expected_restriction)
    }

    fn sync_mode(&self) -> crate::fs::SyncMode {
        // SSTs live in the index tree; its durability mode governs them.
        self.index.sync_mode()
    }

    fn wait_for_table_file_refills(&self) {
        self.index.wait_for_table_file_refills();
    }

    fn prefix_extractor(&self) -> Option<alloc::sync::Arc<dyn crate::prefix::PrefixExtractor>> {
        // The prefix filter is built over the index tree's SST keys.
        self.index.prefix_extractor()
    }

    fn storage_stats(&self) -> crate::Result<crate::StorageStats> {
        // Forward the index tree's compaction state (the default impl would
        // always report idle), and mark value bytes as NOT user values: large
        // values are KV-separated into blob files, so the SST records only
        // indirection pointers. One version snapshot is reused for both the
        // footprint and the blob-headroom sum below: a second `current_version()`
        // could race a concurrent flush / compaction and mix two snapshots.
        let version = self.current_version();
        let mut stats = crate::storage_stats::compute_storage_stats(
            &version,
            self.index.is_compacting(),
            false,
        )?;
        // Capacity is disk-aware and driven by the index tree's runtime config;
        // its footprint basis (`compute_used_bytes`) already stats blob files, so
        // `used_bytes` here is the blob-inclusive figure capacity is measured
        // against.
        let (capacity, available, compaction_possible) =
            self.index.admission_capacity(stats.used_bytes);
        stats.capacity_bytes = capacity;
        stats.available_bytes = available;
        stats.compaction_possible = compaction_possible;
        // A full compaction of a blob tree also relocates STALE blob files, which
        // the merge gate budgets via `pick_blob_files_to_rewrite` (linked, stale,
        // non-dead). Estimate the same subset across every table — NOT the whole
        // live blob footprint, which would overstate the need (large non-stale
        // blobs are not rewritten) and report tight while the gate would admit
        // the merge. Fold it into the gauge figure.
        let blob_need = self.full_compaction_blob_need(&version)?;
        stats.full_compaction_bytes += blob_need;
        // Surface full-vs-tight compaction availability through the SAME two-layer
        // check the compaction space gate enforces (logical quota + physical free
        // per destination volume), so the status matches what the gate admits.
        if self.index.storage_admission_enabled()
            && capacity.is_some()
            && stats.status == crate::StorageStatus::Healthy
        {
            // SST output lands in the LAST configured level's volume
            // (`level_count - 1`), stale blob relocation in the primary blobs
            // volume; the demand is bounded by the largest level's size. The
            // per-volume gate (not `available >= full_compaction_bytes` against
            // the min-volume free) keeps the status from reporting tight when the
            // SST and blob outputs each fit their own volume — see the gate's
            // two-layer model.
            let sst_need = crate::storage_stats::full_compaction_demand_bytes(&version)?;
            // `saturating_sub`: `level_count >= 1` always (the clamp only guards a
            // degenerate zero-level config) → the last level index.
            let sst_dest_level = self.index.config.level_count.saturating_sub(1);
            let quota_headroom = self.index.quota_headroom(stats.used_bytes);
            let full_fits = crate::compaction::worker::merge_would_fit(
                &self.index.compaction_capacity,
                self.index.id,
                &self.index.config,
                quota_headroom,
                (sst_need, blob_need),
                sst_dest_level,
            );
            stats.status = if full_fits {
                crate::StorageStatus::FullCompactionAvailable
            } else {
                crate::StorageStatus::TightCompactionAvailable
            };
        }
        // Admission is driven by the index tree's runtime config / footprint;
        // a closed gate is the operator-actionable state (see the standard
        // tree's override for the precedence rationale).
        if self.index.is_read_only() {
            stats.status = crate::StorageStatus::ReadOnlyOutOfSpace;
        }
        Ok(stats)
    }

    fn write_admission(&self) -> crate::Result<()> {
        // Admission state lives on the index tree (which holds the runtime
        // config). The forward is blob-aware: the index tree's version IS this
        // blob tree's version (current_version() delegates here), and the gate's
        // footprint basis (`storage_stats::compute_used_bytes`) stats live
        // tables AND blob files, so blob bytes count toward the budget.
        self.index.write_admission()
    }

    fn write_backpressure(
        &self,
        strategy: &dyn crate::compaction::CompactionStrategy,
    ) -> crate::Backpressure {
        // Same delegation as write_admission: the index tree holds the runtime
        // config and its version is this blob tree's version, so the L0 run
        // count and pending-compaction-bytes signals are computed on the right
        // state.
        self.index.write_backpressure(strategy)
    }

    #[cfg(feature = "metrics")]
    fn cache_stats(&self) -> crate::CacheStats {
        self.index.cache_stats()
    }

    #[cfg(feature = "metrics")]
    fn metrics(&self) -> &Arc<crate::Metrics> {
        self.index.metrics()
    }

    fn prefix<K: AsRef<[u8]>>(
        &self,
        prefix: K,
        seqno: SeqNo,
        index: Option<(Arc<Memtable>, SeqNo)>,
    ) -> Box<dyn DoubleEndedIterator<Item = IterGuardImpl> + Send + 'static> {
        use crate::prefix::compute_prefix_hash;
        use crate::range::prefix_to_range;

        let prefix_bytes = prefix.as_ref();

        let prefix_hash =
            compute_prefix_hash(self.index.config.prefix_extractor.as_ref(), prefix_bytes);

        // Resolved once, and the scan keeps the version alive in the guard it
        // returns: an indirection this iterator yields is resolved against the
        // blob files of the same version that held the pointer, so a blob GC
        // running concurrently cannot reclaim the file the pointer names.
        let super_version = match self.index.get_version_for_snapshot(seqno) {
            Ok(super_version) => super_version,
            Err(e) => return Box::new(core::iter::once(crate::tree::error_guard(e))),
        };
        let tree = self.clone();
        // One source for the scan: its read-ahead and its merges cache as the
        // tree said when it began.
        let source = self.blob_source();
        let state = self.scan_prefetch_state(source.fill);

        let range = prefix_to_range(prefix_bytes);

        Box::new(PrefetchScan {
            inner: crate::Tree::create_internal_range_with_prefix_hash(
                super_version.clone(),
                &range,
                seqno,
                index,
                self.index.config.merge_operator.clone(),
                self.index.config.comparator.clone(),
                prefix_hash,
                Some(source),
            ),
            buf: alloc::collections::VecDeque::new(),
            inner_back_done: false,
            armed: false,
            state,
            version: super_version.version,
            tree,
            local: crate::vlog::Prefetched::default(),
        })
    }

    fn range<K: AsRef<[u8]>, R: RangeBounds<K>>(
        &self,
        range: R,
        seqno: SeqNo,
        index: Option<(Arc<Memtable>, SeqNo)>,
    ) -> Box<dyn DoubleEndedIterator<Item = IterGuardImpl> + Send + 'static> {
        // One version for the traversal and for resolving the pointers it
        // yields, as in `prefix` above; the range bounds only narrow which of
        // that version's tables are opened.
        let super_version = match self.index.get_version_for_snapshot(seqno) {
            Ok(super_version) => super_version,
            Err(e) => return Box::new(core::iter::once(crate::tree::error_guard(e))),
        };
        let tree = self.clone();
        // One source for the scan, as in `prefix` above.
        let source = self.blob_source();
        let state = self.scan_prefetch_state(source.fill);

        Box::new(PrefetchScan {
            inner: crate::Tree::create_internal_range_with_prefix_hash(
                super_version.clone(),
                &range,
                seqno,
                index,
                self.index.config.merge_operator.clone(),
                self.index.config.comparator.clone(),
                None,
                Some(source),
            ),
            buf: alloc::collections::VecDeque::new(),
            inner_back_done: false,
            armed: false,
            state,
            version: super_version.version,
            tree,
            local: crate::vlog::Prefetched::default(),
        })
    }

    fn range_seekable<K: AsRef<[u8]>, R: RangeBounds<K>>(
        &self,
        range: R,
        seqno: SeqNo,
        index: Option<(Arc<Memtable>, SeqNo)>,
    ) -> Box<dyn crate::iter_guard::SeekableGuardIter + 'static> {
        let (lo, hi) = crate::tree::range_to_user_bounds(&range);
        // One source for the iterator, its merges and its values alike.
        let source = self.blob_source();
        let fill = source.fill;
        let inner =
            match self
                .index
                .create_seekable_range_bounds_with(lo, hi, seqno, index, Some(source))
            {
                Ok(inner) => inner,
                Err(e) => return Box::new(crate::iter_guard::FailedSeekable::new(e)),
            };
        let version = inner.version();
        Box::new(BlobSeekable {
            inner,
            tree: self.clone(),
            version,
            fill,
        })
    }

    fn batch_range_scan<K: AsRef<[u8]>, R: RangeBounds<K> + 'static, I: IntoIterator<Item = R>>(
        &self,
        intervals: I,
        seqno: SeqNo,
        index: Option<(Arc<Memtable>, SeqNo)>,
    ) -> Box<dyn Iterator<Item = IterGuardImpl> + Send + 'static>
    where
        I::IntoIter: Send + 'static,
    {
        // One source for the scan, as in `prefix` above.
        let source = self.blob_source();
        let state = self.scan_prefetch_state(source.fill);
        let inner = match self.index.create_seekable_range_bounds_with(
            core::ops::Bound::Unbounded,
            core::ops::Bound::Unbounded,
            seqno,
            index,
            Some(source),
        ) {
            Ok(inner) => inner,
            Err(e) => return Box::new(core::iter::once(crate::tree::error_guard(e))),
        };
        let version = inner.version();
        let tree = self.clone();
        let intervals = intervals
            .into_iter()
            .map(|r| crate::tree::range_to_user_bounds(&r));
        Box::new(PrefetchScan {
            inner: crate::range::BatchRangeScan::new(inner, intervals),
            buf: alloc::collections::VecDeque::new(),
            inner_back_done: false,
            armed: false,
            state,
            version,
            tree,
            local: crate::vlog::Prefetched::default(),
        })
    }

    fn tombstone_count(&self) -> u64 {
        self.index.tombstone_count()
    }

    fn weak_tombstone_count(&self) -> u64 {
        self.index.weak_tombstone_count()
    }

    fn weak_tombstone_reclaimable_count(&self) -> u64 {
        self.index.weak_tombstone_reclaimable_count()
    }

    fn clear(&self) -> crate::Result<()> {
        let config = self.tree_config();
        let mut versions = self.get_version_history_lock();

        // Pre-clear snapshot: its tables AND blob files all become garbage once
        // the new empty version is installed.
        let prior = versions.latest_version();

        // The blob dictionary counts here as much as the table one: the next
        // blob file is compressed against it.
        #[cfg(zstd_any)]
        let still_written = self.index.write_referenced_dict_ids();
        #[cfg(not(zstd_any))]
        let still_written: Vec<crate::file::DictId> = Vec::new();

        versions.upgrade_version(
            &config.path,
            |v| {
                let mut copy = v.clone();
                copy.active_memtable = Arc::new(Memtable::new(
                    self.index.memtable_id_counter.next(),
                    config.comparator.clone(),
                ));
                copy.sealed_memtables = Arc::default();
                copy.version = v.version.cleared(&still_written);
                Ok(copy)
            },
            &config.seqno,
            &config.visible_seqno,
            &*config.fs,
            self.index.0.runtime_config.load_full(),
            self.index.0.config.encryption.clone(),
            // Every table and blob file goes: no snapshot up to this install
            // is servable.
            crate::version::RetentionEffect::DropsData,
        )?;

        // Same MVCC-safe reclaim as the standard tree, plus the blob files:
        // mark every obsolete table / blob file deleted and let Inner::Drop
        // reclaim each once its last reference is released (a live reader's
        // snapshot clone defers deletion).
        drop(versions);

        for table in prior.version.iter_tables() {
            table.mark_as_deleted();
        }
        for blob_file in prior.version.blob_files.iter() {
            blob_file.mark_as_deleted();
        }

        Ok(())
    }

    fn major_compact(
        &self,
        target_size: u64,
        gc_watermark: SeqNo,
    ) -> crate::Result<crate::compaction::CompactionResult> {
        self.index.major_compact(target_size, gc_watermark)
    }

    fn clear_active_memtable(&self) {
        self.index.clear_active_memtable();
    }

    fn l0_run_count(&self) -> usize {
        self.index.l0_run_count()
    }

    fn blob_file_count(&self) -> usize {
        self.current_version().blob_file_count()
    }

    // NOTE: We skip reading from the value log
    // because the vHandles already store the value size
    fn size_of<K: AsRef<[u8]>>(&self, key: K, seqno: SeqNo) -> crate::Result<Option<u32>> {
        let key = key.as_ref();
        let super_version = self.index.snapshot_for_read(seqno)?;
        super_version.version.read_mix().note_points(1);
        let Some(item) = crate::Tree::get_internal_entry_from_version(
            &super_version,
            key,
            seqno,
            self.index.config.comparator.as_ref(),
        )?
        else {
            return Ok(None);
        };

        // An operand's size is not the key's: the value is the operands
        // merged onto the base, which only the merge itself knows. Merged from
        // the same read, which counted this lookup already.
        if item.key.value_type.is_merge_operand()
            && let Some(merge_operator) = &self.index.config.merge_operator
        {
            #[expect(clippy::cast_possible_truncation, reason = "values are u32 length max")]
            return Ok(crate::Tree::resolve_point_merge(
                &super_version,
                key,
                seqno,
                merge_operator,
                Some(&self.blob_source()),
            )?
            .map(|value| value.len() as u32));
        }

        Ok(Some(if item.key.value_type.is_indirection() {
            let mut cursor = crate::io::Cursor::new(item.value);
            let vptr = BlobIndirection::decode_from(&mut cursor)?;
            vptr.size
        } else if item.key.value_type.is_cell_row() {
            field_row::logical_len(&item.value)?
        } else {
            #[expect(clippy::cast_possible_truncation, reason = "values are u32 length max")]
            {
                item.value.len() as u32
            }
        }))
    }

    fn stale_blob_bytes(&self) -> u64 {
        self.current_version().gc_stats().stale_bytes()
    }

    fn filter_size(&self) -> u64 {
        self.index.filter_size()
    }

    fn pinned_filter_size(&self) -> usize {
        self.index.pinned_filter_size()
    }

    fn filter_memory(&self) -> crate::FilterMemory {
        self.index.filter_memory()
    }

    fn pinned_block_index_size(&self) -> usize {
        self.index.pinned_block_index_size()
    }

    fn sealed_memtable_count(&self) -> usize {
        self.index.sealed_memtable_count()
    }

    fn get_flush_lock(&self) -> FlushGuard<'_> {
        self.index.get_flush_lock()
    }

    #[expect(clippy::too_many_lines, reason = "flush logic is inherently complex")]
    fn flush_to_tables_with_rt(
        &self,
        stream: impl Iterator<Item = crate::Result<InternalValue>>,
        range_tombstones: Vec<crate::range_tombstone::RangeTombstone>,
        keys: u64,
        hashes: u64,
        overwritten: Option<&overwritten::OverwrittenKeys>,
    ) -> crate::Result<
        Option<(
            Vec<Table>,
            Option<Vec<BlobFile>>,
            crate::runtime_config::WritePin,
        )>,
    > {
        use crate::{coding::Encode, file::BLOBS_FOLDER, table::multi_writer::MultiWriter};

        let start = crate::time::Instant::now();

        let (table_folder, level_fs) = self.index.config.tables_folder_for_level(0);

        let data_block_size = self.index.config.data_block_size_policy.get(0);

        let data_block_restart_interval =
            self.index.config.data_block_restart_interval_policy.get(0);
        let index_block_restart_interval =
            self.index.config.index_block_restart_interval_policy.get(0);

        // Live runtime snapshot, one for the whole flush: the compression
        // policies (tables and blob files alike) and the per-flush SST options
        // below all come from it, so a concurrent `update_runtime_config`
        // cannot give one flush two configurations.
        let rc = self.index.0.runtime_config.load_full();

        let data_block_compression = rc.data_block_compression_policy.get(0);
        let index_block_compression = rc.index_block_compression_policy.get(0);

        let data_block_hash_ratio = self.index.config.data_block_hash_ratio_policy.get(0);

        let index_partitioning = self.index.config.index_block_partitioning_policy.get(0);
        let filter_partitioning = self.index.config.filter_block_partitioning_policy.get(0);

        log::debug!(
            "Flushing memtable(s) and performing key-value separation, data_block_restart_interval={data_block_restart_interval}, index_block_restart_interval={index_block_restart_interval}, data_block_size={data_block_size}, data_block_compression={data_block_compression:?}, index_block_compression={index_block_compression:?}"
        );
        log::debug!("=> to table(s) in {}", table_folder.display());
        log::debug!("=> to blob file(s) at {}", self.blobs_folder.display());

        let table_writer = MultiWriter::from_pool(
            table_folder.clone(),
            Arc::clone(&self.index.table_files),
            None,
            self.index.config.table_target_size,
            0,
            level_fs.clone(),
            crate::fs::WriteOrigin::FlushTable,
        )?;
        // The flush's L0 recency, taken as the standard tree's flush takes
        // it: a pooled file's id says nothing about when it was written, a
        // file made now carries an id taken now.
        let recency = table_writer
            .first_made_now()
            .unwrap_or_else(|| self.index.table_id_counter.next());
        let mut table_writer = table_writer
            .set_comparator(self.index.config.comparator.clone())
            .use_recency(Some(recency))
            .use_data_block_restart_interval(data_block_restart_interval)
            .use_index_block_restart_interval(index_block_restart_interval)
            .use_data_block_compression(data_block_compression)
            .use_index_block_compression(index_block_compression)
            .use_data_block_size(data_block_size)
            .use_index_partition_size(self.index.config.index_block_partition_size_policy.get(0))
            .use_filter_partition_size(self.index.config.filter_block_partition_size_policy.get(0))
            .use_row_group_size(self.index.config.columnar_row_group_size_policy.get(0))
            .use_columnar_page_size(self.index.config.columnar_page_size_policy.get(0))
            .use_column_encoding(self.index.config.column_encoding_policy.get(0))
            .use_data_block_hash_ratio(data_block_hash_ratio);

        let bloom_policy = {
            use crate::config::FilterPolicyEntry::{Bloom, None};
            use crate::table::filter::BloomConstructionPolicy;

            match self.index.config.filter_policy.get(0) {
                Bloom(policy) => policy,
                None => BloomConstructionPolicy::BitsPerKey(0.0),
            }
        };
        table_writer = table_writer.use_bloom_policy(bloom_policy);
        let filter_sizing = self.index.new_data_filter_sizing(
            0,
            bloom_policy,
            crate::filter_budget::FilterCount { keys, hashes },
            rc.ecc_scheme,
        );
        table_writer = table_writer.use_filter_sizing(filter_sizing.clone());

        if index_partitioning {
            // Size-adaptive index: single-level for small SSTs, spill to
            // partitioned only past the threshold (see flush path).
            table_writer = table_writer.use_adaptive_index(rc.index_partition_spill_threshold);
        }
        if filter_partitioning {
            table_writer = table_writer.use_partitioned_filter();
        }

        table_writer =
            table_writer.use_prefix_extractor(self.index.config.prefix_extractor.clone());
        table_writer = table_writer.use_key_domains(self.index.config.key_domains.clone());
        table_writer = table_writer.use_encryption(self.index.config.encryption.clone());
        table_writer = table_writer.use_clock(self.index.config.clock.clone());
        // Runtime-driven SST options, mirroring the standard tree's flush so a
        // blob tree's index honors columnar, zone maps, ECC, seqno-in-index,
        // per-KV checksums, locator policy, sync mode, and CoW disable.
        table_writer = table_writer.use_page_ecc(self.index.config.page_ecc, rc.ecc_scheme);
        table_writer = table_writer
            .use_sync_mode(self.index.config.sync_mode)
            .use_durability(self.index.config.flush_durability)
            .use_writeback_bytes(self.index.config.writeback_bytes);
        table_writer = table_writer.use_seqno_in_index(rc.seqno_in_index);
        table_writer = table_writer.use_zone_map(rc.zone_map);
        table_writer = table_writer
            .use_columnar(rc.columnar)
            .use_cell_rows(rc.columnar);
        table_writer = table_writer.use_disable_cow_on_sst(rc.disable_cow_on_sst_files);
        table_writer = table_writer.use_kv_checksums(rc.kv_checksums, rc.kv_checksum_algo);
        table_writer = table_writer.use_locator(self.index.config.locator_policy.get(0));

        // Resolved from `rc`, which the flush holds until the install, as in
        // the standard flush; one snapshot covers the tables and the blob files.
        #[cfg(zstd_any)]
        let dicts = self.index.config.current_zstd_dictionaries();
        #[cfg(zstd_any)]
        {
            table_writer = table_writer
                .use_zstd_dictionary(dicts.for_compression(data_block_compression)?)
                .use_zstd_two_pass_seed(rc.zstd_two_pass_seed);
        }
        let write_pin = crate::runtime_config::WritePin::new(&rc).with_filter_sizing(filter_sizing);

        // Parallel block compression for the flush writer, mirroring the
        // standard tree's flush: engaged only when the per-block transform does
        // real CPU work (codec / encryption / page ECC). See the standard
        // flush for why this is safe wherever the host runs the flush (the
        // pipeline's help-first drain, see `parallel_compressor`).
        #[cfg(feature = "std")]
        {
            let transform_does_work = data_block_compression != crate::CompressionType::None
                || self.index.config.encryption.is_some()
                || self.index.config.page_ecc;
            if transform_does_work {
                table_writer =
                    table_writer.use_parallel_compression(self.index.config.parallel_compression());
            }
        }

        #[expect(
            clippy::expect_used,
            reason = "cannot create blob tree without defining kv separation options"
        )]
        let kv_opts = self
            .index
            .config
            .kv_separation_opts
            .as_ref()
            .expect("kv separation options should exist");

        let mut blob_writer = {
            let w = BlobFileWriter::new_for(
                self.index.0.blob_file_id_counter.clone(),
                self.index.config.path.join(BLOBS_FOLDER),
                self.id(),
                self.index.config.descriptor_table.clone(),
                self.index.config.fs.clone(),
                crate::fs::WriteOrigin::FlushBlob,
            )?
            .use_target_size(kv_opts.file_target_size)
            .use_compression(rc.blob_compression)
            .use_sync_mode(self.index.config.sync_mode)
            .use_durability(self.index.config.flush_durability)
            .use_writeback_bytes(self.index.config.writeback_bytes)
            .use_lifetime_groups(kv_opts.lifetime_groups)
            .use_clock(self.index.config.clock.clone())
            .use_key_domains(
                self.index.config.key_domains.clone(),
                self.index.config.comparator.clone(),
            );
            #[cfg(zstd_any)]
            let w = w
                .use_zstd_dictionary(dicts.for_compression(rc.blob_compression)?)
                .use_zstd_two_pass_seed(rc.zstd_two_pass_seed)
                .use_zstd_dictionaries(dicts);
            w
        };

        let separation_threshold = kv_opts.separation_threshold;
        let lifetime = kv_opts.lifetime_classifier();
        let comparator = &self.index.config.comparator;

        // Set range tombstones BEFORE writing KV items so that if MultiWriter
        // rotates to a new table during the write loop, earlier tables already
        // carry the RT metadata.
        table_writer.set_range_tombstones(range_tombstones);

        for item in stream {
            let item = item?;

            // Asked for every key, separated or not, so the queue forgets each
            // key once the flush has passed it.
            let overwritten = overwritten
                .is_some_and(|keys| keys.take_before_and_check(&item.key.user_key, comparator));

            if item.is_tombstone() {
                // NOTE: Still need to add tombstone to index tree
                // But no blob to blob writer
                table_writer.write(InternalValue::new(item.key, UserValue::empty()))?;
                continue;
            }

            let value = item.value;

            if item.key.value_type.is_cell_row() {
                let class = lifetime.class_of(&item.key.user_key, overwritten);
                // Each cell is weighed on its own: the large ones go to the
                // blob file and the row keeps owning references to them.
                let row = field_row::separate_row(
                    &value,
                    |column| kv_opts.cell_threshold(column),
                    |bytes| {
                        blob_writer.select_lifetime_class(class)?;
                        blob_writer.write(&item.key.user_key, item.key.seqno, bytes)
                    },
                )?
                .map_or(value, UserValue::from);
                table_writer.write(InternalValue::new(item.key, row.clone()))?;
                table_writer.register_cell_row(&row)?;
                continue;
            }

            #[expect(clippy::cast_possible_truncation, reason = "values are u32 length max")]
            let value_size = value.len() as u32;

            // Only a value is separated: an indirection reads as a put, so an
            // operand stored as one would stop merging onto its base.
            if item.key.value_type == crate::ValueType::Value && value_size >= separation_threshold
            {
                blob_writer
                    .select_lifetime_class(lifetime.class_of(&item.key.user_key, overwritten))?;
                let vhandle = blob_writer.write(&item.key.user_key, item.key.seqno, &value)?;

                let indirection = BlobIndirection {
                    vhandle,
                    size: value_size,
                };

                table_writer.write({
                    let mut vptr =
                        InternalValue::new(item.key.clone(), indirection.encode_into_vec());
                    vptr.key.value_type = crate::ValueType::Indirection;
                    vptr
                })?;

                table_writer.register_blob(indirection);
            } else {
                table_writer.write(InternalValue::new(item.key, value))?;
            }
        }

        let blob_files = blob_writer.finish()?;

        let written = table_writer.finish()?;
        let entries_durable = written.entries_durable;

        log::debug!("Flushed memtable(s) in {:?}", start.elapsed());

        let pin_filter = self.index.config.filter_block_pinning_policy.get(0);
        let pin_index = self.index.config.index_block_pinning_policy.get(0);

        // Load tables
        let tables = written
            .tables
            .into_iter()
            .map(|(table_id, checksum)| -> crate::Result<Table> {
                let mut params = crate::table::RecoverParams::new(
                    table_folder.join(table_id.to_string()),
                    checksum,
                    table_id,
                    level_fs.clone(),
                    self.index.config.comparator.clone(),
                    self.index.config.cache.clone(),
                );
                params.tree_id = self.index.id;
                params
                    .descriptor_table
                    .clone_from(&self.index.config.descriptor_table);
                params.pin_filter = pin_filter;
                params.pin_index = pin_index;
                params.encryption.clone_from(&self.index.config.encryption);
                params
                    .prefix_extractor
                    .clone_from(&self.index.config.prefix_extractor);
                #[cfg(zstd_any)]
                {
                    params.zstd_dictionaries = self.index.config.current_zstd_dictionaries();
                }
                #[cfg(feature = "metrics")]
                {
                    params.metrics = self.index.metrics.clone();
                }
                let table = Table::recover(params)?;
                if entries_durable {
                    table.mark_entry_durable();
                }
                Ok(table)
            })
            .collect::<crate::Result<Vec<_>>>()?;

        // Return Some even when tables is empty (RT-only flush): the caller
        // (AbstractTree::flush) handles empty tables by re-inserting RTs into
        // the active memtable and still needs to delete sealed memtables.
        Ok(Some((tables, Some(blob_files), write_pin)))
    }

    fn register_tables(
        &self,
        tables: &[Table],
        blob_files: Option<&[BlobFile]>,
        frag_map: Option<FragmentationMap>,
        sealed_memtables_to_delete: &[MemtableId],
        gc_watermark: SeqNo,
        collected_below_watermark: bool,
    ) -> crate::Result<()> {
        self.index.register_tables(
            tables,
            blob_files,
            frag_map,
            sealed_memtables_to_delete,
            gc_watermark,
            collected_below_watermark,
        )
    }

    fn compact(
        &self,
        strategy: Arc<dyn crate::compaction::CompactionStrategy>,
        gc_watermark: SeqNo,
    ) -> crate::Result<crate::compaction::CompactionResult> {
        self.index.compact(strategy, gc_watermark)
    }

    fn get_next_table_id(&self) -> TableId {
        self.index.get_next_table_id()
    }

    fn tree_config(&self) -> &Config {
        &self.index.config
    }

    fn get_highest_seqno(&self) -> Option<SeqNo> {
        self.index.get_highest_seqno()
    }

    fn active_memtable(&self) -> Arc<Memtable> {
        self.index.active_memtable()
    }

    fn rotate_memtable(&self) -> Option<Arc<Memtable>> {
        self.index.rotate_memtable()
    }

    fn table_count(&self) -> usize {
        self.index.table_count()
    }

    fn level_table_count(&self, idx: usize) -> Option<usize> {
        self.index.level_table_count(idx)
    }

    fn approximate_len(&self) -> usize {
        self.index.approximate_len()
    }

    // NOTE: Override the default implementation to not fetch
    // data from the value log, so we get much faster key reads
    fn is_empty(&self, seqno: SeqNo, index: Option<(Arc<Memtable>, SeqNo)>) -> crate::Result<bool> {
        self.index.is_empty(seqno, index)
    }

    // NOTE: Override the default implementation to not fetch
    // data from the value log, so we get much faster key reads
    fn contains_key<K: AsRef<[u8]>>(&self, key: K, seqno: SeqNo) -> crate::Result<bool> {
        self.index.contains_key(key, seqno)
    }

    // NOTE: Override the default implementation to delegate directly
    // to the index tree, avoiding extra iterator/guard overhead for
    // prefix checks
    fn contains_prefix<K: AsRef<[u8]>>(
        &self,
        prefix: K,
        seqno: SeqNo,
        index: Option<(Arc<Memtable>, SeqNo)>,
    ) -> crate::Result<bool> {
        self.index.contains_prefix(prefix, seqno, index)
    }

    // NOTE: Override the default implementation to not fetch
    // data from the value log, so we get much faster scans
    fn len(&self, seqno: SeqNo, index: Option<(Arc<Memtable>, SeqNo)>) -> crate::Result<usize> {
        self.index.len(seqno, index)
    }

    fn disk_space(&self) -> u64 {
        let version = self.current_version();
        self.index.disk_space() + version.blob_files.on_disk_size()
    }

    fn approximate_range_stats<K: AsRef<[u8]>, R: core::ops::RangeBounds<K>>(
        &self,
        range: R,
        seqno: SeqNo,
    ) -> crate::Result<crate::ApproximateRangeStats> {
        // The index tree's SSTs record their referenced blob bytes per-SST (at
        // both flush and compaction), so its estimate already apportions the
        // blob bytes by the in-range fraction. Delegating keeps the
        // KV-separated estimate unified with the standard path.
        //
        // The figure tracks the value's CURRENT physical location: an unflushed
        // value still lives inline in the index memtable, so the memtable share
        // counts its full size there; once flushed it is separated into a blob
        // file and counted via the SST's referenced-blob bytes. Both reflect the
        // real footprint at the requested snapshot.
        self.index.approximate_range_stats(range, seqno)
    }

    fn approximate_range_cardinality<K: AsRef<[u8]>, R: core::ops::RangeBounds<K>>(
        &self,
        range: R,
        seqno: SeqNo,
    ) -> crate::Result<crate::RangeCardinality> {
        // Cardinality counts rows (index entries) and their key ranges, both of
        // which live entirely in the index tree; blob files hold only the
        // separated values, so the index tree's zone map and counts are authoritative.
        self.index.approximate_range_cardinality(range, seqno)
    }

    fn get_highest_memtable_seqno(&self) -> Option<SeqNo> {
        self.index.get_highest_memtable_seqno()
    }

    fn get_highest_persisted_seqno(&self) -> Option<SeqNo> {
        self.index.get_highest_persisted_seqno()
    }

    fn oldest_retained_seqno(&self) -> SeqNo {
        // Snapshots are resolved through the index tree's history.
        self.index.oldest_retained_seqno()
    }

    fn retention_floor(&self) -> SeqNo {
        self.index.retention_floor()
    }

    fn lower_retention_floor(&self, floor: SeqNo) -> crate::Result<()> {
        // The blob files share the index tree's version, and its floor.
        self.index.lower_retention_floor(floor)
    }

    fn compaction_rate_limiter(&self) -> alloc::sync::Arc<crate::rate_limiter::RateLimiter> {
        // Compactions run through the index tree, so its limiter is the one.
        self.index.compaction_rate_limiter()
    }

    fn write_rate_limiter(&self) -> alloc::sync::Arc<crate::rate_limiter::RateLimiter> {
        // Writes land in the index tree's memtables, so its limiter is the one.
        self.index.write_rate_limiter()
    }

    fn apply_batch(&self, batch: crate::WriteBatch, seqno: SeqNo) -> crate::Result<(u64, u64)> {
        self.index.apply_batch(batch, seqno)
    }

    fn insert<K: Into<UserKey>, V: Into<UserValue>>(
        &self,
        key: K,
        value: V,
        seqno: SeqNo,
    ) -> (u64, u64) {
        self.index.insert(key, value.into(), seqno)
    }

    fn get<K: AsRef<[u8]>>(&self, key: K, seqno: SeqNo) -> crate::Result<Option<crate::UserValue>> {
        // The index entry and the blob it points at are read from the same
        // version: `resolve_key` follows the indirection through this
        // `super_version`, never through the current one, so the pointer and the
        // blob file it names cannot come from different points in time.
        let super_version = self.index.snapshot_for_read(seqno)?;
        super_version.version.read_mix().note_points(1);
        self.resolve_key(&super_version, &self.blob_source(), key.as_ref(), seqno)
    }

    fn multi_get<K: AsRef<[u8]>>(
        &self,
        keys: impl IntoIterator<Item = K>,
        seqno: SeqNo,
    ) -> crate::Result<Vec<Option<crate::UserValue>>> {
        // The snapshot is validated BEFORE the empty-batch return: the
        // retention contract is per read, not per key, and the standard tree
        // refuses an empty batch below retention the same way.
        let super_version = self.index.snapshot_for_read(seqno)?;

        let keys: Vec<_> = keys.into_iter().collect();
        let n = keys.len();
        if n == 0 {
            return Ok(Vec::new());
        }

        // For small batches, use the simple per-key path
        if n <= 2 {
            super_version
                .version
                .read_mix()
                .note_points(u64::try_from(n).unwrap_or(u64::MAX));
            let source = self.blob_source();
            return keys
                .iter()
                .map(|key| self.resolve_key(&super_version, &source, key.as_ref(), seqno))
                .collect();
        }

        // The read shared with `Tree::multi_get`, which follows a blob
        // indirection into this version's value log. One source for the whole
        // read, so every key caches as the tree said when it began.
        let source = self.blob_source();
        crate::Tree::read_many(&crate::tree::read_job::ReadCtx {
            super_version,
            keys,
            seqno,
            comparator: Arc::clone(&self.index.config.comparator),
            merge_operator: self.index.config.merge_operator.clone(),
            merge_base: Some(source.clone()),
            values: crate::tree::read_job::Values::Blob(source),
            metadata_budget: self.index.config.multi_get_metadata_budget,
        })
    }

    fn merge<K: Into<UserKey>, V: Into<UserValue>>(
        &self,
        key: K,
        operand: V,
        seqno: SeqNo,
    ) -> (u64, u64) {
        self.index.merge(key, operand, seqno)
    }

    fn remove<K: Into<UserKey>>(&self, key: K, seqno: SeqNo) -> (u64, u64) {
        self.index.remove(key, seqno)
    }

    fn remove_weak<K: Into<UserKey>>(&self, key: K, seqno: SeqNo) -> (u64, u64) {
        self.index.remove_weak(key, seqno)
    }

    fn remove_range<K: Into<UserKey>>(&self, start: K, end: K, seqno: SeqNo) -> u64 {
        self.index.remove_range(start, end, seqno)
    }
}

#[cfg(test)]
mod tests;