coordinode-lsm-tree 7.4.2

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

mod filter;
mod index;
mod meta;
#[cfg(feature = "std")] // no-std: parallel compaction unavailable (no threads)
pub(crate) mod ordered_pipeline;
#[cfg(feature = "std")] // no-std: parallel compaction unavailable (no threads)
pub(crate) mod parallel_compressor;

pub(crate) use index::DEFAULT_SPILL_THRESHOLD;
#[cfg(feature = "std")]
pub use parallel_compressor::CompactionSpawner;
#[cfg(feature = "std")]
pub(crate) use parallel_compressor::ParallelCompression;
#[cfg(feature = "parallel")]
pub use parallel_compressor::RayonSpawner;

use super::{Block, BlockOffset, DataBlock, KeyedBlockHandle, filter::BloomConstructionPolicy};
use crate::{
    Checksum, CompressionType, InternalValue, TableId, UserKey, ValueType,
    checksum::{ChecksumType, ChecksummedWriter},
    coding::Encode,
    encryption::EncryptionProvider,
    fs::{FileHint, Fs, FsFile, FsOpenOptions, SyncMode},
    prefix::PrefixExtractor,
    range_tombstone::RangeTombstone,
    table::{
        BlockHandle,
        writer::{
            filter::{FilterWriter, FullFilterWriter},
            index::FullIndexWriter,
        },
    },
    vlog::BlobFileId,
};
use alloc::sync::Arc;
#[cfg(not(feature = "std"))]
use alloc::{boxed::Box, vec::Vec};
use index::BlockIndexWriter;

use crate::io::BufWriter;
use crate::path::PathBuf;

#[cfg(feature = "std")] // no-std: parallel compaction unavailable (no threads)
use {parallel_compressor::BlockCompressor, std::collections::VecDeque};

/// Order-independent index-handle data for a spilled block, captured before the
/// chunk is consumed so the parallel pipeline can write + register the block
/// later (after a worker has compressed it) without the original chunk in hand.
#[cfg(feature = "std")]
struct HandleMeta {
    last_key: UserKey,
    last_seqno: crate::SeqNo,
    seqno_bounds: Option<(u64, u64)>,
    item_count: usize,
    /// The block's first (min) user key, captured for the zone-map synthetic
    /// column. `Some` only when the zone-map policy is on (else no clone).
    zone_block_min: Option<UserKey>,
    /// The bound on the block's frame counted in flight, taken back out when
    /// it is written.
    frame_bound: u64,
    /// The heap the block holds in flight, its encoded payload and its frame,
    /// taken back out when it is written.
    heap_bound: u64,
}

/// Bytes of the padding section between the two meta copies, so a bad sector
/// cannot take both.
const META_SEPARATOR_LEN: usize = 4_096;

/// Every section a table can carry, which bounds its table of contents.
const TABLE_SECTIONS: [&str; 18] = [
    "data",
    "index",
    "tli",
    "filter",
    "filter_tli",
    "block_layout",
    "seqno_bounds",
    "zone_map",
    "delete_bitmap",
    "owned_blob_objects",
    "locator",
    "range_tombstones",
    "meta_mid",
    "linked_blob_files",
    "table_version",
    "meta_separator",
    "tli_tail",
    "meta",
];

/// Bytes every table ends with besides its two meta copies: the version
/// byte, the separator, and its table of contents and trailer, from above.
const FIXED_TAIL_LEN: u64 =
    (1 + META_SEPARATOR_LEN + crate::sfa::toc_and_trailer_len(&TABLE_SECTIONS)) as u64;

/// New keys between two refreshes of the estimates within a block: at least
/// this many, and at least 1/[`REFRESH_KEY_FRACTION`] of the keys so far, so
/// the per-key state they leave out stays within about 1.6% of the table's.
/// Refreshing on every key cost 3% of ingestion time.
const REFRESH_KEY_STEP: u64 = 256;
const REFRESH_KEY_FRACTION: u64 = 64;

/// Filter hashes a key may register, on average, before their count alone
/// refreshes the estimates sooner than the key count does.
const REFRESH_HASH_STEP_KEYS: u64 = 8;

/// Bytes a `linked_blob_files` record takes before its two keys: the blob
/// file's id, entry count, bytes and on-disk bytes.
pub(crate) const LINKED_BLOB_FILE_FIXED_LEN: usize = 4 * core::mem::size_of::<u64>();

/// Bytes one `linked_blob_files` record takes: its fixed fields and its first
/// and last key, each preceded by a two-byte length.
#[must_use]
fn linked_blob_file_record_len(first_key: &[u8], last_key: &[u8]) -> u64 {
    (LINKED_BLOB_FILE_FIXED_LEN + 2 + first_key.len() + 2 + last_key.len()) as u64
}

/// The blob files the table being written links, keyed by file, with the
/// bytes their `linked_blob_files` section takes kept up to date as entries
/// register, since a writer asks for that size before every key.
#[derive(Default)]
pub(crate) struct LinkedBlobFiles {
    files: crate::HashMap<BlobFileId, LinkedFile>,
    /// The records' bytes, without the section's four-byte count.
    records_len: u64,
}

impl LinkedBlobFiles {
    /// Records that the table's entry at `key` points at a value of `bytes`
    /// (`on_disk_bytes` stored) in `blob_file_id`. Keys arrive in order, so
    /// the first key seen for a file is its first and the latest its last.
    ///
    /// Only an `owned` reference adds to the counts: the counts are what
    /// dropping the table charges to the file, and an object is charged
    /// through its owner alone. A borrowed one still links the file, which is
    /// what keeps it from being removed or relocated under this table.
    pub(crate) fn register(
        &mut self,
        blob_file_id: BlobFileId,
        bytes: u64,
        on_disk_bytes: u64,
        key: &UserKey,
        owned: bool,
    ) {
        let (len, bytes, on_disk_bytes) = if owned {
            (1, bytes, on_disk_bytes)
        } else {
            (0, 0, 0)
        };
        match self.files.entry(blob_file_id) {
            hashbrown::hash_map::Entry::Occupied(mut entry) => {
                let link = entry.get_mut();
                link.bytes += bytes;
                link.on_disk_bytes += on_disk_bytes;
                link.len += len;
                // The record grows or shrinks by the change in its last key.
                self.records_len = self.records_len - link.last_key.len() as u64 + key.len() as u64;
                link.last_key.clone_from(key);
            }
            hashbrown::hash_map::Entry::Vacant(entry) => {
                self.records_len += linked_blob_file_record_len(key, key);
                entry.insert(LinkedFile {
                    blob_file_id,
                    bytes,
                    on_disk_bytes,
                    len,
                    first_key: key.clone(),
                    last_key: key.clone(),
                });
            }
        }
    }

    /// Bytes the `linked_blob_files` section takes: nothing without links,
    /// else the count and the records.
    #[must_use]
    pub(crate) fn section_len(&self) -> u64 {
        if self.files.is_empty() {
            0
        } else {
            4 + self.records_len
        }
    }

    /// Number of files linked.
    #[must_use]
    pub(crate) fn len(&self) -> usize {
        self.files.len()
    }

    /// Entries the map has room for without reallocating.
    #[cfg(test)]
    pub(crate) fn capacity(&self) -> usize {
        self.files.capacity()
    }

    /// Takes the links out, leaving none, so the map is freed rather than
    /// kept under the next table.
    pub(crate) fn take(&mut self) -> impl Iterator<Item = LinkedFile> {
        self.records_len = 0;
        core::mem::take(&mut self.files).into_values()
    }
}

/// Parses a `linked_blob_files` section written by [`Writer::finish`]: the
/// record count, then each record's fixed fields and its two length-prefixed
/// keys. The records must fill the section exactly.
///
/// This is the only layout the V6 format has; there is no key-less variant to
/// fall back to. V6 tables were never released without the keys, and tables in
/// the older format are converted by the offline migration tool, which derives
/// each span from the table's own entries, instead of being read here.
///
/// # Errors
///
/// [`crate::Error::InvalidHeader`] when the count, a key length or the section
/// length disagree with each other.
pub(crate) fn parse_linked_blob_files(buf: &[u8]) -> crate::Result<Vec<LinkedFile>> {
    use crate::io::{LE, ReadBytesExt};

    /// A record with two empty keys: the fewest bytes one can take.
    const MIN_RECORD_LEN: usize = LINKED_BLOB_FILE_FIXED_LEN + 2 + 2;

    fn read_key(reader: &mut &[u8]) -> crate::Result<UserKey> {
        let len = usize::from(reader.read_u16::<LE>()?);
        let Some((key, rest)) = reader.split_at_checked(len) else {
            return Err(crate::Error::InvalidHeader(
                "linked_blob_files: a key runs past the section",
            ));
        };
        *reader = rest;
        Ok(UserKey::from(key))
    }

    let mut reader = buf;
    let count = reader.read_u32::<LE>()? as usize;
    // Bound the declared count by the bytes that remain BEFORE reserving, so a
    // forged count header fails as invalid data rather than as a multi-GB
    // allocation.
    if count > reader.len() / MIN_RECORD_LEN {
        return Err(crate::Error::InvalidHeader(
            "linked_blob_files: declared record count exceeds section size",
        ));
    }
    let mut links = Vec::with_capacity(count);
    for _ in 0..count {
        let blob_file_id = reader.read_u64::<LE>()?;
        let len = usize::try_from(reader.read_u64::<LE>()?).map_err(|_| {
            crate::Error::InvalidHeader("linked_blob_files: entry count exceeds usize")
        })?;
        let bytes = reader.read_u64::<LE>()?;
        let on_disk_bytes = reader.read_u64::<LE>()?;
        let first_key = read_key(&mut reader)?;
        let last_key = read_key(&mut reader)?;
        links.push(LinkedFile {
            blob_file_id,
            bytes,
            on_disk_bytes,
            len,
            first_key,
            last_key,
        });
    }
    if !reader.is_empty() {
        return Err(crate::Error::InvalidHeader(
            "linked_blob_files: bytes remain past the declared records",
        ));
    }
    Ok(links)
}

/// Bytes the payload of an `owned_blob_objects` block holding `count` objects
/// takes: a four-byte count, then each object's blob file and offset.
#[must_use]
pub(crate) fn owned_blob_objects_len(count: usize) -> u64 {
    4 + count as u64 * 16
}

/// Encodes `objects`, sorted by blob file and offset with no repeat, as the
/// payload of an `owned_blob_objects` block.
///
/// # Errors
///
/// [`crate::Error::InvalidHeader`] when there are more objects than the
/// four-byte count holds.
pub(crate) fn encode_owned_blob_objects(
    out: &mut Vec<u8>,
    objects: &[(BlobFileId, u64)],
) -> crate::Result<()> {
    use crate::io::{LE, WriteBytesExt};

    let count = u32::try_from(objects.len()).map_err(|_| {
        crate::Error::InvalidHeader("owned_blob_objects: more objects than the count holds")
    })?;
    out.write_u32::<LE>(count)?;
    for &(blob_file_id, offset) in objects {
        out.write_u64::<LE>(blob_file_id)?;
        out.write_u64::<LE>(offset)?;
    }
    Ok(())
}

/// Parses the payload of an `owned_blob_objects` block: the objects, sorted by
/// blob file and offset with no repeat, filling the payload exactly.
///
/// # Errors
///
/// [`crate::Error::InvalidHeader`] when the count disagrees with the payload
/// length or the objects are not strictly ascending.
pub(crate) fn parse_owned_blob_objects(buf: &[u8]) -> crate::Result<Vec<(BlobFileId, u64)>> {
    use crate::io::{LE, ReadBytesExt};

    let mut reader = buf;
    let count = reader.read_u32::<LE>()? as usize;
    const MISMATCH: crate::Error = crate::Error::InvalidHeader(
        "owned_blob_objects: the count disagrees with the payload length",
    );
    if count.checked_mul(16).ok_or(MISMATCH)? != reader.len() {
        return Err(MISMATCH);
    }
    let mut objects: Vec<(BlobFileId, u64)> = Vec::with_capacity(count);
    for _ in 0..count {
        let object = (reader.read_u64::<LE>()?, reader.read_u64::<LE>()?);
        if objects.last().is_some_and(|&last| last >= object) {
            return Err(crate::Error::InvalidHeader(
                "owned_blob_objects: the objects are not strictly ascending",
            ));
        }
        objects.push(object);
    }
    Ok(objects)
}

/// Heap bytes a vector of handles holds: `logical`, its entries and their
/// end keys, plus the slots its allocation reserves past its `len` entries.
fn handles_held(logical: usize, handles: &[KeyedBlockHandle], capacity: usize) -> usize {
    logical + (capacity - handles.len()) * core::mem::size_of::<KeyedBlockHandle>()
}

/// The [`Writer::register_written_block`] inputs computed once by
/// [`Writer::account_direct_block`] and shared by the columnar-ingest block path
/// and the verbatim copy-through salvage path.
struct DirectBlockInputs {
    last_key: UserKey,
    last_seqno: crate::SeqNo,
    seqno_bounds: Option<(u64, u64)>,
    item_count: usize,
    zone_block_min: Option<UserKey>,
}

/// One compaction input as an output sees it for its age: the input's key
/// range and its `created_at` (see [`AgeSweep`]).
#[derive(Clone, Debug)]
pub(crate) struct InputAge {
    pub min: UserKey,
    pub max: UserKey,
    pub created_at: u128,
}

/// The inputs of one compaction, swept in step with the outputs one writer
/// chain produces in ascending key order: an input joins the live set once
/// an output reaches its min key and leaves it once outputs pass its max key,
/// so each joins and leaves once, whatever its width, and the whole run costs
/// near-linear time in inputs and outputs rather than their product.
pub(crate) struct AgeSweep {
    /// The inputs by ascending min key.
    by_min: Vec<InputAge>,
    /// Each input's rank among the inputs by ascending max key.
    max_rank: Vec<usize>,
    /// The next input of `by_min` to join.
    next: usize,
    /// The live inputs by ascending max rank, as `(rank, index)`.
    live: alloc::collections::BinaryHeap<core::cmp::Reverse<(usize, usize)>>,
    /// The ages of the live inputs, with their counts.
    ages: alloc::collections::BTreeMap<u128, usize>,
    comparator: crate::SharedComparator,
    /// The last range queried, held in debug builds to check the ascending
    /// order the sweep relies on.
    #[cfg(debug_assertions)]
    last_range: Option<(UserKey, UserKey)>,
}

impl AgeSweep {
    pub(crate) fn new(mut inputs: Vec<InputAge>, comparator: crate::SharedComparator) -> Self {
        inputs.sort_by(|a, b| comparator.compare(&a.min, &b.min));
        let mut by_max: Vec<usize> = (0..inputs.len()).collect();
        by_max.sort_by(|&a, &b| match (inputs.get(a), inputs.get(b)) {
            (Some(a), Some(b)) => comparator.compare(&a.max, &b.max),
            _ => core::cmp::Ordering::Equal,
        });
        let mut max_rank = alloc::vec![0; inputs.len()];
        for (rank, index) in by_max.into_iter().enumerate() {
            if let Some(slot) = max_rank.get_mut(index) {
                *slot = rank;
            }
        }
        Self {
            by_min: inputs,
            max_rank,
            next: 0,
            live: alloc::collections::BinaryHeap::new(),
            ages: alloc::collections::BTreeMap::new(),
            comparator,
            #[cfg(debug_assertions)]
            last_range: None,
        }
    }

    /// The newest `created_at` among the inputs whose key range meets
    /// `first..=last`, or `None` when none does. Zero, the age of a table
    /// stamped while the clock read zero, is no age, and an output that meets
    /// such an input has none either. Successive calls must pass ranges that
    /// follow one another in ascending key order, as the outputs of one
    /// writer chain do.
    pub(crate) fn age_of(&mut self, first: &[u8], last: &[u8]) -> Option<u128> {
        use core::cmp::Ordering::{Greater, Less};

        #[cfg(debug_assertions)]
        {
            if let Some((_, previous_last)) = &self.last_range {
                debug_assert!(
                    self.comparator.compare(previous_last, first) != Greater,
                    "outputs swept out of key order"
                );
            }
            self.last_range = Some((first.into(), last.into()));
        }

        // Join every input that starts at or before `last`.
        while let Some(input) = self.by_min.get(self.next) {
            if self.comparator.compare(&input.min, last) == Greater {
                break;
            }
            let rank = self.max_rank.get(self.next).copied().unwrap_or_default();
            self.live.push(core::cmp::Reverse((rank, self.next)));
            *self.ages.entry(input.created_at).or_insert(0) += 1;
            self.next += 1;
        }
        // Leave every input that ends before `first`: the live input with the
        // lowest max rank goes first, so once it reaches `first`, all do.
        while let Some(&core::cmp::Reverse((_, index))) = self.live.peek() {
            let Some(input) = self.by_min.get(index) else {
                break;
            };
            if self.comparator.compare(&input.max, first) != Less {
                break;
            }
            self.live.pop();
            if let Some(count) = self.ages.get_mut(&input.created_at) {
                *count -= 1;
                if *count == 0 {
                    self.ages.remove(&input.created_at);
                }
            }
        }
        // The smallest live age is zero exactly when an input without one is
        // met: then the output's data is partly of unknown age, and dating it
        // by the other inputs would make it expire under a TTL.
        match self.ages.first_key_value() {
            Some((&0, _)) => Some(0),
            _ => self.ages.last_key_value().map(|(&age, _)| age),
        }
    }
}

/// One blob file a table references: how many of its objects the table owns,
/// their bytes, and the span of the table's keys that point there.
///
/// A table whose rows only borrow objects of the file (cell-row references
/// whose owner lives in another version) links it with zero counts.
#[derive(Clone, PartialEq, Eq, Debug, core::hash::Hash)]
pub struct LinkedFile {
    pub blob_file_id: BlobFileId,
    /// Bytes of the objects the table owns, before compression.
    pub bytes: u64,
    /// Bytes of the objects the table owns, as stored.
    pub on_disk_bytes: u64,
    /// Objects of the file the table owns.
    pub len: usize,
    /// The table's first key whose value lives in this blob file.
    pub first_key: UserKey,
    /// The table's last key whose value lives in this blob file.
    pub last_key: UserKey,
}

/// Serializes and compresses values into blocks and writes them to disk as a table.
///
/// `BlockIndexWriter` / `FilterWriter` are generic over a writer `W: Write + Seek`.
/// `Fs::open()` returns `Box<dyn FsFile>` which implements `Write + Seek`,
/// so `BufWriter<Box<dyn FsFile>>` satisfies the required trait bounds.
#[expect(
    clippy::struct_excessive_bools,
    reason = "each bool is an independent on-disk format toggle (bulk-ingest \
              provenance, lineage transformed / last markers, test-only \
              legacy-meta flag); enums would obscure the per-flag wiring"
)]
pub struct Writer {
    /// Filesystem backend
    fs: Arc<dyn Fs>,

    /// Table file path
    pub(crate) path: PathBuf,

    pub(crate) table_id: TableId,

    data_block_restart_interval: u8,
    index_block_restart_interval: u8,

    /// Size a partitioned block index cuts its partitions at.
    index_partition_size: u32,

    /// Size a partitioned filter cuts its partitions at.
    filter_partition_size: u32,

    data_block_size: u32,

    /// Uncompressed bytes a columnar row group is cut at: the spill threshold
    /// in place of `data_block_size` when the table is columnar.
    row_group_size: u32,

    /// Uncompressed bytes of row data a columnar row page is closed at.
    columnar_page_size: u32,

    /// How a columnar table's column pages store their values.
    column_encoding: crate::config::ColumnEncoding,

    data_block_hash_ratio: f32,

    /// Whether the data blocks carry a hash index. Follows
    /// `data_block_hash_ratio`, for the row path.
    hash_indexed: bool,

    /// Whether keys are registered in a filter built here. Follows
    /// `bloom_policy` and `prebuilt_filter`, for the row path.
    filtered: bool,

    /// Whether something keys by a key's hash: the filter, the locator or the
    /// hash index. All three flags are kept by [`Self::refresh_key_hashing`],
    /// which every setter of a field they follow calls.
    hashes_keys: bool,

    /// Compression to use for data blocks
    data_block_compression: CompressionType,

    /// Compression to use for data blocks
    index_block_compression: CompressionType,

    /// Buffer to serialize blocks into
    block_buffer: Vec<u8>,

    /// File writer
    #[expect(clippy::struct_field_names)]
    file_writer: crate::sfa::Writer<ChecksummedWriter<BufWriter<Box<dyn FsFile>>>>,

    /// Writer of index blocks
    #[expect(clippy::struct_field_names)]
    index_writer: Box<dyn BlockIndexWriter<BufWriter<Box<dyn FsFile>>>>,

    /// Writer of filter
    #[expect(clippy::struct_field_names)]
    filter_writer: Box<dyn FilterWriter<BufWriter<Box<dyn FsFile>>>>,

    /// Buffer of KVs
    chunk: Vec<InternalValue>,
    chunk_size: usize,

    /// The user-key hash of each item in `chunk`, kept while the data blocks
    /// carry a hash index, which buckets the items by it; empty otherwise.
    chunk_key_hashes: Vec<u64>,

    /// The hash of the current user key, taken when the key is first seen
    /// and something keys by it: its filter, its locator, the hash index.
    current_key_hash: u64,

    pub(crate) meta: meta::Metadata,

    /// Stores the previous block position (used for creating back links)
    prev_pos: (BlockOffset, BlockOffset),

    /// The last user key written, kept only for when `chunk` no longer holds
    /// it: set when a block leaves the chunk, so a row pays no handle clone.
    /// Read through [`Self::last_key_in`].
    spilled_key: Option<UserKey>,

    /// Seqno of the LAST entry written (any path — row `write()` or a direct
    /// block), tracked alongside the last key so a direct block that begins
    /// with the SAME user key can be validated against the internal order
    /// `(user_key asc, seqno desc)` across the block boundary
    /// ([`Self::validate_direct_block_order`]). `None` only before the first
    /// entry.
    current_key_seqno: Option<crate::SeqNo>,

    bloom_policy: BloomConstructionPolicy,

    /// Stored so `use_partitioned_filter()` can re-apply it to the new writer
    filter_sizing: Option<crate::filter_budget::FilterPlan>,

    /// Stored so `use_partitioned_filter()` can re-apply it to the new writer
    prefix_extractor: Option<Arc<dyn PrefixExtractor>>,

    /// Whether the filter is partitioned, which indexes no prefix.
    partitioned_filter: bool,

    /// Value type of the previously written item (merge-operand ordering
    /// checks in the columnar path).
    previous_type: Option<ValueType>,

    /// Key of the previously written item ONLY when it was a weak tombstone,
    /// so a following `Value` for the same key counts as reclaimable. Kept
    /// separate from `previous_type` so the common path holds no key handle:
    /// one refcount bump and drop per written entry otherwise.
    previous_weak_tombstone_key: Option<UserKey>,

    linked_blob_files: Vec<LinkedFile>,

    /// The blob objects this table's cell rows own, as `(blob file, offset)`,
    /// for the `owned_blob_objects` section.
    owned_blob_objects: Vec<(BlobFileId, u64)>,

    /// Range tombstones to be written as a separate block
    range_tombstones: Vec<RangeTombstone>,

    /// Their least start and greatest end under the tree's comparator, set by
    /// the caller for a table that may hold no record.
    range_tombstone_coverage: Option<(UserKey, UserKey)>,

    /// Inner zstd-block layout per data block, accumulated in write order and
    /// serialized into the optional `block_layout` SST section at finish. Only
    /// data blocks that split into >= 2 inner zstd blocks contribute an entry
    /// `(block_offset, cumulative_decompressed_ends)`; small single-inner-block
    /// blocks (the default) contribute nothing, so the section is absent for a
    /// table with no large multi-inner-block data blocks. Powers range-query
    /// partial decode.
    block_layouts: Vec<(BlockOffset, Vec<u32>)>,

    /// Per-data-block `(offset, (seqno_min, seqno_max))`, accumulated in write
    /// order when `use_seqno_in_index` is on and serialized into the optional
    /// `seqno_bounds` SST section at finish. The bounds live here (a parallel
    /// section) rather than inline in the index entries, so a point read's index
    /// walk never pays for them; only a seqno-scoped scan loads the section.
    /// Empty when `use_seqno_in_index` is off (section absent, zero extra bytes).
    seqno_bounds_section: Vec<(BlockOffset, (u64, u64))>,

    /// Per-data-block zone-map stats, accumulated in write order when
    /// `use_zone_map` is on and serialized into the optional `zone_map` SST
    /// section at finish. Kept parallel to the index (like
    /// [`Self::seqno_bounds_section`]) so a point read never loads it. Empty when
    /// `use_zone_map` is off (section absent, zero extra bytes).
    zone_map_section: Vec<(BlockOffset, Vec<crate::table::zone_map::ColumnStats>)>,

    /// Positional delete-bitmap accumulated for this segment, serialized into
    /// the optional `delete_bitmap` SST section at finish when non-empty. Empty
    /// by default (section absent, zero extra bytes).
    delete_bitmap: crate::table::delete_bitmap::DeleteBitmap,

    /// This segment's delete strategy (the per-SST resolution of the level
    /// policy). Under copy-on-write the bitmap is not persisted (deleted rows are
    /// dropped instead); under merge-on-read / adaptive a non-empty bitmap is
    /// written. Default: adaptive (writes the bitmap).
    delete_strategy: crate::config::DeleteStrategy,

    /// Test-only: omit `descriptor#delete_bitmap_hash` from the meta, producing
    /// the file a pre-hash writer emitted (see `MetaSectionParams`).
    #[cfg(test)]
    pub(crate) omit_delete_bitmap_hash_for_test: bool,

    /// Resolved retrieval-ribbon locator settings for this table, or `None`
    /// (default) when the level's [`crate::config::LocatorPolicy`] is disabled.
    /// `Some` makes the writer accumulate a per-key locator and emit the
    /// optional `locator` section at finish. Wired via [`Self::use_locator`].
    locator: Option<crate::table::locator::LocatorSpec>,

    /// Accumulated `(key_hash, block_id, slot)` triples, one per unique user key
    /// (the newest version's position), captured at the global new-key boundary
    /// in [`Self::write`]. `block_id` is the data block's ordinal; `slot` is the
    /// restart index (or exact entry index for `Entry` precision) of the key's
    /// newest version within that block. Empty unless `locator` is `Some`;
    /// packed and built into the `locator` section at finish.
    locators: Vec<(u64, u64, u64)>,

    /// Ordinal of the data block currently being filled (0-based, in write
    /// order). Incremented once per spilled block, so it matches the block's
    /// eventual registration ordinal in both the serial and parallel paths.
    /// Stamped into each accumulated locator's `block_id`.
    locator_block_id: u64,

    /// Largest locator slot recorded so far, which sizes the section's width.
    locator_max_slot: u64,

    /// A membership filter built elsewhere, written at `finish` in place of
    /// one built from the keys; while it is set no key is registered.
    prebuilt_filter: Option<PrebuiltFilter>,

    /// Whether the filter is one built elsewhere, whose policy this writer
    /// does not know, so the table records no width for it.
    carries_filter: bool,

    /// The prebuilt partition the table's keys have reached.
    prebuilt_partition: usize,

    /// A `locator` section built elsewhere, written at `finish` in place of
    /// one built from the keys.
    prebuilt_locator: Option<Vec<u8>>,

    /// Heap bytes the per-key and per-block state holds for `finish`, and the
    /// bytes `finish` will append, as of the last data block. Refreshed once
    /// per block rather than per key: a table rotates on them at a key
    /// boundary, and one block's keys move neither by much.
    held_state_bytes: u64,
    finish_metadata_bytes: u64,

    /// Heap bytes the entries of `block_layouts` and of `zone_map_section` own
    /// beyond their slots, kept as they are pushed.
    layout_bytes: u64,
    zone_map_bytes: u64,

    /// Bytes of the meta block payload besides its key bounds, taken at the
    /// first block, once the table's settings are final.
    meta_base_len: Option<u64>,

    /// Key count at which the estimates are refreshed between blocks, so the
    /// per-key state a large block gathers is counted before it completes.
    next_refresh_keys: u64,

    /// Hashes registered in the filter, and the count at which the estimates
    /// are refreshed too: a prefix extractor registers many hashes a key, far
    /// more than the key count paces. Without one they equal the key count.
    filter_hashes: u64,
    next_refresh_hashes: u64,

    initial_level: u8,

    /// Block encryption provider (if encryption at rest is enabled)
    encryption: Option<Arc<dyn EncryptionProvider>>,

    /// Per-block Page ECC scheme. `Some(params)` makes every
    /// `Block::write_into` call this writer makes upgrade its
    /// `BlockTransform` to the matching `*Ecc` variant carrying that
    /// scheme (so the writer emits a parity trailer sized by the scheme
    /// and sets the `ECC_PARITY` flag). `None` (default) = no parity.
    /// Caller resolves the `page_ecc` flag and `ecc_scheme` into this
    /// via [`Self::use_ecc`] before the first key is added.
    ecc: Option<crate::table::block::EccParams>,

    /// Durability level for the SST file sync at finish. Default
    /// [`SyncMode::Normal`]; caller wires `Config::sync_mode` via
    /// [`Self::use_sync_mode`].
    sync_mode: SyncMode,

    /// Whether `finish` syncs the file; a deferred one is synced by the
    /// install that persists it. Set via [`Self::use_durability`].
    durability: crate::config::FlushDurability,

    /// Bytes gathered between writebacks the writer starts while writing; `0`
    /// starts none. Set via [`Self::use_writeback_bytes`].
    writeback_bytes: u64,

    /// Where the last writeback ended: every byte before it was handed over.
    written_back: u64,

    /// Per-KV checksum policy + algorithm for data blocks. `None` (default)
    /// means no per-KV checksums: data blocks carry no per-KV footer, with
    /// the `KV_CHECKSUM_FOOTER` header flag clear.
    /// When `Some((policy, algo))`, each data block whose `(level,
    /// table_id)` satisfies `policy.applies` is emitted with a per-entry
    /// checksum footer under `algo` and the `KV_CHECKSUM_FOOTER` flag set
    /// (the block role stays [`BlockType::Data`](crate::table::block::BlockType::Data)). Wired from the tree's
    /// runtime `kv_checksums` config via [`Self::use_kv_checksums`].
    kv_checksum: Option<(
        crate::runtime_config::KvChecksumPolicy,
        crate::runtime_config::ChecksumAlgorithm,
    )>,

    /// Per-block seqno bounds opt-in (the `seqno_in_index` runtime config).
    /// When `true`, the writer accumulates each data block's `(seqno_min,
    /// seqno_max)` and emits them in the optional parallel `seqno_bounds`
    /// SST section, letting a seqno-scoped scan skip blocks that cannot
    /// overlap the target window without reading them. The index entries
    /// stay byte-identical to the off-mode layout, so a point read pays
    /// nothing. Default `false` (no section emitted). Caller wires the live
    /// runtime config into this field via [`Self::use_seqno_in_index`]
    /// before the first key is added.
    use_seqno_in_index: bool,

    /// Zone-map opt-in (the zone-map policy). When `true`, the writer records
    /// each data block's per-column stats (for row blocks: a single synthetic
    /// column with the block's key min / max + row count) and emits them in the
    /// optional parallel `zone_map` SST section, letting a predicate scan skip
    /// blocks that cannot match without reading them. Default `false` (no section
    /// emitted). Caller wires the live runtime config in via
    /// [`Self::use_zone_map`] before the first key is added.
    use_zone_map: bool,

    /// Columnar opt-in. When `true`, each spill writes its entries
    /// column-organized (a PAX row-group of the intrinsic fields) instead of
    /// row-major, as a page directory followed by one
    /// [`BlockType::ColumnPage`](crate::table::block::BlockType::ColumnPage) per
    /// column, so the reader reconstructs the exact entries. Default `false`
    /// (row-major).
    /// Caller wires the live runtime config in via [`Self::use_columnar`] before
    /// the first key is added.
    use_columnar: bool,

    /// How the columnar blocks store each row's value, fixed by the first one
    /// written (or set up front by a rewrite that mirrors its source): a table
    /// holds one layout, since its descriptor records one.
    value_layout: Option<crate::table::meta::ValueLayout>,

    /// Whether this SST is written by the bulk-ingest path, which stores every
    /// entry at LOCAL seqno 0 and relies on a manifest-only `global_seqno` for
    /// its effective MVCC ordering. Persisted as `descriptor#bulk_ingested` so
    /// manifest repair can tell such a table apart from a normal flush whose
    /// entries merely happen to sit at seqno 0 (a fresh tree's first batch),
    /// whose offset genuinely is 0.
    ///
    /// Three-state: `Some(true)` bulk-ingested, `Some(false)` authoritatively a
    /// normal flush / compaction, `None` UNKNOWN. `None` omits the on-disk key
    /// entirely (matching a legacy SST written before the descriptor existed) so
    /// that salvaging a legacy table via [`Self::mirror_from`] preserves the
    /// unknown provenance instead of falsely stamping "not ingested" — which
    /// would let repair install it with `global_seqno` 0 and corrupt MVCC.
    /// Default `Some(false)`.
    bulk_ingested: Option<bool>,

    /// L0 recency key, persisted as `recency`: the table's own id by default,
    /// as a flush or ingest table's is, or the highest
    /// [`Table::l0_recency`](crate::table::Table::l0_recency) among a
    /// compaction output's INPUTS. An output's own id says nothing about how
    /// new its CONTENT is (it is allocated when the compaction starts writing,
    /// while newer flushes with lower ids can install first), so L0 is
    /// ordered by this key.
    recency: TableId,

    /// Compaction lineage for manifest repair: the sorted ids of the INPUT
    /// tables this output was merged from. A crashed compaction that
    /// finalized its outputs before its version edit committed leaves BOTH
    /// histories on disk; the lineage lets a manifest-loss rebuild recognize
    /// the output as DERIVED and exclude it when every input survived,
    /// instead of publishing both and applying the same merge operands twice
    /// on read. `None` — a flush / ingest table — omits the on-disk key.
    /// Persisted as `lineage` (consecutive little-endian ids).
    lineage: Option<Vec<TableId>>,

    /// The PREVIOUS output of the same compaction run (`None` for the run's
    /// first output). Persisted as `lineage_prev`, it proves ADJACENCY:
    /// manifest repair may only union sibling output ranges whose chain is
    /// unbroken — a gap could hide a LOST sibling whose span the union does
    /// not actually cover. Written only alongside `lineage`.
    lineage_prev: Option<TableId>,

    /// A compaction filter TRANSFORMED this output's window (any non-`Keep`
    /// verdict), so its content is not derivable from its inputs — manifest
    /// repair must never trade it back for resurrected inputs, which would
    /// revive the removed values. Persisted as the marker key
    /// `lineage_transformed`, only alongside `lineage`.
    lineage_transformed: bool,

    /// This output is the LAST of its compaction run. Together with an
    /// unbroken adjacency chain from the run's FIRST output (absent
    /// `lineage_prev`), it proves the surviving set is the run's COMPLETE
    /// output set, whose span covers every input end to end — keys outside
    /// the written ranges were consumed by the merge (obsolete versions,
    /// annihilated weak deletes, filter removals), not lost. Persisted as
    /// the marker key `lineage_last`, only alongside `lineage`.
    lineage_last: bool,

    /// This table's place among the files its folder is writing, when its
    /// file came from the folder's pool: the ids below it still being written
    /// when it finishes are persisted as `writing_below`, and the place is
    /// given up once the table is synced (see
    /// [`crate::table::file_pool::Writing`]).
    writing: Option<crate::table::file_pool::Writing>,

    /// The age to record instead of the clock or the inputs' age; `None` for
    /// every table the engine writes itself.
    created_at: Option<u128>,

    /// The tree's clock, read for the age when nothing above sets it; `None`
    /// reads the process clock.
    clock: Option<Arc<dyn crate::Clock>>,

    /// The key range to record instead of the rows' own, set by a caller that
    /// ordered it under the tree's comparator; `None` for every table the
    /// engine writes itself.
    key_range_cover: Option<(UserKey, UserKey)>,

    /// Every tag a row group of this table was written under. Tags are unique
    /// within the table, which is what makes a page's stamp name exactly one
    /// group: a group encoded here takes the first free tag from its own
    /// ([`Self::next_group_tag`]), and a group copied in keeps its own when it
    /// is free.
    #[cfg(feature = "columnar")]
    group_tags: crate::HashSet<u64>,

    /// The age the group tags of this table are derived under, read once at
    /// the first group encoded here.
    #[cfg(feature = "columnar")]
    group_tag_age: Option<u128>,

    /// Pre-trained zstd dictionary for dictionary compression
    #[cfg(zstd_any)]
    zstd_dictionary: Option<Arc<crate::compression::ZstdDictionary>>,

    /// Whether zstd levels 19-22 run the `btultra2` two-pass seed. Defaults to
    /// `true`, the codec's own behaviour; set from the live runtime config via
    /// [`Self::use_zstd_two_pass_seed`].
    #[cfg(zstd_any)]
    zstd_two_pass_seed: bool,

    /// Optional parallel block compression. `None` (default) = serial path:
    /// each block is compressed and written inline. `Some` = blocks are
    /// compressed on worker threads while writes stay ordered here. Wired from
    /// the tree's compaction pool via [`Self::use_parallel_compression`].
    #[cfg(feature = "std")]
    parallel_settings: Option<ParallelCompression>,

    /// Lazily built on the first spill once all transform params are finalized,
    /// so builder-call order doesn't matter. Only `Some` when
    /// `parallel_settings` is set.
    #[cfg(feature = "std")]
    parallel: Option<BlockCompressor>,

    /// Max blocks in flight (submitted but not yet drained) before the writer
    /// drains one to bound buffered-output memory. Derived from thread count.
    #[cfg(feature = "std")]
    parallel_cap: usize,

    /// Per-block index-handle data queued in submission order, popped as each
    /// prepared block is drained and written. Length tracks the pipeline's
    /// `pending` count.
    #[cfg(feature = "std")]
    pending_meta: VecDeque<HandleMeta>,

    /// Uncompressed bytes of submitted-but-not-yet-drained blocks. The parallel
    /// path writes blocks lazily, so `meta.file_pos` lags; this keeps a
    /// synchronous size proxy for size-based table rotation (see
    /// [`Self::output_size_hint`]). Added on submit, subtracted on drain.
    #[cfg(feature = "std")]
    parallel_pending_bytes: u64,

    /// Heap the submitted-but-not-yet-drained blocks hold: their payloads and
    /// frames. Added on submit, subtracted on drain.
    #[cfg(feature = "std")]
    parallel_pending_heap: u64,
}

/// Refuses a batch of more columns than a row group's directory can list: it
/// counts the group's pages in a `u16`, and every column has at least one.
#[cfg(feature = "columnar")]
fn check_group_column_count(batch: &crate::table::columnar::ColumnBatch) -> crate::Result<()> {
    if batch.columns.len() > usize::from(u16::MAX) {
        return Err(crate::Error::InvalidHeader(
            "columnar: more columns than a row group's directory can list",
        ));
    }
    Ok(())
}

/// Where the raw bytes of a verbatim copy were read from: the table they
/// belong to and the offset of their first block, which the copied blocks'
/// stored checksums are bound to until the copy re-stamps them.
#[derive(Clone, Copy, Debug)]
pub(crate) struct VerbatimSource {
    /// The source table's id.
    pub table_id: TableId,
    /// The source offset of the extent's first block.
    pub offset: u64,
}

/// How the table a carried data block comes from wrote its blocks, as its
/// metadata records it: what a table taking the block must share.
#[derive(Clone, Copy, Debug)]
pub(crate) struct CarriedBlockShape {
    /// The data block restart interval.
    pub restart_interval: u8,
    /// The per-KV checksum footer's algorithm, `None` without one.
    pub kv_checksum: Option<crate::runtime_config::ChecksumAlgorithm>,
    /// The bits of the data block hash index ratio.
    pub hash_ratio_bits: u32,
}

/// The pages of another table's row group that a group written here copies:
/// the group keeps the source's tag and its row pages one for one, and on the
/// row pages named copied takes the source's pages as they lie.
#[cfg(feature = "columnar")]
pub(crate) struct PageCarry<'a> {
    /// The source group's tag, which every copied page's stamp names.
    pub tag: u64,
    /// Rows of each row page of the group written, one per source row page.
    pub row_pages: Vec<u32>,
    /// Per row page, whether its pages are the source's.
    pub copied: Vec<bool>,
    /// The source group's bytes on disk.
    pub raw: &'a [u8],
    /// Where they lay.
    pub source: VerbatimSource,
    /// The source group's directory and its on-disk length.
    pub directory: &'a crate::table::column_page::PageDirectory,
    /// The on-disk length of that directory, its header included.
    pub directory_len: u32,
}

#[cfg(feature = "columnar")]
impl<'a> PageCarry<'a> {
    /// Whether the group of `batch`'s rows can take the copied pages: its
    /// columns are the source's, in the source's order, its row pages as many
    /// as the source's, and every copied row page as many rows as the
    /// source's.
    fn fits(&self, batch: &crate::table::columnar::ColumnBatch) -> bool {
        let source_pages = self.directory.row_pages();
        let row_pages = source_pages.len();
        let entries = self.directory.entries();
        row_pages == self.row_pages.len()
            && row_pages == self.copied.len()
            && entries.len() == batch.columns.len() * row_pages
            && self.row_pages.iter().sum::<u32>() == batch.row_count
            && batch.columns.iter().enumerate().all(|(column, col)| {
                entries
                    .get(column * row_pages)
                    .is_some_and(|e| e.id.column_id == col.column_id && e.id.part == 0)
            })
            && self
                .copied
                .iter()
                .zip(self.row_pages.iter().zip(source_pages))
                .all(|(&copied, (rows, source_rows))| !copied || rows == source_rows)
    }

    /// The source's page of the `column`-th column on row page `row_page`,
    /// when that row page is copied.
    fn page(&self, column: usize, row_page: usize) -> Option<CopiedPage<'a>> {
        use crate::coding::Decode;

        if !self.copied.get(row_page).copied().unwrap_or(false) {
            return None;
        }
        let entry = self
            .directory
            .entries()
            .get(column * self.directory.row_pages().len() + row_page)?;
        let start = self.directory_len as usize
            + self.directory.pages_start() as usize
            + entry.offset as usize;
        let frame = self.raw.get(start..start + entry.length as usize)?;
        let header = crate::table::block::Header::decode_from(&mut &*frame).ok()?;
        Some(CopiedPage {
            frame,
            header,
            at: crate::table::block::ChecksumAt::table(
                self.source.table_id,
                self.source.offset + start as u64,
            ),
        })
    }
}

/// A source page a group copies: its frame, its header, and the place its
/// checksum is bound to.
#[cfg(feature = "columnar")]
#[derive(Clone, Copy)]
struct CopiedPage<'a> {
    frame: &'a [u8],
    header: crate::table::block::Header,
    at: crate::table::block::ChecksumAt,
}

/// A page of a group being written: encoded here, or copied.
#[cfg(feature = "columnar")]
enum PagePayload<'a> {
    Encoded(Vec<u8>),
    Copied(CopiedPage<'a>),
}

/// A block of a group being written, ready to go out.
#[cfg(feature = "columnar")]
enum PageOut<'a, 'p> {
    Prepared(crate::table::block::PreparedBlock<'p>),
    Copied(CopiedPage<'a>),
}

#[cfg(feature = "columnar")]
impl PageOut<'_, '_> {
    /// Its length on disk under `ecc`.
    fn on_disk_len(&self, ecc: Option<crate::table::block::EccParams>) -> u32 {
        match self {
            Self::Prepared(prepared) => prepared.on_disk_len(ecc),
            Self::Copied(copied) => copied.header.on_disk_size_with(ecc),
        }
    }
}

/// A membership filter built elsewhere, as the wire bytes it is written as.
pub(crate) enum PrebuiltFilter {
    /// One filter over the whole table.
    Full(Vec<u8>),
    /// Filters over key ranges, in key order: the last key each covers, its
    /// bytes, and the distinct keys of the table that fall in it, which the
    /// writer counts as they arrive (a partition holds no prefix hashes, so
    /// that is the hashes it holds).
    Partitioned(Vec<(UserKey, Vec<u8>, u64)>),
}

impl PrebuiltFilter {
    /// Counts `key`, the table's next distinct key, against the partition
    /// that covers it.
    ///
    /// # Errors
    ///
    /// Returns [`crate::Error::InvalidHeader`] for a key past the last
    /// partition: no filter would answer for it, so a read would miss it.
    fn count_key(
        &mut self,
        key: &[u8],
        cursor: &mut usize,
        comparator: &dyn crate::comparator::UserComparator,
    ) -> crate::Result<()> {
        let Self::Partitioned(partitions) = self else {
            return Ok(());
        };
        while partitions
            .get(*cursor)
            .is_some_and(|(end, _, _)| comparator.compare(key, end) == core::cmp::Ordering::Greater)
        {
            *cursor += 1;
        }
        let (_, _, keys) = partitions
            .get_mut(*cursor)
            .ok_or(crate::Error::InvalidHeader(
                "a key past the last prebuilt filter partition",
            ))?;
        *keys += 1;
        Ok(())
    }
}

/// Where the next block `file_writer` receives lands in table `table_id`'s
/// file: the place its stored checksum is bound to.
pub(crate) fn next_block_at<W: crate::io::Write + crate::io::Seek>(
    table_id: TableId,
    file_writer: &crate::sfa::Writer<ChecksummedWriter<W>>,
) -> crate::table::block::ChecksumAt {
    crate::table::block::ChecksumAt::table(table_id, file_writer.get_ref().position())
}

/// The row group tagged `tag` whose directory and head zone block are
/// `directory_len` and `head_zones_len` bytes on disk, as the index names it,
/// refusing the zero tag and the empty directory no group is written with.
#[cfg(feature = "columnar")]
fn row_group_ref(
    tag: u64,
    directory_len: u32,
    head_zones_len: u32,
) -> crate::Result<crate::table::index_block::RowGroupRef> {
    Ok(crate::table::index_block::RowGroupRef {
        tag: core::num::NonZeroU64::new(tag).ok_or(crate::Error::InvalidHeader(
            "columnar: a row group's tag is zero",
        ))?,
        directory_len: core::num::NonZeroU32::new(directory_len).ok_or(
            crate::Error::InvalidHeader("columnar: a row group's directory is empty"),
        )?,
        head_zones_len,
    })
}

/// The header of a block among bytes copied from another table. A frame too
/// short to hold one is a malformed copy, refused as one: the short read is of
/// bytes in memory, not a failure of the file being written.
fn carried_header(frame: &[u8]) -> crate::Result<crate::table::block::Header> {
    use crate::coding::Decode;
    crate::table::block::Header::decode_from(&mut &*frame).map_err(|e| match e {
        crate::Error::Io(_) => crate::Error::InvalidHeader("carried block header is cut short"),
        e => e,
    })
}

impl Writer {
    pub fn new(
        path: PathBuf,
        table_id: TableId,
        initial_level: u8,
        fs: Arc<dyn Fs>,
    ) -> crate::Result<Self> {
        Self::open(path, table_id, initial_level, fs, None)
    }

    /// [`Self::new`], the file written as `origin`'s.
    pub(crate) fn new_for(
        path: PathBuf,
        table_id: TableId,
        initial_level: u8,
        fs: Arc<dyn Fs>,
        origin: crate::fs::WriteOrigin,
    ) -> crate::Result<Self> {
        Self::open(path, table_id, initial_level, fs, Some(origin))
    }

    fn open(
        path: PathBuf,
        table_id: TableId,
        initial_level: u8,
        fs: Arc<dyn Fs>,
        origin: Option<crate::fs::WriteOrigin>,
    ) -> crate::Result<Self> {
        // Normalize path once so open(), remove_file(), and fsync_directory()
        // all see the same absolute path.
        // no-std: caller must pass an already-absolute path (no cwd concept)
        #[cfg(feature = "std")]
        let path = std::path::absolute(path)?;

        let mut opts = FsOpenOptions::new().write(true).create_new(true);
        opts.write_origin = origin;
        let file = fs.open(&path, &opts)?;
        // From here the file exists and is OURS (`create_new` proved it), so
        // this constructor owns cleanup: an init failure below must not leak a
        // partial file, and a CALLER must never remove `path` on a constructor
        // error — an open failure above created nothing, so caller-side
        // cleanup would race a concurrent creator and delete a file it does
        // not own.
        Self::with_file(path, table_id, initial_level, fs, file)
    }

    /// A writer into `claimed`, an empty pool file under its table's name.
    /// The file is the writer's from here, so an init failure removes it.
    pub(crate) fn into_claimed(
        claimed: crate::table::file_pool::ClaimedFile,
        initial_level: u8,
        fs: Arc<dyn Fs>,
    ) -> crate::Result<Self> {
        let mut writer =
            Self::with_file(claimed.path, claimed.id, initial_level, fs, claimed.file)?;
        writer.writing = Some(claimed.writing);
        Ok(writer)
    }

    fn with_file(
        path: PathBuf,
        table_id: TableId,
        initial_level: u8,
        fs: Arc<dyn Fs>,
        file: Box<dyn FsFile>,
    ) -> crate::Result<Self> {
        let writer = BufWriter::with_capacity(u16::MAX.into(), file);
        let writer = ChecksummedWriter::new(writer);
        let mut writer = crate::sfa::Writer::from_writer(writer);
        if let Err(e) = writer.start("data") {
            drop(writer);
            // Best-effort cleanup of the partial file we just created; log a
            // failure (it can make the next create_new(true) retry fail) rather
            // than discard it, matching the blob-file writer.
            if let Err(remove_err) = fs.remove_file(&path) {
                log::warn!(
                    "failed to remove partial table file {} after a writer-init error: {remove_err}",
                    path.display(),
                );
            }
            return Err(e.into());
        }

        Ok(Self {
            fs,
            initial_level,

            meta: meta::Metadata::default(),

            table_id,

            data_block_restart_interval: 16,
            index_block_restart_interval: 1,

            data_block_hash_ratio: 0.0,
            hash_indexed: false,
            filtered: BloomConstructionPolicy::default().is_active(),
            hashes_keys: BloomConstructionPolicy::default().is_active(),

            index_partition_size: 4_096,
            filter_partition_size: 4_096,

            data_block_size: 4_096,

            row_group_size: crate::config::DEFAULT_COLUMNAR_ROW_GROUP_SIZE,
            columnar_page_size: crate::config::DEFAULT_COLUMNAR_PAGE_SIZE,
            column_encoding: crate::config::ColumnEncoding::Plain,

            data_block_compression: CompressionType::None,
            index_block_compression: CompressionType::None,

            path,

            // Promote to trait object via the use_table_id call —
            // the field type forces the W coercion, no inline cast
            // needed (mirrors the use_partitioned_* pattern below).
            // The trait method returns Box<dyn …<W>> with W inferred
            // from the assignment context.
            index_writer: Box::new(FullIndexWriter::new()).use_table_id(table_id),
            filter_writer: Box::new(FullFilterWriter::new(BloomConstructionPolicy::default()))
                .use_table_id(table_id),

            block_layouts: Vec::new(),
            seqno_bounds_section: Vec::new(),
            zone_map_section: Vec::new(),
            delete_bitmap: crate::table::delete_bitmap::DeleteBitmap::new(),
            delete_strategy: crate::config::DeleteStrategy::default(),
            #[cfg(test)]
            omit_delete_bitmap_hash_for_test: false,

            locator: None,
            locators: Vec::new(),
            locator_block_id: 0,
            locator_max_slot: 0,
            prebuilt_filter: None,
            carries_filter: false,
            prebuilt_partition: 0,
            prebuilt_locator: None,
            held_state_bytes: 0,
            finish_metadata_bytes: 0,
            layout_bytes: 0,
            zone_map_bytes: 0,
            meta_base_len: None,
            next_refresh_keys: 0,
            filter_hashes: 0,
            next_refresh_hashes: 0,

            #[cfg(feature = "columnar")]
            group_tags: crate::HashSet::default(),
            #[cfg(feature = "columnar")]
            group_tag_age: None,

            block_buffer: Vec::new(),
            file_writer: writer,
            chunk: Vec::new(),
            chunk_key_hashes: Vec::new(),
            current_key_hash: 0,

            prev_pos: (BlockOffset(0), BlockOffset(0)),

            chunk_size: 0,

            spilled_key: None,
            current_key_seqno: None,

            bloom_policy: BloomConstructionPolicy::default(),
            filter_sizing: None,

            prefix_extractor: None,
            partitioned_filter: false,

            previous_type: None,
            previous_weak_tombstone_key: None,

            linked_blob_files: Vec::new(),
            owned_blob_objects: Vec::new(),
            range_tombstones: Vec::new(),
            range_tombstone_coverage: None,

            encryption: None,

            ecc: None,
            sync_mode: SyncMode::Normal,
            durability: crate::config::FlushDurability::Synced,
            writeback_bytes: 0,
            written_back: 0,

            kv_checksum: None,
            use_seqno_in_index: false,
            use_zone_map: false,
            use_columnar: false,
            value_layout: None,
            bulk_ingested: Some(false),
            recency: table_id,
            lineage: None,
            lineage_prev: None,
            lineage_transformed: false,
            lineage_last: false,
            writing: None,
            created_at: None,
            clock: None,
            key_range_cover: None,

            #[cfg(zstd_any)]
            zstd_dictionary: None,
            #[cfg(zstd_any)]
            zstd_two_pass_seed: true,

            #[cfg(feature = "std")]
            parallel_settings: None,
            #[cfg(feature = "std")]
            parallel: None,
            #[cfg(feature = "std")]
            parallel_cap: 0,
            #[cfg(feature = "std")]
            pending_meta: VecDeque::new(),
            #[cfg(feature = "std")]
            parallel_pending_bytes: 0,
            #[cfg(feature = "std")]
            parallel_pending_heap: 0,
        })
    }

    /// Current output-size estimate for table-rotation decisions: the bytes
    /// already on disk, the uncompressed size of blocks still in flight on the
    /// parallel pipeline (whose on-disk size isn't known until written), and
    /// the filter, index and other sections `finish` will append.
    pub(crate) fn output_size_hint(&self) -> u64 {
        #[cfg(feature = "std")]
        let data = *self.meta.file_pos + self.parallel_pending_bytes;
        #[cfg(not(feature = "std"))]
        let data = *self.meta.file_pos;
        data + self.finish_metadata_bytes
    }

    /// The part of [`Self::output_size_hint`] that `finish` appends.
    pub(crate) fn finish_metadata_bytes(&self) -> u64 {
        self.finish_metadata_bytes
    }

    /// Bytes a table holding range tombstones alone writes, and holds at the
    /// peak of `finish`, besides its tombstones, bounded from above, with
    /// keys of up to `key_len` bytes yielding up to `prefixes` filter
    /// prefixes: the synthetic weak-tombstone block, the index and filter over
    /// it, every section the writer records it in, the two meta copies and
    /// the fixed tail. The index and filter are counted in their partitioned
    /// layout, which holds the single-level one: a partition, the top level
    /// and its tail mirror. In memory, `finish` builds the filter over the
    /// sentinel's hashes and encodes each piece before it frames it.
    pub(crate) fn tombstone_only_overhead(
        &self,
        key_len: u64,
        prefixes: usize,
    ) -> crate::Result<(u64, u64)> {
        use crate::table::{
            block::{BlockType, EccParams, framed_len_bound, transform_scratch_bound},
            zone_map::ColumnStats,
        };

        let encryption = self.encryption.as_deref();
        let none = CompressionType::None;
        let framed = |len, block_type, compression| {
            framed_len_bound(len, block_type, compression, encryption, self.ecc)
        };
        // An entry holds its key besides a few varints, a seqno and a type.
        let entry = key_len + 32;
        let meta = self.meta_payload_len_without_keys()? + 2 * key_len + 4;
        let index = framed(entry, BlockType::Index, self.index_block_compression);
        let filter = self.bloom_policy.filter_size_bound(1 + prefixes) as u64;
        let seqno_bounds = if self.use_seqno_in_index {
            framed(
                core::mem::size_of::<(BlockOffset, (u64, u64))>() as u64,
                BlockType::SeqnoBounds,
                none,
            )
        } else {
            0
        };
        // One synthetic column holding the sentinel as both bounds.
        let zone_map = if self.use_zone_map {
            framed(
                (core::mem::size_of::<(BlockOffset, Vec<ColumnStats>)>()
                    + core::mem::size_of::<ColumnStats>()) as u64
                    + 2 * key_len,
                BlockType::ZoneMap,
                none,
            )
        } else {
            0
        };
        let locator = self
            .locator
            .and_then(|spec| crate::table::locator::section_size_estimate(1, spec, 0, 0))
            .map_or(0, |section| {
                framed(section as u64, BlockType::Locator, none)
            });
        let written = FIXED_TAIL_LEN
            + 2 * framed_len_bound(
                meta,
                BlockType::Meta,
                none,
                encryption,
                self.ecc.map(|_| EccParams::RS_4_2),
            )
            + framed(entry, BlockType::Data, self.data_block_compression)
            + 3 * index
            + framed(filter, BlockType::Filter, none)
            + index
            + seqno_bounds
            + zone_map
            + locator;
        // The sentinel's hashes, the build over them and the filter it makes,
        // framed; every other piece is encoded before it is framed, which the
        // written bytes bound twice over.
        let members = 1 + prefixes;
        let held = (members * core::mem::size_of::<u64>()) as u64
            + crate::table::filter::ribbon::burr::builder::build_peak_bytes(members, false) as u64
            + filter
            + transform_scratch_bound(filter, BlockType::Filter, none, encryption, self.ecc)
            + 2 * written;
        Ok((written, held))
    }

    /// Bytes of data the table cuts a block at: a row group's for a columnar
    /// table, a data block's otherwise.
    pub(crate) fn block_len(&self) -> u64 {
        u64::from(if self.use_columnar {
            self.row_group_size
        } else {
            self.data_block_size
        })
    }

    /// Bytes the table writes and holds on top of its estimates if range
    /// tombstones widen its key range to bounds of up to `bound_len` bytes:
    /// the meta block carries the range, is written twice, and is encoded and
    /// framed in memory at `finish`. `(0, 0)` before the table has a key.
    pub(crate) fn widened_range_growth(&self, bound_len: u64) -> (u64, u64) {
        use crate::table::block::{BlockType, EccParams, framed_len_bound};

        let (Some(base), Some(first), Some(last)) = (
            self.meta_base_len,
            &self.meta.first_key,
            Self::last_key_in(&self.chunk, self.spilled_key.as_ref())
                .or(self.meta.last_key.as_ref()),
        ) else {
            return (0, 0);
        };
        let (first, last) = (first.len() as u64, last.len() as u64);
        let growth = (bound_len.max(first) - first) + (bound_len.max(last) - last);
        if growth == 0 {
            return (0, 0);
        }
        let payload = base + first + last + 4;
        let frame = |payload| {
            framed_len_bound(
                payload,
                BlockType::Meta,
                CompressionType::None,
                self.encryption.as_deref(),
                self.ecc.map(|_| EccParams::RS_4_2),
            )
        };
        // Framing only grows with its payload.
        let frame_growth = frame(payload + growth) - frame(payload);
        (2 * frame_growth, 2 * growth + frame_growth)
    }

    /// Heap bytes the per-key and per-block state holds until `finish`,
    /// including what `finish` allocates on top to build its sections. It
    /// grows with the keys, not the data bytes, so a table of well-compressing
    /// rows can reach its memory before its size. Blocks in flight on the
    /// parallel pipeline hold their payloads and frames until written.
    ///
    /// The forming block is not counted: it is bounded by the block size, not
    /// by the table, and rotating does not free it, since the next table
    /// builds the same block. Counting it would only close tables earlier
    /// with the same peak.
    pub(crate) fn held_state_bytes(&self) -> u64 {
        #[cfg(feature = "std")]
        let in_flight = self.parallel_pending_heap;
        #[cfg(not(feature = "std"))]
        let in_flight = 0;
        self.held_state_bytes + in_flight
    }

    /// The keys of the forming block are all recorded: the next key belongs to
    /// the next locator ordinal. Called where a block is cut, which on the
    /// parallel path is before the block is written and registered.
    fn locator_block_done(&mut self) {
        let Some(spec) = self.locator else {
            return;
        };
        self.locator_block_id += 1;
        // Positions only grow within a block and block ids across blocks, so
        // the last triple carries the largest block id recorded and its
        // block's largest slot. A block of older versions alone records none;
        // before any triple, block and slot 0 fit every width.
        let (max_block, slot) = self
            .locators
            .last()
            .map_or((0, 0), |&(_, block, slot)| (block, slot));
        let slot_grew = slot > self.locator_max_slot;
        self.locator_max_slot = self.locator_max_slot.max(slot);
        // Widths the table has outgrown stay outgrown: the section would be
        // skipped at `finish`, so stop collecting for it now.
        let outgrown =
            crate::table::locator::section_widths(spec, max_block, self.locator_max_slot).is_none();
        if outgrown {
            log::debug!("locator section dropped at block {max_block}: explicit widths too narrow");
            self.locator = None;
            self.locators = Vec::new();
            self.refresh_key_hashing();
        }
        // A direct block registers before its slots are folded in here, so the
        // estimates it left count neither a wider slot nor a dropped locator.
        if slot_grew || outgrown {
            self.refresh_state_estimates();
        }
    }

    /// The meta block's contents for a table spanning `first_key..=last_key`.
    /// `block_counts` are the index and filter block counts, `delete_bitmap`
    /// the recorded bitmap length and content hash.
    fn meta_section_params<'k>(
        &self,
        first_key: &'k [u8],
        last_key: &'k [u8],
        block_counts: (usize, usize),
        range_tombstone_count: u64,
        delete_bitmap: (u64, u128),
        created_at_nanos: u128,
    ) -> MetaSectionParams<'k> {
        MetaSectionParams {
            section_name: "meta_mid",
            index_block_count: block_counts.0,
            filter_block_count: block_counts.1,
            // MID and TAIL must encode the SAME file_size — recovery
            // falls back transparently and downstream consumers
            // (compaction window picking, etc.) read it as if it were
            // authoritative. `self.meta.file_pos` is only ever bumped
            // inside spill_block, so its value is identical at MID
            // and TAIL write time.
            file_size: *self.meta.file_pos,
            table_id: self.table_id,
            data_block_count: self.meta.data_block_count as u64,
            item_count: self.meta.item_count as u64,
            tombstone_count: self.meta.tombstone_count as u64,
            weak_tombstone_count: self.meta.weak_tombstone_count as u64,
            weak_tombstone_reclaimable: self.meta.weak_tombstone_reclaimable_count as u64,
            key_count: self.meta.key_count as u64,
            filter_hashes: self.meta.filter_hashes,
            // A carried filter was built under its source's policy, not this
            // writer's: the table records none, and the width the filter
            // itself has stands for it.
            filter_bits: (!self.carries_filter)
                .then(|| crate::filter_budget::bits_of(self.bloom_policy)),
            filter_partition_hashes: self.meta.filter_partition_hashes,
            // Borrowed from the writer, which the caller still writes through:
            // set by the caller, see `indexed_extractor`.
            prefix_extractor: None,
            sum_user_key_bytes: self.meta.sum_user_key_bytes,
            sum_value_bytes: self.meta.sum_value_bytes,
            uncompressed_size: self.meta.uncompressed_size,
            first_key,
            last_key,
            lowest_seqno: self.meta.lowest_seqno,
            highest_seqno: self.meta.highest_seqno,
            highest_kv_seqno: self.meta.highest_kv_seqno,
            data_block_compression: self.data_block_compression,
            index_block_compression: self.index_block_compression,
            // Evaluate the kv-checksum policy once for this table's
            // (level, table_id). The result is constant across all the
            // table's data blocks (homogeneous SST), so it doubles as
            // the per-SST descriptor value — identical to the per-block
            // decision `spill_block` makes via the same expression.
            // Columnar blocks carry no per-KV footer (spill_columnar_block writes
            // no footer flags), so a columnar SST must report no footer here
            // rather than advertise one its blocks omit.
            kv_checksum_algo: if self.use_columnar {
                None
            } else {
                self.kv_checksum.and_then(|(policy, algo)| {
                    policy
                        .applies(self.initial_level, self.table_id)
                        .then_some(algo)
                })
            },
            data_block_hash_ratio: self.data_block_hash_ratio,
            data_block_restart_interval: self.data_block_restart_interval,
            index_block_restart_interval: self.index_block_restart_interval,
            initial_level: self.initial_level,
            use_columnar: self.use_columnar,
            value_layout: self
                .value_layout
                .unwrap_or(crate::table::meta::ValueLayout::Whole),
            group_shape: crate::table::meta::GroupShape {
                encoding: self.column_encoding,
                row_group_size: self.row_group_size,
                page_size: self.columnar_page_size,
            },
            bulk_ingested: self.bulk_ingested,
            recency: self.recency,
            lineage: self.lineage.clone(),
            lineage_prev: self.lineage_prev,
            lineage_transformed: self.lineage_transformed,
            lineage_last: self.lineage_last,
            writing_below: self
                .writing
                .as_ref()
                .map(crate::table::file_pool::Writing::below)
                .unwrap_or_default(),
            range_tombstone_count,
            delete_bitmap_len: delete_bitmap.0,
            delete_bitmap_hash: delete_bitmap.1,
            created_at_nanos,
            #[cfg(test)]
            omit_delete_bitmap_hash: self.omit_delete_bitmap_hash_for_test,
        }
    }

    /// Bytes of the meta block payload with empty key bounds, from above.
    ///
    /// Every other value it holds is fixed-width or fixed for the table once
    /// writing starts, so the payload of the finished table is this plus its
    /// key bounds. The lineage markers `finish` may still add are counted.
    fn meta_payload_len_without_keys(&self) -> crate::Result<u64> {
        let mut params = self.meta_section_params(&[], &[], (0, 0), 0, (0, 0), 0);
        // From above: a filter this writer builds is not built yet, so its
        // entries are counted as if it were, under fixed-width counts. A
        // partitioned one records its partitions and names no extractor, a
        // full one the reverse; a writer that builds none writes none of them.
        if self.filtered {
            params.filter_hashes = params.filter_hashes.max(1);
            if self.partitioned_filter {
                params.filter_partition_hashes = params.filter_partition_hashes.max(1);
            }
            params.prefix_extractor = indexed_extractor(
                self.prefix_extractor.as_deref(),
                self.carries_filter,
                self.partitioned_filter,
            );
        }
        let has_lineage = params.lineage.is_some();
        params.lineage_transformed |= has_lineage;
        params.lineage_last |= has_lineage;
        // From above: as many files written below this one as are recorded.
        if self.writing.is_some() {
            params.writing_below =
                alloc::vec![TableId::MAX; crate::table::file_pool::WRITING_BELOW_CAP];
        }
        let mut payload = Vec::new();
        encode_meta_payload(self.ecc, &params, &mut payload)?;
        Ok(payload.len() as u64)
    }

    /// Refreshes the held-state and finish-metadata estimates.
    ///
    /// `finish` builds its sections one after another and frees each input
    /// once built, so the state peaks at the largest of three phases, not at
    /// the sum of every scratch: the block index built over everything still
    /// held; the filter built once the index is gone; the locator built once
    /// the filter is gone. The top-level index bytes stay held to the end,
    /// for the tail mirror.
    fn refresh_state_estimates(&mut self) {
        use crate::table::{
            block::{BlockType, framed_len_bound},
            zone_map::ColumnStats,
        };
        const WORD: u64 = core::mem::size_of::<u64>() as u64;

        let index_held = self.index_writer.held_bytes();
        let index_scratch = self.index_writer.finish_scratch_bytes();
        let filter_held = self.filter_writer.held_bytes();
        let filter_scratch = self.filter_writer.finish_scratch_bytes();

        // Each section's slots, by its allocation, and what its entries own.
        // Its in-memory size bounds its encoding from above; a section with
        // no entry writes no block.
        let none = crate::CompressionType::None;
        let section = |empty: bool, held: u64, block_type| {
            (
                held,
                if empty {
                    0
                } else {
                    framed_len_bound(held, block_type, none, self.encryption.as_deref(), self.ecc)
                },
            )
        };
        let slots = |capacity: usize, slot: usize| (capacity * slot) as u64;
        let sections = [
            section(
                self.block_layouts.is_empty(),
                slots(
                    self.block_layouts.capacity(),
                    core::mem::size_of::<(BlockOffset, Vec<u32>)>(),
                ) + self.layout_bytes,
                BlockType::BlockLayout,
            ),
            section(
                self.seqno_bounds_section.is_empty(),
                slots(
                    self.seqno_bounds_section.capacity(),
                    core::mem::size_of::<(BlockOffset, (u64, u64))>(),
                ),
                BlockType::SeqnoBounds,
            ),
            section(
                self.zone_map_section.is_empty(),
                slots(
                    self.zone_map_section.capacity(),
                    core::mem::size_of::<(BlockOffset, Vec<ColumnStats>)>(),
                ) + self.zone_map_bytes,
                BlockType::ZoneMap,
            ),
        ];
        // The delete bitmap holds and encodes containers, not rows, so it is
        // counted by its own accounting.
        let bitmap = &self.delete_bitmap;
        let sections_held: u64 =
            sections.iter().map(|&(held, _)| held).sum::<u64>() + bitmap.heap_len_bound();
        let bitmap_out = if bitmap.is_empty() {
            0
        } else {
            framed_len_bound(
                bitmap.encoded_len(),
                BlockType::DeleteBitmap,
                none,
                self.encryption.as_deref(),
                self.ecc,
            )
        };
        // `finish` encodes each section into one buffer it keeps to the end,
        // and frames it into a second; the largest section sizes both.
        let section_scratch = sections
            .iter()
            .map(|&(_, out)| out)
            .fold(bitmap_out, u64::max);

        let mut metadata = self.filter_writer.finish_output_bytes()
            + self.index_writer.finish_output_bytes()
            + sections.iter().map(|&(_, out)| out).sum::<u64>()
            + bitmap_out;

        // The tail every table ends with. The meta block holds the key bounds
        // besides its base, each with a length varint that grows from one byte
        // to at most three (keys are at most `u16::MAX` bytes). It is written
        // twice, under the fixed RS(4,2) parity when the table has any, since
        // a reader decodes it before learning the table's scheme. The last key
        // written is the table's last key if it closes now.
        let mut meta_payload = 0;
        let mut meta_frame = 0;
        if let (Some(base), Some(first), Some(last)) = (
            self.meta_base_len,
            &self.meta.first_key,
            Self::last_key_in(&self.chunk, self.spilled_key.as_ref())
                .or(self.meta.last_key.as_ref()),
        ) {
            meta_payload = base + (first.len() + last.len()) as u64 + 4;
            let ecc = self.ecc.map(|_| crate::table::block::EccParams::RS_4_2);
            meta_frame = framed_len_bound(
                meta_payload,
                BlockType::Meta,
                none,
                self.encryption.as_deref(),
                ecc,
            );
            metadata += FIXED_TAIL_LEN + 2 * meta_frame;
        }
        // A compaction's lineage stays in the writer to the end. Encoding the
        // meta block clones it into the parameters and again into its encoded
        // ids, copies those into the meta entries, and encodes the entries into
        // the block buffer, which it then frames.
        let lineage_held = self.lineage.as_ref().map_or(0, |ids| {
            (ids.capacity() * core::mem::size_of::<TableId>()) as u64
        });
        let lineage_ids = self.lineage.as_ref().map_or(0, |ids| {
            (ids.len() * core::mem::size_of::<TableId>()) as u64
        });
        let meta_scratch = 2 * lineage_ids + 2 * meta_payload + meta_frame;
        // Triples of a block that outgrew explicit widths stay allocated until
        // the block is cut and drops the locator, so they count whatever the
        // section estimate says.
        let locator_held =
            (self.locators.capacity() * core::mem::size_of::<(u64, u64, u64)>()) as u64;
        let mut locator_scratch = 0;

        // Block ids only grow, so the last triple carries the largest one
        // recorded; `locator_block_id` is already the next, unwritten block.
        // Its slot is the largest of a block not yet cut, which
        // `locator_max_slot` folds in only once the block is.
        if let Some(spec) = self.locator
            && let Some(&(_, max_block, slot)) = self.locators.last()
            && let Some(section) = crate::table::locator::section_size_estimate(
                self.locators.len(),
                spec,
                max_block,
                self.locator_max_slot.max(slot),
            )
        {
            let n = self.locators.len() as u64;
            let section = section as u64;
            // The split into hashes and packed values, the retrieval build over
            // them, the section bytes and their framed copy.
            locator_scratch = n * 2 * WORD
                + crate::table::filter::ribbon::burr::builder::build_peak_bytes(
                    self.locators.len(),
                    true,
                ) as u64
                + section
                + crate::table::block::transform_scratch_bound(
                    section,
                    BlockType::Locator,
                    none,
                    self.encryption.as_deref(),
                    self.ecc,
                );
            metadata += framed_len_bound(
                section,
                BlockType::Locator,
                none,
                self.encryption.as_deref(),
                self.ecc,
            );
        }

        // The sections are written between the filter and the locator; the
        // buffer they were encoded into stays allocated to the end. The meta
        // block is encoded last.
        let index_phase = index_held + index_scratch + filter_held + locator_held + sections_held;
        let filter_phase =
            index_scratch + filter_held + filter_scratch + locator_held + sections_held;
        // The bitmap's encoding is kept past its frame for the meta's content
        // hash, besides the copy the block buffer takes of it.
        let bitmap_retained = if bitmap.is_empty() {
            0
        } else {
            bitmap.encoded_len()
        };
        let section_phase =
            index_scratch + locator_held + sections_held + 2 * section_scratch + bitmap_retained;
        let locator_phase = index_scratch
            + locator_held
            + locator_scratch
            + sections_held
            + section_scratch
            + bitmap_retained;
        let meta_phase =
            index_scratch + sections_held + section_scratch + meta_scratch + bitmap_retained;

        // The tags of the groups written stay to the end, a slot and a
        // control byte each.
        #[cfg(feature = "columnar")]
        let tags_held = self.group_tags.capacity() as u64 * (WORD + 1);
        #[cfg(not(feature = "columnar"))]
        let tags_held = 0;

        self.held_state_bytes = lineage_held
            + tags_held
            + index_phase
                .max(filter_phase)
                .max(section_phase)
                .max(locator_phase)
                .max(meta_phase);
        self.finish_metadata_bytes = metadata;

        let keys = self.meta.key_count as u64;
        self.next_refresh_keys = keys + REFRESH_KEY_STEP.max(keys / REFRESH_KEY_FRACTION);
        // The same fraction in hashes, from a step that leaves room for a few
        // tokens a key, so an extractor yielding a handful never refreshes on
        // them alone and one yielding thousands does within a key.
        let hashes = self.filter_hashes;
        self.next_refresh_hashes =
            hashes + (REFRESH_HASH_STEP_KEYS * REFRESH_KEY_STEP).max(hashes / REFRESH_KEY_FRACTION);
    }

    /// Enables parallel block compression on this writer: per-block transform
    /// work runs on `parallel`'s workers, up to its thread count at once, and
    /// the in-flight cap (`2 * threads`) bounds buffered-output memory. The
    /// compressor is built lazily on the first spill, so this may be called in
    /// any builder order. Default (not called) keeps the serial path.
    #[cfg(feature = "std")]
    #[must_use]
    pub(crate) fn use_parallel_compression(mut self, parallel: ParallelCompression) -> Self {
        self.parallel_cap = (parallel.threads * 2).max(1);
        self.filter_writer = self.filter_writer.use_parallel(Some(parallel.clone()));
        self.parallel_settings = Some(parallel);
        self
    }

    /// Records that this table references `link`'s blob file, for the
    /// `linked_blob_files` section.
    pub fn link_blob_file(&mut self, link: LinkedFile) {
        self.linked_blob_files.push(link);
    }

    /// Records the blob objects, as `(blob file, offset)`, that this table's
    /// cell rows own, for the `owned_blob_objects` section.
    pub fn own_blob_objects(&mut self, objects: Vec<(BlobFileId, u64)>) {
        if self.owned_blob_objects.is_empty() {
            self.owned_blob_objects = objects;
        } else {
            self.owned_blob_objects.extend(objects);
        }
    }

    fn assert_not_started(&self, setting: &str) {
        // A parallel spill clears `chunk` and submits the block to the
        // pipeline immediately, but `data_block_count` only ticks once
        // `drain_one_parallel` registers it — so neither alone proves "no
        // block is in flight". Also check the pipeline's pending queue /
        // byte counter so a setter can't change the per-table layout after
        // a block was already queued under the old one.
        #[cfg(feature = "std")]
        let in_flight = !self.pending_meta.is_empty() || self.parallel_pending_bytes > 0;
        #[cfg(not(feature = "std"))]
        let in_flight = false;
        assert!(
            self.meta.data_block_count == 0 && self.chunk.is_empty() && !in_flight,
            "{setting} must be configured before writing starts",
        );
    }

    #[must_use]
    pub fn use_partitioned_filter(mut self) -> Self {
        self.assert_not_started("partitioned filter");
        let filter_writer = Box::new(filter::PartitionedFilterWriter::new(self.bloom_policy))
            .use_tli_compression(self.index_block_compression)
            .use_partition_size(self.filter_partition_size)
            .set_prefix_extractor(self.prefix_extractor.clone())
            .use_encryption(self.encryption.clone())
            .use_table_id(self.table_id)
            .use_sizing(self.filter_sizing.clone());
        // Replacing a subwriter re-applies the settings already chosen, so the
        // builder reads the same whatever order its methods are called in.
        #[cfg(zstd_any)]
        let filter_writer = filter_writer.use_zstd_two_pass_seed(self.zstd_two_pass_seed);
        #[cfg(feature = "std")]
        let filter_writer = filter_writer.use_parallel(self.parallel_settings.clone());
        self.filter_writer = filter_writer;
        self.partitioned_filter = true;
        self
    }

    #[must_use]
    pub fn use_partitioned_index(mut self) -> Self {
        self.assert_not_started("partitioned index");
        let index_writer = Box::new(index::PartitionedIndexWriter::new())
            .use_compression(self.index_block_compression)
            .use_partition_size(self.index_partition_size)
            .use_restart_interval(self.index_block_restart_interval)
            .use_encryption(self.encryption.clone())
            .use_table_id(self.table_id)
            // Reapply page_ecc — swapping the index writer would otherwise drop
            // a flag set earlier in the builder chain (order-independence).
            .use_ecc(self.ecc);
        // Same reason as page_ecc above.
        #[cfg(zstd_any)]
        let index_writer = index_writer.use_zstd_two_pass_seed(self.zstd_two_pass_seed);
        self.index_writer = index_writer;
        self
    }

    /// Size-adaptive index: single-level until the index exceeds
    /// `spill_threshold` bytes, then a streaming two-level (partitioned)
    /// index. See [`index::AdaptiveIndexWriter`]. The per-bottom-partition
    /// size (once spilled) is the index partition size, matching
    /// [`Self::use_partitioned_index`].
    #[must_use]
    pub fn use_adaptive_index(mut self, spill_threshold: u64) -> Self {
        self.assert_not_started("adaptive index");
        let index_writer = Box::new(index::AdaptiveIndexWriter::new(spill_threshold))
            .use_compression(self.index_block_compression)
            .use_partition_size(self.index_partition_size)
            .use_restart_interval(self.index_block_restart_interval)
            .use_encryption(self.encryption.clone())
            .use_table_id(self.table_id)
            // Swapping in a fresh index writer drops any flag set earlier in the
            // builder chain; reapply page_ecc so it survives regardless of call
            // order (otherwise the table mixes ECC and non-ECC index blocks
            // while `Writer` still records ECC as enabled).
            .use_ecc(self.ecc);
        // Same reason as page_ecc above.
        #[cfg(zstd_any)]
        let index_writer = index_writer.use_zstd_two_pass_seed(self.zstd_two_pass_seed);
        self.index_writer = index_writer;
        self
    }

    #[must_use]
    pub fn use_data_block_restart_interval(mut self, interval: u8) -> Self {
        assert!(
            interval > 0,
            "data block restart interval must be greater than zero",
        );
        self.assert_not_started("data block restart interval");
        self.data_block_restart_interval = interval;
        self
    }

    #[must_use]
    pub fn use_index_block_restart_interval(mut self, interval: u8) -> Self {
        assert!(
            interval > 0,
            "index block restart interval must be greater than zero",
        );
        self.assert_not_started("index block restart interval");
        self.index_block_restart_interval = interval;
        self.index_writer = self.index_writer.use_restart_interval(interval);
        self
    }

    /// Sets the hash index ratio of the data blocks.
    ///
    /// # Panics
    ///
    /// Panics when a write has started: the table records one ratio for all
    /// its blocks, and a block's index buckets the key hashes kept from its
    /// first item on.
    #[must_use]
    pub fn use_data_block_hash_ratio(mut self, ratio: f32) -> Self {
        self.assert_not_started("data block hash ratio");
        self.data_block_hash_ratio = ratio;
        self.refresh_key_hashing();
        self
    }

    /// Sets the uncompressed size a row-major table cuts its data blocks at.
    /// Ignored for a columnar table, which cuts its row groups at
    /// [`Self::use_row_group_size`].
    ///
    /// # Panics
    ///
    /// Panics when `size` exceeds 4 MiB.
    #[must_use]
    pub fn use_data_block_size(mut self, size: u32) -> Self {
        assert!(
            size <= crate::config::MAX_BLOCK_SIZE,
            "data block size must be <= 4 MiB",
        );
        self.data_block_size = size;
        self
    }

    /// Sets the uncompressed size a columnar row group is cut at. Ignored for
    /// a row-major table, which cuts its blocks at the data block size.
    ///
    /// # Panics
    ///
    /// Panics when `size` exceeds 4 MiB, the same bound as a data block.
    #[must_use]
    pub fn use_row_group_size(mut self, size: u32) -> Self {
        assert!(
            size <= crate::config::MAX_BLOCK_SIZE,
            "row group size must be <= 4 MiB",
        );
        self.row_group_size = size;
        self
    }

    /// Sets the uncompressed size a columnar row group's rows are cut into row
    /// pages at. Ignored for a row-major table.
    ///
    /// # Panics
    ///
    /// Panics when `size` exceeds 4 MiB, the same bound as a row group.
    #[must_use]
    pub fn use_columnar_page_size(mut self, size: u32) -> Self {
        assert!(
            size <= crate::config::MAX_BLOCK_SIZE,
            "columnar page size must be <= 4 MiB"
        );
        self.columnar_page_size = size;
        self
    }

    /// Sets how a columnar table's column pages store their values. Ignored
    /// for a row-major table. Must be set before the first key is written, so
    /// one table's pages follow one policy.
    #[must_use]
    pub fn use_column_encoding(mut self, encoding: crate::config::ColumnEncoding) -> Self {
        self.assert_not_started("column encoding");
        self.column_encoding = encoding;
        self
    }

    /// Sets the partition size of both the partitioned block index and the
    /// partitioned filter; see [`Self::use_index_partition_size`] and
    /// [`Self::use_filter_partition_size`] to set them apart.
    ///
    /// # Panics
    ///
    /// Panics when `size` exceeds 4 MiB.
    #[must_use]
    pub fn use_meta_partition_size(self, size: u32) -> Self {
        self.use_index_partition_size(size)
            .use_filter_partition_size(size)
    }

    /// Sets the size a partitioned block index cuts its partitions at. Takes
    /// effect whether the index is made partitioned before or after.
    ///
    /// # Panics
    ///
    /// Panics when `size` exceeds 4 MiB.
    #[must_use]
    pub fn use_index_partition_size(mut self, size: u32) -> Self {
        assert!(
            size <= crate::config::MAX_BLOCK_SIZE,
            "index partition size must be <= 4 MiB",
        );
        self.index_partition_size = size;
        self.index_writer = self.index_writer.use_partition_size(size);
        self
    }

    /// Sets the size a partitioned filter cuts its partitions at, in the bytes
    /// a partition takes on disk. Takes effect whether the filter is made
    /// partitioned before or after.
    ///
    /// # Panics
    ///
    /// Panics when `size` exceeds 4 MiB.
    #[must_use]
    pub fn use_filter_partition_size(mut self, size: u32) -> Self {
        assert!(
            size <= crate::config::MAX_BLOCK_SIZE,
            "filter partition size must be <= 4 MiB",
        );
        self.filter_partition_size = size;
        self.filter_writer = self.filter_writer.use_partition_size(size);
        self
    }

    #[must_use]
    pub fn use_data_block_compression(mut self, compression: CompressionType) -> Self {
        self.data_block_compression = compression;
        self
    }

    #[must_use]
    pub fn use_index_block_compression(mut self, compression: CompressionType) -> Self {
        // ZstdDict is not useful for index/filter blocks (dictionaries are trained
        // on data block content). Downgrade to plain Zstd to avoid ZstdDictMismatch
        // errors on read — index readers never carry a dictionary.
        #[cfg(zstd_any)]
        let compression = match compression {
            CompressionType::ZstdDict { level, .. } => CompressionType::Zstd(level),
            other => other,
        };

        self.index_block_compression = compression;
        self.index_writer = self.index_writer.use_compression(compression);
        self.filter_writer = self.filter_writer.use_tli_compression(compression);
        self
    }

    /// Sets the encryption provider for block-level encryption at rest.
    ///
    /// When set, all blocks (data, index, filter, meta) are encrypted after
    /// compression and before checksumming.
    #[must_use]
    pub fn use_encryption(mut self, encryption: Option<Arc<dyn EncryptionProvider>>) -> Self {
        self.index_writer = self.index_writer.use_encryption(encryption.clone());
        self.filter_writer = self.filter_writer.use_encryption(encryption.clone());
        self.encryption = encryption;
        self
    }

    /// Selects whether zstd levels 19-22 run the `btultra2` two-pass seed.
    ///
    /// See [`crate::runtime_config::RuntimeConfig::zstd_two_pass_seed`]. Has no
    /// effect below level 19, where another strategy applies.
    #[cfg(zstd_any)]
    #[must_use]
    pub fn use_zstd_two_pass_seed(mut self, enabled: bool) -> Self {
        self.zstd_two_pass_seed = enabled;
        // Index and filter blocks follow their own compression policy, so they
        // run their own encoders and have to be told separately.
        self.index_writer = self.index_writer.use_zstd_two_pass_seed(enabled);
        self.filter_writer = self.filter_writer.use_zstd_two_pass_seed(enabled);
        self
    }

    /// Sets the zstd dictionary for dictionary compression of data blocks.
    #[cfg(zstd_any)]
    #[must_use]
    pub fn use_zstd_dictionary(
        mut self,
        dictionary: Option<Arc<crate::compression::ZstdDictionary>>,
    ) -> Self {
        self.zstd_dictionary = dictionary;
        self
    }

    /// Wires the tree's `Config::page_ecc` flag into this writer.
    /// When `true`, every block this writer emits upgrades its
    /// `BlockTransform` to the matching `*Ecc` variant (so the
    /// `Block::write_into` call emits a Reed-Solomon parity
    /// trailer and sets the `ECC_PARITY` flag in the header).
    /// Must be called BEFORE the first key is added so all blocks
    /// in the table use the same setting; the contract is
    /// enforced by callers (`Tree::open` + compaction worker pass
    /// the config once at writer construction).
    ///
    /// No-op on builds without the `page_ecc` cargo feature —
    /// `BlockTransform::with_ecc` becomes the identity function
    /// in that build and the flag is dead.
    #[must_use]
    pub fn use_ecc(mut self, ecc: Option<crate::table::block::EccParams>) -> Self {
        // The ECC scheme is recorded once per-SST in the final descriptor,
        // so every block in the table must be written under the same layout.
        // Toggling it mid-write would stamp blocks with mixed layouts under a
        // single descriptor; enforce the "configure before first key" contract
        // (shard counts are already non-zero by `EccParams` construction).
        self.assert_not_started("page ecc");
        self.ecc = ecc;
        self.index_writer = self.index_writer.use_ecc(ecc);
        self.filter_writer = self.filter_writer.use_ecc(ecc);
        self
    }

    /// Seeds this writer with the persisted block-layout settings of an existing
    /// table's [`ParsedMeta`](crate::table::meta::ParsedMeta), so a rewrite (e.g.
    /// salvage) reproduces the source's layout instead of falling back to writer
    /// defaults: data + index block compression, ECC scheme, data-block restart
    /// interval, columnar layout (with its zone map), and the per-KV checksum
    /// footer algorithm.
    ///
    /// Only settings the source actually carries are mirrored. Layout choices
    /// that are not persisted in metadata (partitioned filter / index, bloom
    /// policy, hash ratio, locator, prefix extractor) fall back to writer
    /// defaults and are not restored here. Must be called before the first key,
    /// like the individual `use_*` setters it delegates to.
    ///
    /// `has_zone_map` is the source's ACTUAL zone-map presence (a section, not
    /// a metadata property — the caller reads it off the open table): a row SST
    /// written with the zone-map policy keeps its zone map, and a columnar SST
    /// without one does not gain one. `has_seqno_bounds` is likewise the
    /// source's actual `seqno_bounds` SECTION presence (`Table::has_seqno_bounds`,
    /// `regions.seqno_bounds.is_some()`), NOT whether the best-effort loaded map
    /// happens to be non-empty — so a source whose section is present but
    /// unreadable still salvages into a copy that carries one (the writer
    /// re-derives the ranges from the re-emitted entries).
    #[must_use]
    pub(crate) fn mirror_from(
        self,
        meta: &crate::table::meta::ParsedMeta,
        has_zone_map: bool,
        has_seqno_bounds: bool,
    ) -> Self {
        // Mirror ECC only when this build can actually emit parity. Without the
        // `page_ecc` feature `with_ecc()` is the identity (no trailer is ever
        // written), but a `Some` here would still size every registered block
        // handle with phantom parity bytes (`on_disk_size_with`), so the
        // re-emitted table would reopen with handles that over-read into the
        // following blocks. Matches the metadata writer's effective-ECC logic,
        // which likewise resolves to "off" on such builds.
        let effective_ecc = if cfg!(feature = "page_ecc") {
            meta.ecc_params
        } else {
            None
        };
        // A restart interval MUST be > 0 (the setters assert it). A forged or
        // corrupt source meta can carry 0; salvage must not panic on it, so a
        // zero falls back to this writer's own default instead of the setter's
        // assertion.
        let data_restart_interval = if meta.data_block_restart_interval > 0 {
            meta.data_block_restart_interval
        } else {
            self.data_block_restart_interval
        };
        let index_restart_interval = if meta.index_block_restart_interval > 0 {
            meta.index_block_restart_interval
        } else {
            self.index_block_restart_interval
        };
        let writer = self
            .use_data_block_compression(meta.data_block_compression)
            .use_index_block_compression(meta.index_block_compression)
            .use_ecc(effective_ecc)
            .use_data_block_restart_interval(data_restart_interval)
            .use_index_block_restart_interval(index_restart_interval)
            // A columnar source re-emits as columnar (the writer transposes the
            // recovered rows back into PAX blocks), rather than degrading to a
            // row-major copy.
            .use_columnar(meta.columnar)
            // Its blocks are re-emitted as the source stored them, so the
            // copy stores the value the same way.
            .use_value_layout(meta.value_layout)
            // A bulk-ingested source stays flagged so a salvaged / re-emitted
            // copy is still recognized by manifest repair as relying on a
            // manifest-only global_seqno offset. A legacy source of unknown
            // provenance (`None`) re-emits unflagged (we do not guess).
            .use_bulk_ingested(meta.bulk_ingested)
            // A rewrite keeps the source's L0 recency key AND its compaction
            // lineage: the copy carries the same content, so it belongs at the
            // same position and stays recognizable as the same derived output
            // for the manifest-repair lineage dedup.
            .use_recency(meta.recency)
            .use_lineage(meta.lineage.clone())
            .use_lineage_prev(meta.lineage_prev)
            .use_lineage_transformed(meta.lineage_transformed)
            .use_lineage_last(meta.lineage_last)
            .use_zone_map(has_zone_map)
            // A source with a seqno_bounds section keeps its seqno-scoped
            // block-skip: the writer re-derives the per-block ranges from
            // the re-emitted entries (never copies the source's map).
            .use_seqno_in_index(has_seqno_bounds);
        // Groups it copies keep the source's shape, and groups it re-encodes
        // are cut the same way, so the copy records the shape all of them
        // have.
        let writer = match meta.group_shape {
            Some(shape) => writer
                .use_column_encoding(shape.encoding)
                .use_row_group_size(shape.row_group_size)
                .use_columnar_page_size(shape.page_size),
            None => writer,
        };
        // Re-emit per-KV checksum footers under the source's algorithm when it
        // carried them (an SST is footer-homogeneous, so `AllLevels` reproduces
        // the same per-block footer state).
        match meta.kv_checksum_algo {
            Some(algo) => {
                writer.use_kv_checksums(crate::runtime_config::KvChecksumPolicy::AllLevels, algo)
            }
            None => writer,
        }
    }

    /// Convenience wiring: resolves the tree's `Config::page_ecc` flag +
    /// the configured `EccScheme` into the per-block [`EccParams`] and
    /// applies it via [`Self::use_ecc`]. `page_ecc == false` (or a
    /// non-shard scheme such as `Secded`, which is gated until #255)
    /// yields no parity. See [`resolve_ecc`].
    #[must_use]
    pub fn use_page_ecc(self, page_ecc: bool, scheme: crate::runtime_config::EccScheme) -> Self {
        self.use_ecc(resolve_ecc(page_ecc, scheme))
    }

    /// Wires the tree's `Config::sync_mode` into this writer's final SST
    /// sync.
    #[must_use]
    pub fn use_sync_mode(mut self, sync_mode: SyncMode) -> Self {
        self.sync_mode = sync_mode;
        self
    }

    /// Leaves the file unsynced at finish under
    /// [`FlushDurability::Deferred`](crate::config::FlushDurability::Deferred):
    /// the install that persists the table syncs it.
    #[must_use]
    pub(crate) fn use_durability(mut self, durability: crate::config::FlushDurability) -> Self {
        self.durability = durability;
        self
    }

    /// Starts writing back every `bytes` the table gathers while it is
    /// written ([`FsFile::start_writeback`]), so its final sync is short;
    /// `0` starts none.
    #[must_use]
    pub fn use_writeback_bytes(mut self, bytes: u64) -> Self {
        self.writeback_bytes = bytes;
        self
    }

    /// Starts writing back what the table wrote since the last writeback,
    /// once at least `writeback_bytes` of it gathered.
    fn writeback_if_due(&mut self) -> crate::Result<()> {
        // `written_back` only ever takes a value of `file_pos`, which grows.
        let pending = *self.meta.file_pos - self.written_back;
        if self.writeback_bytes == 0 || pending < self.writeback_bytes {
            return Ok(());
        }
        #[cfg(not(feature = "std"))]
        use crate::io::Write;
        #[cfg(feature = "std")]
        use std::io::Write;
        // The range is in the buffer until it is flushed to the file.
        let checksummed = self.file_writer.get_mut();
        checksummed.flush()?;
        crate::fs::hint_writeback(
            &**checksummed.get_ref().get_ref(),
            self.written_back,
            pending,
            &mut self.writeback_bytes,
        );
        self.written_back = *self.meta.file_pos;
        Ok(())
    }

    /// Wires the tree's runtime `kv_checksums` policy + algorithm into this
    /// writer. When `policy != Off`, every data block whose
    /// `(level, table_id)` satisfies `policy.applies` gets a per-entry
    /// checksum footer and the `KV_CHECKSUM_FOOTER` header flag set; all
    /// other data blocks stay plain (flag clear). `Off` (or never calling
    /// this) leaves data blocks without a per-KV footer (flag clear).
    ///
    /// Must be called BEFORE the first key is added (same contract as
    /// [`Self::use_page_ecc`]).
    #[must_use]
    pub fn use_kv_checksums(
        mut self,
        policy: crate::runtime_config::KvChecksumPolicy,
        algo: crate::runtime_config::ChecksumAlgorithm,
    ) -> Self {
        // Must be fixed before the first key: toggling mid-write would mix
        // footer-bearing and plain data blocks in one table at an arbitrary
        // boundary, which the per-SST policy contract does not allow.
        self.assert_not_started("use_kv_checksums");
        self.kv_checksum = if matches!(policy, crate::runtime_config::KvChecksumPolicy::Off) {
            None
        } else {
            Some((policy, algo))
        };
        self
    }

    /// Enables the optional per-block seqno-bounds section (the
    /// `seqno_in_index` runtime config). When `true`, the writer records each
    /// data block's `(seqno_min, seqno_max)` and emits the parallel
    /// `seqno_bounds` section at `finish()`, powering the `scan_since_seqno`
    /// block-skip. Must be called BEFORE the first key is added so the section
    /// covers every block; callers (`Tree` flush + compaction worker) pass the
    /// live config once at writer construction. Default `false` (no section).
    #[must_use]
    pub fn use_seqno_in_index(mut self, seqno_in_index: bool) -> Self {
        // Must be fixed before the first key: the section must cover every
        // block, so the flag is snapshotted per block in `spill_block` and a
        // mid-write toggle would leave some blocks unrecorded. Same contract
        // as `use_page_ecc` / `use_kv_checksums`.
        self.assert_not_started("use_seqno_in_index");
        self.use_seqno_in_index = seqno_in_index;
        self
    }

    /// Enables the zone-map section: the writer records each data block's stats
    /// (for row blocks, a synthetic column with the block's key min / max + row
    /// count) and emits the optional `zone_map` section at finish, for
    /// predicate-based block-skip. Must be fixed before the first key (the
    /// section must cover every block); a mid-write toggle would leave some
    /// blocks unrecorded. Default off.
    #[must_use]
    pub fn use_zone_map(mut self, zone_map: bool) -> Self {
        self.assert_not_started("use_zone_map");
        self.use_zone_map = zone_map;
        self
    }

    /// Mutable access to this segment's positional delete-bitmap, so a caller
    /// (e.g. compaction materialization) can mark rows deleted by position
    /// before [`finish`](Self::finish). A non-empty bitmap is serialized into
    /// the optional `delete_bitmap` SST section.
    pub fn delete_bitmap_mut(&mut self) -> &mut crate::table::delete_bitmap::DeleteBitmap {
        &mut self.delete_bitmap
    }

    /// Sets this segment's delete strategy (the per-SST resolution of the level
    /// policy). Under [`DeleteStrategy::CopyOnWrite`] the delete-bitmap is not
    /// persisted even if rows were marked, since copy-on-write drops deleted rows
    /// rather than masking them. Default: adaptive.
    ///
    /// [`DeleteStrategy::CopyOnWrite`]: crate::config::DeleteStrategy::CopyOnWrite
    #[must_use]
    pub fn delete_strategy(mut self, strategy: crate::config::DeleteStrategy) -> Self {
        self.delete_strategy = strategy;
        self
    }

    /// Enables column-organized data blocks (see [`Self::use_columnar`] field).
    /// Must be set before the first key is written.
    #[must_use]
    pub fn use_columnar(mut self, columnar: bool) -> Self {
        self.assert_not_started("use_columnar");
        // Without the `columnar` feature the spill branch is compiled out, so a
        // requested columnar layout cannot be honored. Drop the flag here so the
        // persisted `descriptor#columnar` never advertises a layout the blocks
        // do not have (which would misroute the reader's block identity).
        self.use_columnar = columnar && cfg!(feature = "columnar");
        self
    }

    /// Sets how the columnar blocks store each row's value, for a rewrite that
    /// re-emits a source's blocks as they were. Must be set before the first
    /// key is written.
    #[must_use]
    pub(crate) fn use_value_layout(mut self, layout: crate::table::meta::ValueLayout) -> Self {
        self.assert_not_started("use_value_layout");
        self.value_layout = Some(layout);
        self
    }

    /// Sets up a columnar table of a tree whose rows may be written as cells
    /// ([`ValueLayout::Cells`](crate::table::meta::ValueLayout::Cells)): its
    /// blocks split each cell row into the columns of its fields. Must be set
    /// before the first key is written; `false` leaves the writer as it is.
    #[must_use]
    pub(crate) fn use_cell_rows(self, cell_rows: bool) -> Self {
        if cell_rows {
            self.use_value_layout(crate::table::meta::ValueLayout::Cells)
        } else {
            self
        }
    }

    /// Records that a columnar block storing values as `layout` is written:
    /// the first fixes the table's layout, and one of the other layout is
    /// refused, since the descriptor records one per table.
    #[cfg(feature = "columnar")]
    fn claim_value_layout(&mut self, layout: crate::table::meta::ValueLayout) -> crate::Result<()> {
        match self.value_layout {
            Some(fixed) if fixed != layout => Err(crate::Error::FeatureUnsupported(
                "a columnar table stores every value one way: whole or split into fields",
            )),
            _ => {
                self.value_layout = Some(layout);
                Ok(())
            }
        }
    }

    /// Sets the bulk-ingest provenance (see [`Self::bulk_ingested`] field), so
    /// the persisted `descriptor#bulk_ingested` lets manifest repair distinguish
    /// a bulk-ingested table from a normal flush whose entries merely sit at
    /// seqno 0. `None` omits the key (unknown provenance, preserving a legacy
    /// SST's absence through salvage). Must be set before the first key is
    /// written.
    #[must_use]
    pub(crate) fn use_bulk_ingested(mut self, bulk_ingested: Option<bool>) -> Self {
        self.assert_not_started("use_bulk_ingested");
        self.bulk_ingested = bulk_ingested;
        self
    }

    /// Sets the recency key (see [`Self::recency`] field): the highest
    /// `l0_recency` among a compaction's inputs, so L0 places this output
    /// where its CONTENT belongs instead of where its id falls. The default,
    /// the table's own id, is a flush or ingest table's.
    #[must_use]
    pub(crate) fn use_recency(mut self, recency: TableId) -> Self {
        self.assert_not_started("use_recency");
        self.recency = recency;
        self
    }

    /// Sets the compaction lineage (see [`Self::lineage`] field): the ids of
    /// the inputs this output merges, sorted for a canonical encoding.
    #[must_use]
    pub(crate) fn use_lineage(mut self, lineage: Option<Vec<TableId>>) -> Self {
        self.assert_not_started("use_lineage");
        self.lineage = lineage.map(|mut ids| {
            ids.sort_unstable();
            ids
        });
        self
    }

    /// Sets the previous-output link (see [`Self::lineage_prev`]).
    #[must_use]
    pub(crate) fn use_lineage_prev(mut self, prev: Option<TableId>) -> Self {
        self.assert_not_started("use_lineage_prev");
        self.lineage_prev = prev;
        self
    }

    /// Replaces the previous-output link mid-write, returning the old one: an
    /// output written after this one but lying below it in key order goes
    /// ahead of it in the run's chain.
    pub(crate) fn relink_lineage_prev(&mut self, prev: Option<TableId>) -> Option<TableId> {
        core::mem::replace(&mut self.lineage_prev, prev)
    }

    /// Marks the pending lineage TRANSFORMED (callable mid-write, unlike the
    /// builder): a compaction FILTER transformed at least one record of THIS
    /// output, so it is no longer derivable from its inputs and must never be
    /// traded back for them by the manifest-repair dedup. The lineage itself
    /// stays — it still proves which inputs this output supersedes.
    pub(crate) fn mark_lineage_transformed(&mut self) {
        self.lineage_transformed = true;
    }

    /// Sets the transformed marker at build time (see
    /// [`Self::lineage_transformed`]); the rewrite path carries a source
    /// table's marker onto its replacement.
    #[must_use]
    pub(crate) fn use_lineage_transformed(mut self, transformed: bool) -> Self {
        self.assert_not_started("use_lineage_transformed");
        self.lineage_transformed = transformed;
        self
    }

    /// Marks this output as its run's LAST (callable at finish time; see
    /// [`Self::lineage_last`]).
    pub(crate) fn mark_lineage_last(&mut self) {
        self.lineage_last = true;
    }

    /// Sets the last-output marker at build time (see [`Self::lineage_last`]);
    /// the rewrite path carries a source table's marker onto its replacement.
    #[must_use]
    pub(crate) fn use_lineage_last(mut self, last: bool) -> Self {
        self.assert_not_started("use_lineage_last");
        self.lineage_last = last;
        self
    }

    /// Sets the age the table records, in nanoseconds since the Unix epoch,
    /// instead of the time it is finished: a table rebuilt from another store
    /// keeps the age of the data it carries.
    #[must_use]
    pub(crate) fn use_created_at(mut self, created_at: u128) -> Self {
        self.assert_not_started("use_created_at");
        self.created_at = Some(created_at);
        self
    }

    /// Reads the time on `clock`, the tree's own, instead of the process clock.
    #[must_use]
    pub(crate) fn use_clock(mut self, clock: Option<Arc<dyn crate::Clock>>) -> Self {
        self.assert_not_started("use_clock");
        self.clock = clock;
        self
    }

    /// The age this table records when nothing set it: now, on its clock.
    fn now_nanos(&self) -> u128 {
        crate::time::now_on(self.clock.as_ref()).as_nanos()
    }

    /// Wires the resolved per-level retrieval-ribbon locator policy entry.
    ///
    /// `Enabled` makes the writer accumulate a per-key `(block_id, slot)`
    /// locator and emit the optional `locator` section at finish; `None` leaves
    /// the writer producing byte-identical SSTs (no section, no padding). Must
    /// be set before the first key, like the other format-affecting `use_*`
    /// toggles, since it changes which sections the table emits.
    #[must_use]
    pub fn use_locator(mut self, entry: crate::config::LocatorPolicyEntry) -> Self {
        self.assert_not_started("use_locator");
        self.locator = match entry {
            crate::config::LocatorPolicyEntry::None => None,
            crate::config::LocatorPolicyEntry::Enabled {
                precision,
                block_id_bits,
                slot_bits,
            } => Some(crate::table::locator::LocatorSpec {
                precision,
                block_id_bits,
                slot_bits,
            }),
        };
        self.refresh_key_hashing();
        self
    }

    /// Derives the row path's flags from the settings they follow.
    fn refresh_key_hashing(&mut self) {
        self.hash_indexed = self.data_block_hash_ratio > 0.0;
        self.filtered = self.bloom_policy.is_active() && self.prebuilt_filter.is_none();
        self.hashes_keys = self.filtered || self.locator.is_some() || self.hash_indexed;
    }

    /// Writes `filter`, built elsewhere for exactly the keys this table will
    /// hold, as the table's membership filter instead of building one from
    /// them. Must be set before the first key.
    #[must_use]
    pub(crate) fn use_prebuilt_filter(mut self, filter: PrebuiltFilter) -> Self {
        self.assert_not_started("use_prebuilt_filter");
        self.prebuilt_filter = Some(filter);
        self.carries_filter = true;
        self.refresh_key_hashing();
        self
    }

    /// Writes `section`, a `locator` section built elsewhere for exactly the
    /// keys and blocks this table will hold, instead of building one from
    /// them. Must be set before the first key.
    #[must_use]
    pub(crate) fn use_prebuilt_locator(mut self, section: Vec<u8>) -> Self {
        self.assert_not_started("use_prebuilt_locator");
        self.locator = None;
        self.prebuilt_locator = Some(section);
        self.refresh_key_hashing();
        self
    }

    /// When `enabled`, clears per-file copy-on-write on this table's (still
    /// empty) file via [`crate::fs::Fs::try_disable_cow`], so a write-once SST
    /// on a copy-on-write filesystem (Btrfs) avoids the fragmentation penalty.
    ///
    /// Called at construction, before any block is written, because the inode
    /// flag only takes effect on a file with no data blocks yet. Best-effort:
    /// a failure (or a filesystem that does not support the flag) is logged and
    /// ignored: disabling copy-on-write is a throughput optimization, never a
    /// correctness requirement, so it must not fail SST creation.
    #[must_use]
    pub fn use_disable_cow(self, enabled: bool) -> Self {
        if enabled && let Err(e) = self.fs.try_disable_cow(&self.path) {
            log::warn!(
                "try_disable_cow({}) failed; continuing with CoW enabled: {e}",
                self.path.display(),
            );
        }
        self
    }

    #[must_use]
    pub fn use_bloom_policy(mut self, bloom_policy: BloomConstructionPolicy) -> Self {
        self.bloom_policy = bloom_policy;
        self.filter_writer = self.filter_writer.set_filter_policy(bloom_policy);
        self.refresh_key_hashing();
        self
    }

    /// Sizes this table's filters by probe load against the tree's filter
    /// budget; `None` builds them at the bloom policy.
    #[must_use]
    pub(crate) fn use_filter_sizing(
        mut self,
        sizing: Option<crate::filter_budget::FilterPlan>,
    ) -> Self {
        self.filter_sizing.clone_from(&sizing);
        self.filter_writer = self.filter_writer.use_sizing(sizing);
        self
    }

    #[must_use]
    pub fn use_prefix_extractor(mut self, extractor: Option<Arc<dyn PrefixExtractor>>) -> Self {
        self.prefix_extractor.clone_from(&extractor);
        self.filter_writer = self.filter_writer.set_prefix_extractor(extractor);
        self
    }

    /// Records `seqno` as the table's KV seqno bound instead of the one its
    /// entries give: for a table rebuilt from another store, whose writer
    /// counted an entry it synthesized for range tombstones alone as no KV.
    #[cfg(feature = "std")]
    pub(crate) fn record_highest_kv_seqno(&mut self, seqno: crate::SeqNo) {
        self.meta.highest_kv_seqno = seqno;
    }

    /// Adds a range tombstone to be written into this table's RT block.
    pub(crate) fn write_range_tombstone(&mut self, rt: RangeTombstone) {
        self.meta.lowest_seqno = self.meta.lowest_seqno.min(rt.seqno);
        self.meta.highest_seqno = self.meta.highest_seqno.max(rt.seqno);
        self.range_tombstones.push(rt);
    }

    /// The least start and greatest end of the tombstones written, under the
    /// tree's comparator: the key range of a table that holds no record, which
    /// this writer, knowing only byte order, cannot take from them itself.
    pub(crate) fn cover_range_tombstones(&mut self, start: UserKey, end: UserKey) {
        self.range_tombstone_coverage = Some((start, end));
    }

    /// Records `[start, end]` as the table's key range, whatever entries it
    /// holds: for a table rebuilt from another store, whose range tombstones
    /// reach past its entries. The caller orders the range under the tree's
    /// comparator, and it covers every entry the table receives.
    pub(crate) fn cover_key_range(&mut self, start: UserKey, end: UserKey) {
        self.key_range_cover = Some((start, end));
    }

    /// Writes an item.
    ///
    /// # Note
    ///
    /// It's important that the incoming stream of items is correctly
    /// sorted as described by the [`UserKey`], otherwise the block layout will
    /// be non-sense.
    pub fn write(&mut self, item: InternalValue) -> crate::Result<()> {
        let continues = Self::last_key_in(&self.chunk, self.spilled_key.as_ref())
            .is_some_and(|c| crate::comparator::same_user_key(c, &item.key.user_key));
        self.write_keyed(item, continues)
    }

    /// [`Self::write`] of a row whose caller has compared its key with the
    /// last row it wrote: `continues` says they are the same. This table may
    /// hold no row of that key yet, when it was rotated to between versions.
    pub(crate) fn write_keyed(
        &mut self,
        item: InternalValue,
        continues: bool,
    ) -> crate::Result<()> {
        // A prebuilt partitioned filter counts the keys each partition covers,
        // which needs the comparator; an entry written here would escape that.
        if matches!(self.prebuilt_filter, Some(PrebuiltFilter::Partitioned(_))) {
            return Err(crate::Error::InvalidHeader(
                "a table with a prebuilt partitioned filter takes its entries with a comparator",
            ));
        }
        let new_key =
            !continues || Self::last_key_in(&self.chunk, self.spilled_key.as_ref()).is_none();
        self.write_entry(item, new_key)
    }

    /// [`Self::write`] into a table with a prebuilt filter of any shape: a new
    /// key is counted against the partition that covers it, and a key none
    /// covers is refused before it is written.
    ///
    /// # Errors
    ///
    /// The last user key written. Takes the two fields rather than `&self` so a
    /// caller can hold another field mutably.
    #[inline]
    fn last_key_in<'a>(
        chunk: &'a [InternalValue],
        spilled_key: Option<&'a UserKey>,
    ) -> Option<&'a UserKey> {
        chunk
            .last()
            .map(|entry| &entry.key.user_key)
            .or(spilled_key)
    }

    /// The last user key written, by any path.
    pub(crate) fn last_key(&self) -> Option<&UserKey> {
        Self::last_key_in(&self.chunk, self.spilled_key.as_ref())
    }

    /// As [`Self::write`], plus the refusal of an uncovered key.
    pub(crate) fn write_counted(
        &mut self,
        item: InternalValue,
        comparator: &crate::SharedComparator,
    ) -> crate::Result<()> {
        let new_key = !Self::last_key_in(&self.chunk, self.spilled_key.as_ref())
            .is_some_and(|c| crate::comparator::same_user_key(c, &item.key.user_key));
        if new_key && let Some(prebuilt) = self.prebuilt_filter.as_mut() {
            prebuilt.count_key(
                &item.key.user_key,
                &mut self.prebuilt_partition,
                comparator.as_ref(),
            )?;
        }
        self.write_entry(item, new_key)
    }

    /// The body of [`Self::write`], past the prebuilt-filter guard. `new_key`
    /// is whether the table holds no row of this key yet.
    fn write_entry(&mut self, item: InternalValue, new_key: bool) -> crate::Result<()> {
        let value_type = item.key.value_type;
        let seqno = item.key.seqno;
        // Borrow the key for the bookkeeping below; `item` moves into the
        // chunk as-is at the end, so no per-entry handle clone is needed.
        let user_key = &item.key.user_key;
        let user_key_len = user_key.len();
        let value_len = item.value.len();

        // Per-entry shape accounting (pairs with item_count: counts every
        // version). usize -> u64 is a widening cast on every supported target.
        self.meta.sum_user_key_bytes += user_key_len as u64;
        self.meta.sum_value_bytes += value_len as u64;

        if item.is_tombstone() {
            self.meta.tombstone_count += 1;
        }

        if value_type == ValueType::WeakTombstone {
            self.meta.weak_tombstone_count += 1;
        }

        if value_type.is_put()
            && let Some(prev_key) = &self.previous_weak_tombstone_key
            && crate::comparator::same_user_key(prev_key, user_key)
        {
            self.meta.weak_tombstone_reclaimable_count += 1;
        }

        // NOTE: Check if we visit a new key. Identity is the engine's one
        // relation — see `same_user_key`; the filter registered just below hashes
        // these same raw bytes, which is why it can be nothing else.
        if new_key {
            self.meta.key_count += 1;

            // The tail is sized from the first key on, so a table that closes
            // before its first block is written still counts it.
            if self.meta.first_key.is_none() {
                self.meta.first_key = Some(user_key.clone());
                self.meta_base_len = Some(self.meta_payload_len_without_keys()?);
            }

            // IMPORTANT: Do not buffer *every* item's key
            // because there may be multiple versions
            // of the same key

            if self.hashes_keys {
                // The filter, the locator and the hash index key the same
                // bytes: one hash.
                let key_hash = crate::hash::hash64(user_key);
                self.current_key_hash = key_hash;
                if self.filtered {
                    self.filter_hashes +=
                        self.filter_writer.register_key(user_key, key_hash)? as u64;
                }

                // Retrieval-ribbon locator: record this key's newest version
                // (its first occurrence in scan order) at its position in the
                // forming data block. `chunk.len()` is the item index this key
                // will take (it is pushed below at the same index);
                // `locator_block_id` is the current block's ordinal. `slot` is
                // the restart index the encoder will assign
                // (`item_index / restart_interval`) for `Restart`, or the exact
                // item index for `Entry`.
                if let Some(spec) = self.locator {
                    let pos = self.chunk.len() as u64;
                    let slot = match spec.precision {
                        // Per-block precision drops slot (the section build
                        // masks it to 0-width); record 0 to keep max_slot
                        // minimal.
                        crate::config::LocatorPrecision::Block => 0,
                        crate::config::LocatorPrecision::Restart => {
                            pos / u64::from(self.data_block_restart_interval)
                        }
                        crate::config::LocatorPrecision::Entry => pos,
                    };
                    self.locators.push((key_hash, self.locator_block_id, slot));
                }
            }
        }

        // Only a weak tombstone's key is retained (for the reclaimable check
        // on the next entry); every other entry leaves no handle behind.
        let weak_tombstone_key = (value_type == ValueType::WeakTombstone).then(|| user_key.clone());

        self.chunk_size += self.entry_size(user_key_len, value_len);
        self.chunk.push(item);
        if self.hash_indexed {
            // Every version of a key shares its hash.
            self.chunk_key_hashes.push(self.current_key_hash);
        }
        self.previous_type = Some(value_type);
        self.previous_weak_tombstone_key = weak_tombstone_key;
        self.current_key_seqno = Some(seqno);

        // The per-key state grows between blocks too; see `REFRESH_KEY_STEP`. A
        // prefix extractor grows the filter by many hashes a key, so its state
        // paces the refresh as well. After the push: the estimates size the
        // table's last key from the chunk.
        if self.meta.key_count as u64 >= self.next_refresh_keys
            || self.filter_hashes >= self.next_refresh_hashes
        {
            self.refresh_state_estimates();
        }

        let cut_at = if self.use_columnar {
            self.row_group_size
        } else {
            self.data_block_size
        };
        if self.chunk_size >= cut_at as usize {
            self.spill_block()?;
        }

        self.meta.lowest_seqno = self.meta.lowest_seqno.min(seqno);
        self.meta.highest_seqno = self.meta.highest_seqno.max(seqno);
        // highest_kv_seqno tracks the highest seqno among user KV entries
        // written via write() (values, point tombstones, weak tombstones).
        // Range tombstones (via write_range_tombstone) are excluded. In
        // RT-only tables, finish() writes a synthetic sentinel via write()
        // but restores highest_kv_seqno afterwards, so this bound reflects
        // only actual user KV items.
        self.meta.highest_kv_seqno = self.meta.highest_kv_seqno.max(seqno);

        Ok(())
    }

    /// What an entry of a `key_len`-byte key and a `value_len`-byte value
    /// counts toward the block or row group it is written into. A row group
    /// counts it as its row pages do, every column's share of the row, so a
    /// group size and a page size of the same bytes hold the same rows.
    #[cfg_attr(
        not(feature = "columnar"),
        expect(
            clippy::unused_self,
            reason = "only a columnar writer counts differently"
        )
    )]
    fn entry_size(&self, key_len: usize, value_len: usize) -> usize {
        #[cfg(feature = "columnar")]
        if self.use_columnar {
            return crate::table::columnar::entry_row_bytes(key_len, value_len);
        }
        key_len + value_len
    }

    /// Writes a compressed block to disk.
    ///
    /// This is triggered when a `Writer::write` causes the buffer to grow to the configured `block_size`.
    ///
    /// Should only be called when the block has items in it.
    pub(crate) fn spill_block(&mut self) -> crate::Result<()> {
        let Some(last) = self.chunk.last() else {
            return Ok(());
        };

        // Order-independent index-handle data, captured before the chunk is
        // consumed. The parallel path needs it owned because the chunk is
        // cleared at submit time, before a worker has produced the block. It
        // stays with the index until the table ends, so it holds its own bytes
        // rather than whatever block or key chunk the row's key is a view of.
        let last_key = UserKey::new(&last.key.user_key);
        let last_seqno = last.key.seqno;
        let item_count = self.chunk.len();
        // The chunk stops holding the last key once this block leaves it.
        self.spilled_key = Some(last_key.clone());

        // This spill flushes the block whose keys were recorded with the current
        // `locator_block_id`, so the next block's keys belong to the next
        // ordinal. Done here (not per write path) so it stays correct for both
        // the serial and parallel spills.
        self.locator_block_done();
        // Per-block seqno bounds let a seqno-scoped scan skip this whole block
        // when its `max` is below the target: one fold over entries in hand.
        let seqno_bounds = self.use_seqno_in_index.then(|| {
            self.chunk
                .iter()
                .fold((u64::MAX, u64::MIN), |(min, max), e| {
                    (min.min(e.key.seqno), max.max(e.key.seqno))
                })
        });
        // Block's first (min) key for the zone-map synthetic column; cloned only
        // when the policy is on. The chunk is non-empty here (early return
        // above), so `first()` is `Some`.
        let zone_block_min = self
            .use_zone_map
            .then(|| self.chunk.first().map(|e| e.key.user_key.clone()))
            .flatten();

        // Columnar layout: transpose the chunk into a PAX block rather than a
        // row-major data block. Serial path only for now (the parallel pipeline
        // assumes a Data block); columnar is opt-in.
        #[cfg(feature = "columnar")]
        if self.use_columnar {
            return self.spill_columnar_block(
                last_key,
                last_seqno,
                seqno_bounds,
                item_count,
                zone_block_min,
            );
        }

        // Decide per-block whether to emit the per-KV checksum footer.
        // `kv_checksum` is None unless the tree opted in; even then only blocks
        // whose (level, table_id) satisfy the policy carry the footer. A
        // per-KV-checked block is a plain Data block plus a footer (role stays
        // Data; the footer is a header flag bit, not a distinct block type).
        let kv_emit = self.kv_checksum.and_then(|(policy, algo)| {
            policy
                .applies(self.initial_level, self.table_id)
                .then_some(algo)
        });

        // Parallel path: encode into a fresh owned buffer (the reusable
        // block_buffer can't serve here — it would be in flight on a worker),
        // submit it, and let an ordered drain write it later.
        #[cfg(feature = "std")]
        if self.parallel_settings.is_some() {
            self.ensure_parallel();
            let mut encoded = Vec::with_capacity(self.chunk_size);
            let kv_flags = Self::encode_chunk_into(
                &self.chunk,
                Self::key_hashes_of(&self.chunk, &self.chunk_key_hashes),
                self.data_block_restart_interval,
                self.data_block_hash_ratio,
                kv_emit,
                &mut encoded,
            )?;
            // Backpressure: bound buffered output by draining before the cap.
            while self.parallel.as_ref().map_or(0, BlockCompressor::pending) >= self.parallel_cap {
                self.drain_one_parallel()?;
            }
            // The rotation size hint counts the block by the frame it will be
            // written as (file_pos only advances once it is drained): the
            // header, and the codec, encryption and parity overheads.
            let frame_bound = crate::table::block::framed_len_bound(
                encoded.len() as u64,
                super::block::BlockType::Data,
                self.data_block_compression,
                self.encryption.as_deref(),
                self.ecc,
            );
            // Its heap in flight: the encoded payload, and what a worker
            // prepares from it, the compressed bytes and the sealed frame both
            // when the block is compressed and encrypted.
            let heap_bound = encoded.capacity() as u64
                + crate::table::block::transform_scratch_bound(
                    encoded.len() as u64,
                    super::block::BlockType::Data,
                    self.data_block_compression,
                    self.encryption.as_deref(),
                    self.ecc,
                );
            self.pending_meta.push_back(HandleMeta {
                last_key,
                last_seqno,
                seqno_bounds,
                item_count,
                zone_block_min,
                frame_bound,
                heap_bound,
            });
            self.parallel_pending_bytes += frame_bound;
            self.parallel_pending_heap += heap_bound;
            if let Some(par) = self.parallel.as_mut() {
                par.submit(encoded, kv_flags);
            }
            self.chunk.clear();
            self.chunk_key_hashes.clear();
            self.chunk_size = 0;
            return Ok(());
        }

        // Serial path: encode into the reusable buffer, prepare + write inline.
        self.block_buffer.clear();
        let kv_flags = Self::encode_chunk_into(
            &self.chunk,
            Self::key_hashes_of(&self.chunk, &self.chunk_key_hashes),
            self.data_block_restart_interval,
            self.data_block_hash_ratio,
            kv_emit,
            &mut self.block_buffer,
        )?;

        // Prepare then write in two steps (instead of `write_into_with_flags`)
        // so the inner-block layout can be taken from the prepared block before
        // `write_to` consumes it.
        let transform = {
            let t = crate::table::block::BlockTransform::from_parts(
                self.data_block_compression,
                self.encryption.as_deref(),
                #[cfg(zstd_any)]
                self.zstd_dictionary.as_deref(),
            )?;
            #[cfg(zstd_any)]
            let t = t.with_two_pass_seed(self.zstd_two_pass_seed);
            if let Some(ecc) = self.ecc {
                t.with_ecc(ecc)
            } else {
                t
            }
        };
        let mut prepared = Block::prepare_with_flags(
            &self.block_buffer,
            super::block::BlockIdentity {
                table_id: self.table_id,
                block_type: super::block::BlockType::Data,
                dict_id: self.data_block_compression.dict_id(),
                window_log: 0,
            },
            // Data blocks use the configured codec and may carry a zstd dict;
            // encryption is optional. page_ecc upgrades the transform to its
            // matching `*Ecc` variant when the tree was opened with page_ecc.
            &transform,
            kv_flags,
        )?;
        let layout = core::mem::take(&mut prepared.layout);
        let at = next_block_at(self.table_id, &self.file_writer);
        let header = prepared.write_to(&mut self.file_writer, at)?;

        self.register_written_block(
            header,
            layout,
            last_key,
            last_seqno,
            seqno_bounds,
            item_count,
            zone_block_min,
            None, // row block: register synthesizes the whole-block key range
        )?;

        // IMPORTANT: Clear chunk after everything else
        self.chunk.clear();
        self.chunk_key_hashes.clear();
        self.chunk_size = 0;

        Ok(())
    }

    /// Spills the current chunk as a columnar (PAX) block: transpose the entries
    /// into a `ColumnBatch`, encode it, and write + register it like a data block
    /// (same index handle, seqno bounds, and zone-map key range). Serial path
    /// only; the parallel pipeline keeps producing row blocks.
    #[cfg(feature = "columnar")]
    fn spill_columnar_block(
        &mut self,
        last_key: crate::UserKey,
        last_seqno: crate::SeqNo,
        seqno_bounds: Option<(u64, u64)>,
        item_count: usize,
        zone_block_min: Option<crate::UserKey>,
    ) -> crate::Result<()> {
        // The transpose keeps each row's value whole, as it was written, or,
        // in a table set up for cell rows, splits each cell row into its
        // fields and keeps every other value whole.
        let batch = if self.value_layout == Some(crate::table::meta::ValueLayout::Cells) {
            crate::table::columnar::entries_to_cells_batch(&self.chunk)?
        } else {
            self.claim_value_layout(crate::table::meta::ValueLayout::Whole)?;
            crate::table::columnar::entries_to_column_batch(&self.chunk)?
        };
        self.encode_columnar_batch_block(
            &batch,
            last_key,
            last_seqno,
            seqno_bounds,
            item_count,
            zone_block_min,
            None,
        )?;
        self.chunk.clear();
        self.chunk_key_hashes.clear();
        self.chunk_size = 0;
        Ok(())
    }

    /// Writes `rows` as one row group that keeps another table's row group's
    /// tag and copies the pages of its row pages `carry` names as unchanged,
    /// as they lie on disk, encoding only the rest.
    ///
    /// Returns `false`, writing nothing, when this table cannot take the group
    /// so: a row-major table, one whose values are not stored whole, one that
    /// already holds the tag, or rows whose columns are not the source's, in
    /// its order. The caller then writes the rows.
    ///
    /// # Errors
    ///
    /// As [`Self::append_verbatim_row_group`].
    #[cfg(feature = "columnar")]
    pub(crate) fn append_partly_carried_row_group(
        &mut self,
        rows: &[InternalValue],
        carry: &PageCarry<'_>,
        comparator: &crate::SharedComparator,
    ) -> crate::Result<bool> {
        use crate::table::meta::ValueLayout;

        if !self.use_columnar
            || self
                .value_layout
                .is_some_and(|fixed| fixed != ValueLayout::Whole)
            || !self.accepts_group_tag(carry.tag)
        {
            return Ok(false);
        }
        // A key two inputs held at one seqno comes out twice, an order the
        // block check refuses; such rows go through the ordinary path, as for
        // a group copied whole.
        if self.validate_direct_block_order(rows, comparator).is_err() {
            return Ok(false);
        }
        let batch = crate::table::columnar::entries_to_column_batch(rows)?;
        if !carry.fits(&batch) {
            return Ok(false);
        }
        self.claim_value_layout(ValueLayout::Whole)?;
        let Some(inputs) = self.account_direct_block(rows)? else {
            return Ok(false);
        };
        self.encode_columnar_batch_block(
            &batch,
            inputs.last_key,
            inputs.last_seqno,
            inputs.seqno_bounds,
            inputs.item_count,
            inputs.zone_block_min,
            Some(carry),
        )?;
        self.locator_block_done();
        Ok(true)
    }

    /// Encodes a `ColumnBatch` as a columnar (PAX) block and registers it like a
    /// data block (index handle, seqno bounds, zone-map key range). Shared by the
    /// intrinsic transpose spill ([`Self::spill_columnar_block`]) and the
    /// consumer columnar-ingest path ([`Self::write_columnar_batch`]).
    #[cfg(feature = "columnar")]
    #[expect(
        clippy::too_many_arguments,
        reason = "the group's per-block index fields, plus the pages it copies"
    )]
    fn encode_columnar_batch_block(
        &mut self,
        batch: &crate::table::columnar::ColumnBatch,
        last_key: crate::UserKey,
        last_seqno: crate::SeqNo,
        seqno_bounds: Option<(u64, u64)>,
        item_count: usize,
        zone_block_min: Option<crate::UserKey>,
        carry: Option<&PageCarry<'_>>,
    ) -> crate::Result<()> {
        use crate::table::column_page::{PageDirectory, PageId, PageStamp};

        // Pages carry the data and go through the table's data codec; the
        // directory is a handful of offsets that no codec shrinks, and is kept
        // plain so a reader never needs the codec or its dictionary to find
        // out where the pages are. Both are encrypted and ECC-protected exactly
        // like data.
        check_group_column_count(batch)?;
        // A group copying another's pages keeps that group's tag, which every
        // copied page's stamp names.
        let group_tag = match carry {
            Some(carry) => carry.tag,
            None => self.next_group_tag(batch, &last_key),
        };
        let page_transform = self.data_transform(self.data_block_compression)?;
        let directory_transform = self.data_transform(crate::CompressionType::None)?;

        // The group's rows are cut into row pages shared by every column, and
        // each column gets one page per row page, all of one column's pages
        // together: a read of a column is then one run of adjacent pages, and
        // a read of a few rows reads only the row pages that hold them. Today
        // every column's encoding names a single part; a codec whose encoding
        // names several gets its pages per part here, and nothing downstream
        // changes, because pages are found by `(column_id, part, row_page)`
        // and never by position.
        // Pages are cut by the page size, groups by the group size, both by
        // the same count of the bytes each row adds, so a group the writer
        // closed at its size is one page when the page size is the group size. A
        // short tail page is kept rather than folded into the page before it:
        // folding costs the sparse scan more, since the enlarged page is read
        // whole whenever a match lands on it.
        // A group copying pages keeps the source's row pages one for one, since
        // a copied page's stamp names its row page by ordinal.
        let row_pages = if let Some(carry) = carry {
            carry.row_pages.clone()
        } else {
            let cuts = batch.row_page_cuts(self.columnar_page_size)?;
            // The directory counts the group's column pages in a `u16`, so a
            // page size that gives more row pages than it can list merges
            // adjacent ones: the grid stays complete and only pruning coarsens.
            // The column count was checked against the same field, so at
            // least one row page fits.
            let fit = usize::from(u16::MAX) / batch.columns.len().max(1);
            if cuts.len() <= fit {
                cuts
            } else {
                // Each merged page is a run of whole cuts, so its rows are at
                // most the group's `u32` row count.
                cuts.chunks(cuts.len().div_ceil(fit))
                    .map(|run| run.iter().sum::<u32>())
                    .collect()
            }
        };
        // Each page is encoded here, or, on a row page a carry names as
        // unchanged, the source's page as it lies on disk.
        let mut payloads = Vec::with_capacity(batch.columns.len() * row_pages.len());
        for (column, col) in batch.columns.iter().enumerate() {
            let id = PageId {
                column_id: col.column_id,
                part: 0,
            };
            let mut start = 0u32;
            for (index, &rows) in row_pages.iter().enumerate() {
                let row_page = u16::try_from(index).map_err(|_| {
                    crate::Error::InvalidHeader("columnar: more row pages than a group holds")
                })?;
                let end = start + rows;
                if let Some(copied) = carry.and_then(|carry| carry.page(column, index)) {
                    payloads.push((id, row_page, PagePayload::Copied(copied)));
                    start = end;
                    continue;
                }
                let piece;
                let page_rows = if row_pages.len() == 1 {
                    col
                } else {
                    piece = col.rows(batch.row_count, start, end)?;
                    &piece
                };
                let stamp = PageStamp {
                    group_tag,
                    id,
                    row_page,
                };
                payloads.push((
                    id,
                    row_page,
                    PagePayload::Encoded(page_rows.encode_page(
                        rows,
                        stamp,
                        self.column_encoding,
                    )?),
                ));
                start = end;
            }
        }
        let pages = payloads
            .iter()
            .map(|(id, row_page, payload)| {
                let page = match payload {
                    PagePayload::Encoded(payload) => PageOut::Prepared(Block::prepare_with_flags(
                        payload,
                        super::block::BlockIdentity {
                            table_id: self.table_id,
                            block_type: super::block::BlockType::ColumnPage,
                            dict_id: self.data_block_compression.dict_id(),
                            window_log: 0,
                        },
                        &page_transform,
                        0, // pages carry no per-KV checksum footer
                    )?),
                    PagePayload::Copied(copied) => PageOut::Copied(*copied),
                };
                Ok((*id, *row_page, page))
            })
            .collect::<crate::Result<Vec<_>>>()?;

        // Statistics zones for a group of several row pages; one row page's
        // zone is the group's zone-map entry. Every column's zones go into a
        // zone block of its own, which only a read that prunes on that column
        // reads. The key column's lies right after the directory, so a point
        // read takes both in one request; every other column's after the
        // pages.
        let (key_zones, other_zones) = batch.group_zones(&row_pages)?;
        let zone_block_payload = |zones: &crate::table::column_page::PageZones| {
            let mut payload = Vec::new();
            PageDirectory::encode_zone_block(group_tag, zones, &mut payload);
            payload
        };
        let head_payload = key_zones
            .columns()
            .first()
            .map(|&column_id| (column_id, zone_block_payload(&key_zones)));
        let zone_payloads: Vec<(u16, Vec<u8>)> = other_zones
            .columns()
            .iter()
            .map(|&column_id| {
                (
                    column_id,
                    zone_block_payload(&other_zones.only(|c| c == column_id)),
                )
            })
            .collect();
        let mut zone_blocks = head_payload
            .iter()
            .chain(&zone_payloads)
            .map(|(column_id, payload)| {
                let prepared = Block::prepare_with_flags(
                    payload,
                    super::block::BlockIdentity {
                        table_id: self.table_id,
                        block_type: super::block::BlockType::ColumnZones,
                        dict_id: 0,
                        window_log: 0,
                    },
                    &directory_transform,
                    0,
                )?;
                Ok((*column_id, prepared))
            })
            .collect::<crate::Result<Vec<_>>>()?
            .into_iter();
        let head_zone_block = if head_payload.is_some() {
            zone_blocks.next()
        } else {
            None
        };
        let zone_blocks: Vec<_> = zone_blocks.collect();
        let listing = |(column_id, prepared): &(u16, super::block::PreparedBlock<'_>)| {
            crate::table::column_page::ZoneBlock {
                column_id: *column_id,
                length: prepared.on_disk_len(self.ecc),
            }
        };
        let directory = PageDirectory::contiguous(
            batch.row_count,
            group_tag,
            row_pages,
            pages
                .iter()
                .map(|(id, row_page, page)| (*id, *row_page, page.on_disk_len(self.ecc))),
            head_zone_block.as_ref().map(listing),
            zone_blocks.iter().map(listing).collect(),
        )?;
        let mut directory_payload = Vec::new();
        directory.encode_into(&mut directory_payload);
        let directory_block = Block::prepare_with_flags(
            &directory_payload,
            super::block::BlockIdentity {
                table_id: self.table_id,
                block_type: super::block::BlockType::ColumnPageDirectory,
                dict_id: 0,
                window_log: 0,
            },
            &directory_transform,
            0,
        )?;

        // The group is written directory first, then its head zone block, then
        // its pages back to back, which is exactly the layout
        // `PageDirectory::contiguous` recorded, then the other zone blocks in
        // the order the directory lists them.
        let directory_len = directory_block.on_disk_len(self.ecc);
        let head_zones_len = directory.pages_start();
        let mut bytes_written = directory_len;
        let at = next_block_at(self.table_id, &self.file_writer);
        let mut uncompressed = u64::from(
            directory_block
                .write_to(&mut self.file_writer, at)?
                .uncompressed_length,
        );
        for page in head_zone_block
            .into_iter()
            .map(|(_, prepared)| PageOut::Prepared(prepared))
            .chain(pages.into_iter().map(|(_, _, page)| page))
            .chain(
                zone_blocks
                    .into_iter()
                    .map(|(_, prepared)| PageOut::Prepared(prepared)),
            )
        {
            let at = next_block_at(self.table_id, &self.file_writer);
            let header = match page {
                PageOut::Prepared(prepared) => prepared.write_to(&mut self.file_writer, at)?,
                PageOut::Copied(copied) => {
                    // The page's checksum is bound to the place it lay; the
                    // payload is written as it is.
                    let mut frame = copied.frame.to_vec();
                    super::block::Header::rebind_verified_frame(&mut frame, copied.at, at)?;
                    #[cfg(not(feature = "std"))]
                    use crate::io::Write;
                    #[cfg(feature = "std")]
                    use std::io::Write;
                    self.file_writer.write_all(&frame)?;
                    copied.header
                }
            };
            bytes_written = bytes_written
                .checked_add(header.on_disk_size_with(self.ecc))
                .ok_or(crate::Error::InvalidHeader(
                    "columnar: row group length overflows u32",
                ))?;
            uncompressed += u64::from(header.uncompressed_length);
        }
        self.group_tags.insert(group_tag);

        // Per-column zone-map stats for this row group, derived once from the
        // batch. Gated on the zone-map policy exactly like the row-block
        // synthetic entry (`zone_block_min` is `Some` iff the policy is on), so
        // a zone-map-off table writes no zone-map section. `verify_zone_map`
        // re-derives these from the decoded group to authenticate the section.
        let columnar_columns = zone_block_min.as_ref().map(|_| batch.zone_stats());
        self.register_written_extent(
            bytes_written,
            uncompressed,
            // Pages are decoded whole, so there is no inner-frame layout to
            // record for partial decode.
            Vec::new(),
            last_key,
            last_seqno,
            seqno_bounds,
            item_count,
            zone_block_min,
            columnar_columns,
            Some(row_group_ref(group_tag, directory_len, head_zones_len)?),
        )
    }

    /// The block transform this writer applies to data, under `compression`.
    ///
    /// Shared by every block of a columnar row group, which differ only in
    /// codec: its pages use the table's data codec and its directory none.
    #[cfg(feature = "columnar")]
    fn data_transform(
        &self,
        compression: crate::CompressionType,
    ) -> crate::Result<crate::table::block::BlockTransform<'_>> {
        let t = crate::table::block::BlockTransform::from_parts(
            compression,
            self.encryption.as_deref(),
            #[cfg(zstd_any)]
            self.zstd_dictionary.as_deref(),
        )?;
        #[cfg(zstd_any)]
        let t = t.with_two_pass_seed(self.zstd_two_pass_seed);
        Ok(if let Some(ecc) = self.ecc {
            t.with_ecc(ecc)
        } else {
            t
        })
    }

    /// Validates a columnar batch against the ingest contract without writing
    /// anything: the layout must be columnar, the batch shape valid, every per-row
    /// seqno `0`, and the keys strictly increasing within the batch. This lets the
    /// ingestion reject a bad batch eagerly even though the block emission is
    /// deferred (batches accumulate into one rowgroup). Cross-batch / cross-block
    /// ordering is the caller's responsibility (it tracks the last key); the check
    /// here starts fresh, so it is independent of any not-yet-flushed rows.
    ///
    /// # Errors
    ///
    /// Returns an error if the layout is not columnar, the batch shape is
    /// malformed, a row carries a non-zero seqno, or the keys are not strictly
    /// increasing within the batch.
    #[cfg(feature = "columnar")]
    pub(crate) fn validate_columnar_batch(
        &self,
        batch: &crate::table::columnar::ColumnBatch,
        comparator: &crate::SharedComparator,
    ) -> crate::Result<()> {
        if !self.use_columnar {
            return Err(crate::Error::FeatureUnsupported(
                "columnar batch ingest requires the columnar layout",
            ));
        }
        check_group_column_count(batch)?;
        // Validate the layout, framing, seqno-zero, and intra-batch key order
        // without decoding every row into an `InternalValue`: the batch is
        // re-validated and fully decoded once at flush (on the accumulated
        // rowgroup), so an eager full decode here would deserialise every
        // submitted row twice.
        crate::table::columnar::validate_columnar_ingest_batch(batch, comparator)
    }

    /// Writes a consumer-provided [`ColumnBatch`](crate::table::columnar::ColumnBatch)
    /// as columnar row groups, storing its value sub-columns directly instead
    /// of re-transposing the intrinsic value. The batch must carry the three
    /// intrinsic columns plus one or more value sub-columns; the columnar layout
    /// must be enabled. Keys must be strictly increasing by `comparator` (within
    /// the batch and across successive batches / row writes on this writer), and
    /// every per-row seqno must be `0` (the ingestion assigns one sequence
    /// number). Each call appends one row group, or several of the group size
    /// for a batch past it.
    ///
    /// Returns the batch's last user key (or `None` for an empty batch) so a
    /// caller can track the ingest ordering across batches.
    ///
    /// # Errors
    ///
    /// Returns an error if the batch shape is malformed, the layout is not
    /// columnar, the keys are not strictly increasing, a row carries a non-zero
    /// seqno, or the block write fails.
    #[cfg(feature = "columnar")]
    pub(crate) fn write_columnar_batch(
        &mut self,
        batch: &crate::table::columnar::ColumnBatch,
        comparator: &crate::SharedComparator,
    ) -> crate::Result<Option<crate::UserKey>> {
        self.write_columnar_block_inner(batch, comparator, true)
    }

    /// Writes a pre-built [`ColumnBatch`](crate::table::columnar::ColumnBatch) as a
    /// single columnar block, storing its value sub-columns AND its per-row seqnos
    /// **verbatim** — unlike [`Self::write_columnar_batch`], which is the
    /// bulk-ingest path and requires seqno `0` (assigned at finish). Salvage uses
    /// this to re-emit a recovered columnar block under its original sequence
    /// numbers, so the recovered copy keeps both its per-field sub-columns and its
    /// MVCC versions. The table must be written with `global_seqno` `0` so the
    /// stored seqnos are the effective ones.
    ///
    /// Returns the batch's last user key (or `None` for an empty batch).
    ///
    /// # Errors
    ///
    /// As [`Self::write_columnar_batch`], except a non-zero per-row seqno is
    /// accepted (it is stored as the row's sequence number).
    #[cfg(feature = "columnar")]
    pub(crate) fn write_columnar_block_verbatim(
        &mut self,
        batch: &crate::table::columnar::ColumnBatch,
        comparator: &crate::SharedComparator,
    ) -> crate::Result<Option<crate::UserKey>> {
        self.write_columnar_block_inner(batch, comparator, false)
    }

    /// Shared body for the two columnar-block writers. `require_zero_seqno`
    /// enforces the bulk-ingest contract (every row seqno `0`); the verbatim
    /// salvage path passes `false` to store the batch's own seqnos.
    #[cfg(feature = "columnar")]
    fn write_columnar_block_inner(
        &mut self,
        batch: &crate::table::columnar::ColumnBatch,
        comparator: &crate::SharedComparator,
        require_zero_seqno: bool,
    ) -> crate::Result<Option<crate::UserKey>> {
        // A columnar batch only makes sense (and only reads back correctly) when
        // the table is marked columnar; reject a row-mode writer rather than
        // writing a columnar block under a row descriptor.
        if !self.use_columnar {
            return Err(crate::Error::FeatureUnsupported(
                "columnar batch ingest requires the columnar layout",
            ));
        }
        // Validate the batch shape and obtain per-row keys / seqnos / value-types.
        // The framed values feed the shape accounting; the block stores the
        // consumer's sub-columns (not a re-transpose).
        // An ingested batch names only field ids: one the engine keeps would
        // make the group read as rows written as cells.
        if require_zero_seqno {
            let value_cols = batch.columns.get(3..).unwrap_or_default();
            crate::table::columnar::check_ingested_field_ids(value_cols)?;
        }
        let entries = crate::table::columnar::column_batch_to_entries(batch)?;

        // Ingest contract. Unsorted keys would corrupt the sorted block index /
        // zone map; a non-zero seqno would read back shifted by the table's
        // assigned sequence number. Reject both before writing anything.
        if require_zero_seqno && entries.iter().any(|e| e.key.seqno != 0) {
            return Err(crate::Error::FeatureUnsupported(
                "columnar batch ingest requires every row seqno to be 0 (the ingestion assigns the sequence number)",
            ));
        }
        // Bulk columnar ingest (`require_zero_seqno`) is a load of strictly-unique,
        // ascending keys. The verbatim salvage re-emit path (`require_zero_seqno ==
        // false`) can carry repeated user keys across MVCC versions of one key, so
        // it must NOT enforce strict uniqueness — but the block must still be in
        // valid INTERNAL-key order (`account_direct_block` trusts it): salvage
        // feeds this path with entries decoded from an untrusted block, and an
        // unvalidated out-of-order block would register a wrong last-key in the
        // index. Check within the batch and against the writer's prior key before
        // any state mutation.
        if require_zero_seqno {
            let mut prev = Self::last_key_in(&self.chunk, self.spilled_key.as_ref());
            for e in &entries {
                if let Some(p) = prev
                    && comparator.compare(p.as_ref(), e.key.user_key.as_ref())
                        != core::cmp::Ordering::Less
                {
                    return Err(crate::Error::InvalidHeader(
                        "columnar batch ingest requires strictly increasing keys",
                    ));
                }
                prev = Some(&e.key.user_key);
            }
        } else {
            self.validate_direct_block_order(&entries, comparator)?;
        }
        // An ingested batch is stored as the caller split it into fields; a
        // verbatim re-emit stores what its source did, which the rewrite set
        // up front.
        self.claim_value_layout(if require_zero_seqno {
            crate::table::meta::ValueLayout::Split
        } else {
            self.value_layout
                .unwrap_or(crate::table::meta::ValueLayout::Split)
        })?;

        // A batch past the group size is written as groups of that size, cut
        // by the bytes each row adds as flushed rows are: a read bounds what a
        // group decodes to by what a group of that size holds, so a group
        // taking a whole oversized batch would be refused. The whole batch was
        // checked above, so a bad batch writes nothing.
        let groups = batch.row_page_cuts(self.row_group_size)?;
        if groups.len() <= 1 {
            return self.write_columnar_group(batch, &entries);
        }
        let mut last_key = None;
        let mut start = 0u32;
        for rows in groups {
            // The cuts sum to the batch's rows, a u32.
            let end = start + rows;
            let group = batch.rows(start, end)?;
            let group_entries =
                entries
                    .get(start as usize..end as usize)
                    .ok_or(crate::Error::InvalidHeader(
                        "columnar: row range outside the batch",
                    ))?;
            last_key = self
                .write_columnar_group(&group, group_entries)?
                .or(last_key);
            start = end;
        }
        Ok(last_key)
    }

    /// Writes one row group of a checked columnar batch, `entries` its rows.
    #[cfg(feature = "columnar")]
    fn write_columnar_group(
        &mut self,
        batch: &crate::table::columnar::ColumnBatch,
        entries: &[InternalValue],
    ) -> crate::Result<Option<crate::UserKey>> {
        let Some(inputs) = self.account_direct_block(entries)? else {
            return Ok(None); // empty batch: no block
        };

        self.encode_columnar_batch_block(
            batch,
            inputs.last_key.clone(),
            inputs.last_seqno,
            inputs.seqno_bounds,
            inputs.item_count,
            inputs.zone_block_min,
            None,
        )?;
        // This block's keys were recorded with the current locator ordinal;
        // advance it so a following batch's keys belong to the next block (the
        // chunk-based path does this in `spill_block`).
        self.locator_block_done();
        Ok(Some(inputs.last_key))
    }

    /// Encodes `chunk` into `buf`, returning the `block_flags` bits the transform
    /// can't derive (the per-KV checksum-footer bit). Associated fn (no `&self`)
    /// so callers can pass disjoint field borrows (`&self.chunk`,
    /// `&mut self.block_buffer`) without aliasing.
    fn encode_chunk_into(
        chunk: &[InternalValue],
        key_hashes: Option<&[u64]>,
        restart_interval: u8,
        hash_ratio: f32,
        kv_emit: Option<crate::runtime_config::ChecksumAlgorithm>,
        buf: &mut Vec<u8>,
    ) -> crate::Result<u8> {
        if let Some(algo) = kv_emit {
            // One logical-content digest per entry, in scan order.
            let mut digests = Vec::with_capacity(chunk.len());
            for item in chunk {
                let d = crate::table::block::kv_checksum::kv_digest(item, algo)
                    .ok_or(crate::Error::FeatureUnsupported("kv-checksum-algorithm"))?;
                digests.push(d);
            }
            DataBlock::encode_kv_checked_hashed_into(
                buf,
                chunk,
                key_hashes,
                &digests,
                algo,
                restart_interval,
                hash_ratio,
            )?;
            Ok(crate::table::block::header::block_flags::KV_CHECKSUM_FOOTER)
        } else {
            DataBlock::encode_hashed_into(buf, chunk, key_hashes, restart_interval, hash_ratio)?;
            Ok(0)
        }
    }

    /// `key_hashes`, the user-key hash of each item of `chunk`, when the
    /// block's hash index buckets them by it (and so they were kept).
    fn key_hashes_of<'a>(chunk: &[InternalValue], key_hashes: &'a [u64]) -> Option<&'a [u64]> {
        debug_assert!(key_hashes.is_empty() || key_hashes.len() == chunk.len());
        (!key_hashes.is_empty()).then_some(key_hashes)
    }

    /// Records a just-written data block: index entry (with optional seqno
    /// bounds), metadata counters, back link, and the running last key. Single
    /// source of truth for both the serial and parallel write paths.
    #[expect(
        clippy::too_many_arguments,
        reason = "cohesive per-written-block fields; a param struct adds indirection without clarity"
    )]
    fn register_written_block(
        &mut self,
        header: crate::table::block::Header,
        layout: Vec<u32>,
        last_key: UserKey,
        last_seqno: crate::SeqNo,
        seqno_bounds: Option<(u64, u64)>,
        item_count: usize,
        zone_block_min: Option<UserKey>,
        columnar_columns: Option<Vec<crate::table::zone_map::ColumnStats>>,
    ) -> crate::Result<()> {
        // Size the block-handle with the scheme this writer actually wrote
        // the parity under (NOT the fixed RS(4,2) `on_disk_size` assumes),
        // or the handle over-reads on a non-default scheme.
        let bytes_written = header.on_disk_size_with(self.ecc);
        self.register_written_extent(
            bytes_written,
            u64::from(header.uncompressed_length),
            layout,
            last_key,
            last_seqno,
            seqno_bounds,
            item_count,
            zone_block_min,
            columnar_columns,
            None,
        )
    }

    /// Registers one index entry covering `bytes_written` bytes that start at
    /// the current file position.
    ///
    /// Usually that is one block ([`Self::register_written_block`]). A
    /// columnar row group is several — its page directory and its pages — and
    /// is still ONE entry: the index names row groups, so `block_id` keeps its
    /// meaning and every section keyed by a data block's file offset keeps
    /// working, since the group starts exactly where its directory does. The
    /// entry carries the group's tag, which its directory must repeat, and its
    /// directory's length, which its directory's header must repeat.
    #[expect(
        clippy::too_many_arguments,
        reason = "cohesive per-written-block fields; a param struct adds indirection without clarity"
    )]
    fn register_written_extent(
        &mut self,
        bytes_written: u32,
        uncompressed_length: u64,
        layout: Vec<u32>,
        last_key: UserKey,
        last_seqno: crate::SeqNo,
        seqno_bounds: Option<(u64, u64)>,
        item_count: usize,
        zone_block_min: Option<UserKey>,
        columnar_columns: Option<Vec<crate::table::zone_map::ColumnStats>>,
        row_group: Option<crate::table::index_block::RowGroupRef>,
    ) -> crate::Result<()> {
        self.meta.uncompressed_size += uncompressed_length;

        // Record the inner zstd-block layout keyed by this block's file offset
        // (`meta.file_pos`, captured before the increment below). Only
        // multi-inner-block data blocks carry a non-empty layout, so single
        // (default 4 KiB) blocks add nothing.
        if !layout.is_empty() {
            self.layout_bytes += (layout.capacity() * core::mem::size_of::<u32>()) as u64;
            self.block_layouts.push((self.meta.file_pos, layout));
        }

        let handle = KeyedBlockHandle::new(
            last_key.clone(),
            last_seqno,
            BlockHandle::new(self.meta.file_pos, bytes_written).with_row_group(row_group),
        );
        // Seqno bounds go into the parallel `seqno_bounds` section keyed by this
        // block's file offset, NOT inline in the index entry: keeping them out of
        // the index keeps point-read index probes at their legacy cost while the
        // seqno-scoped scan loads the section.
        if let Some((seqno_min, seqno_max)) = seqno_bounds {
            self.seqno_bounds_section
                .push((self.meta.file_pos, (seqno_min, seqno_max)));
        }
        // Zone-map entry for this block (parallel section keyed by file offset,
        // like seqno_bounds). A columnar block records one entry per stored
        // column (`columnar_columns`, pre-derived from the batch); a row block
        // records one synthetic column with the block's key range. The row count
        // never approaches `u32::MAX` (a data block holds at most a few thousand
        // entries), so the cap is defensive. `zone_block_min` is `Some` exactly
        // when the zone-map policy is on, gating both variants alike.
        if let Some(min_key) = zone_block_min {
            let columns = if let Some(cols) = columnar_columns {
                cols
            } else {
                let row_count = u32::try_from(item_count).unwrap_or(u32::MAX);
                alloc::vec![crate::table::zone_map::ColumnStats {
                    column_id: 0,
                    type_tag: 0,
                    codec_id: 0,
                    null_count: 0,
                    row_count,
                    min: min_key.to_vec(),
                    max: last_key.to_vec(),
                }]
            };
            // A valid columnar batch always has at least the user-key column, so
            // this is non-empty in practice; guard defensively so an empty stats
            // vector never registers a column-less zone-map entry.
            if !columns.is_empty() {
                // The entry's column vector and each column's bounds.
                let bounds: usize = columns
                    .iter()
                    .map(|c| c.min.capacity() + c.max.capacity())
                    .sum();
                self.zone_map_bytes += (columns.capacity()
                    * core::mem::size_of::<crate::table::zone_map::ColumnStats>()
                    + bounds) as u64;
                self.zone_map_section.push((self.meta.file_pos, columns));
            }
        }
        self.index_writer.register_data_block(handle)?;

        self.meta.file_pos += u64::from(bytes_written);
        self.meta.item_count += item_count;
        self.meta.data_block_count += 1;
        self.writeback_if_due()?;

        self.prev_pos.0 = self.prev_pos.1;
        self.prev_pos.1 += u64::from(bytes_written);

        self.meta.last_key = Some(last_key);
        if self.meta_base_len.is_none() {
            self.meta_base_len = Some(self.meta_payload_len_without_keys()?);
        }
        // The table rotates on the estimates at its next key, which must count
        // this block.
        self.refresh_state_estimates();
        Ok(())
    }

    /// Writes out every block the writer still holds, so the file position is
    /// where a direct block lands: the row chunk buffered by prior `write()`
    /// calls (its block and locator ordinal registered first, keeping block and
    /// index order), and every block still being compressed on worker threads,
    /// which would otherwise be written over the direct block's bytes.
    /// `spill_block` only drains down to the parallel cap, so the full drain is
    /// separate. A no-op when nothing is pending. Once it returns, the table's
    /// key range reaches its last key written.
    pub(crate) fn settle_pending_blocks(&mut self) -> crate::Result<()> {
        self.spill_block()?;
        #[cfg(feature = "std")]
        while self.parallel.as_ref().map_or(0, BlockCompressor::pending) > 0 {
            self.drain_one_parallel()?;
        }
        Ok(())
    }

    /// Validates key order against prior writes, flushes any pending row chunk to
    /// keep block order, and folds the per-row shape / seqno / filter / locator
    /// accounting for a block written *directly* (not via the row chunk) —
    /// returning the inputs [`Self::register_written_block`] needs. Shared by the
    /// columnar-ingest block path and the verbatim copy-through salvage path so the
    /// bookkeeping has a single source of truth. Does NOT write or register the
    /// block: the caller emits it (encode, or a verbatim byte-copy). Returns `None`
    /// for an empty entry set (no block).
    ///
    /// `entries` must already be in valid block order. Equal user keys are
    /// permitted (a recovered data block can carry several MVCC versions of one
    /// key); the columnar-ingest contract of strictly-unique keys is enforced by
    /// the caller ([`Self::write_columnar_block_inner`]), not here.
    ///
    /// # Errors
    ///
    /// Returns an error if flushing the pending row chunk or registering a filter
    /// key fails.
    fn account_direct_block(
        &mut self,
        entries: &[InternalValue],
    ) -> crate::Result<Option<DirectBlockInputs>> {
        let Some(last) = entries.last() else {
            return Ok(None);
        };
        // Owned bytes for the same reason as in `spill_block`: the index keeps
        // this key until the table ends.
        let last_key = UserKey::new(&last.key.user_key);
        let last_seqno = last.key.seqno;
        let item_count = entries.len();
        let seqno_bounds = self.use_seqno_in_index.then(|| {
            entries.iter().fold((u64::MAX, u64::MIN), |(mn, mx), e| {
                (mn.min(e.key.seqno), mx.max(e.key.seqno))
            })
        });
        let zone_block_min = self
            .use_zone_map
            .then(|| entries.first().map(|e| e.key.user_key.clone()))
            .flatten();

        // The validation above ran first, so an invalid batch never forces a
        // spill.
        self.settle_pending_blocks()?;

        // Per-row shape / seqno / key accounting, mirroring `write()` minus the
        // chunk push (the direct block holds all rows already). The locator
        // records each new key's position within this single block, the same as
        // the flush path does for a transposed columnar block.
        let mut current_key = self.spilled_key.as_ref();
        for (row, e) in entries.iter().enumerate() {
            let user_key = &e.key.user_key;
            self.meta.sum_user_key_bytes += user_key.len() as u64;
            self.meta.sum_value_bytes += e.value.len() as u64;
            if e.is_tombstone() {
                self.meta.tombstone_count += 1;
            }
            if e.key.value_type == ValueType::WeakTombstone {
                self.meta.weak_tombstone_count += 1;
            }
            if e.key.value_type.is_put()
                && let Some(prev_key) = &self.previous_weak_tombstone_key
                && crate::comparator::same_user_key(prev_key, user_key)
            {
                self.meta.weak_tombstone_reclaimable_count += 1;
            }
            if !current_key.is_some_and(|c| crate::comparator::same_user_key(c, user_key)) {
                self.meta.key_count += 1;
                if self.hashes_keys {
                    // The filter and the locator key the same bytes: one hash.
                    // It is the current key's for the hash index too, which a
                    // later row write of another version of this key buckets by.
                    let key_hash = crate::hash::hash64(user_key);
                    self.current_key_hash = key_hash;
                    if self.filtered {
                        self.filter_hashes +=
                            self.filter_writer.register_key(user_key, key_hash)? as u64;
                    }
                    if let Some(spec) = self.locator {
                        // `row` is this key's item index within the forming
                        // block; `locator_block_id` is the block's ordinal.
                        let pos = row as u64;
                        let slot = match spec.precision {
                            crate::config::LocatorPrecision::Block => 0,
                            crate::config::LocatorPrecision::Restart => {
                                pos / u64::from(self.data_block_restart_interval)
                            }
                            crate::config::LocatorPrecision::Entry => pos,
                        };
                        self.locators.push((key_hash, self.locator_block_id, slot));
                    }
                }
            }
            current_key = Some(user_key);
            if self.meta.first_key.is_none() {
                self.meta.first_key = Some(user_key.clone());
            }
            self.previous_type = Some(e.key.value_type);
            self.previous_weak_tombstone_key =
                (e.key.value_type == ValueType::WeakTombstone).then(|| user_key.clone());
            self.current_key_seqno = Some(e.key.seqno);
            self.meta.lowest_seqno = self.meta.lowest_seqno.min(e.key.seqno);
            self.meta.highest_seqno = self.meta.highest_seqno.max(e.key.seqno);
            self.meta.highest_kv_seqno = self.meta.highest_kv_seqno.max(e.key.seqno);
        }
        // The block's own rows are the current key until it ends.
        self.spilled_key = Some(last_key.clone());

        Ok(Some(DirectBlockInputs {
            last_key,
            last_seqno,
            seqno_bounds,
            item_count,
            zone_block_min,
        }))
    }

    /// Appends a data block to the output by copying its raw on-disk bytes
    /// **verbatim** — no decode + re-encode + re-compress + re-ECC — while folding
    /// the same per-row accounting a freshly-encoded block gets. Salvage uses this
    /// to fast-path blocks that read back cleanly (checksum passed without ECC
    /// recovery): the original framing, compression, encryption, ECC parity, and
    /// (for a columnar block) the PAX sub-columns / zone map are preserved
    /// bit-for-bit at the block's new file offset.
    ///
    /// `raw` MUST be the exact `header.on_disk_size_with(self.ecc)` bytes the
    /// source wrote for this block (header + payload + ECC trailer), read from
    /// `source`, whose binding the copy re-stamps for its new place; `layout` is
    /// its inner-zstd block layout (empty for a single-inner block). The inner
    /// layout offsets are decompressed-space and block-relative, so they stay
    /// valid at the new file offset. `entries` are the block's decoded rows, used
    /// only for filter / key-count / seqno / zone accounting — never
    /// re-serialized. The writer's ECC scheme must match the one the source block
    /// was written under (the salvage path mirrors it), or the recorded block size
    /// diverges from the copied bytes.
    ///
    /// Returns the block's last user key (or `None` for an empty entry set).
    /// `entries` may contain repeated user keys (MVCC versions of one key); they
    /// are accounted, not re-ordered, and the raw block already holds them in
    /// valid order.
    ///
    /// # Errors
    ///
    /// Returns an error if `raw`'s length disagrees with the header's on-disk size
    /// under this writer's ECC scheme, if the entries are out of internal-key
    /// order ([`Self::validate_direct_block_order`]), or the write fails.
    pub(crate) fn append_verbatim_data_block(
        &mut self,
        (raw, source): (&[u8], VerbatimSource),
        header: crate::table::block::Header,
        layout: alloc::vec::Vec<u32>,
        entries: &[InternalValue],
        // Per-column zone-map stats for a COLUMNAR block being copied verbatim
        // (`Some`, pre-derived from its decoded `ColumnBatch`); `None` for a row
        // block, where `register_written_block` synthesizes the whole-block key
        // range. A columnar block copied with `None` would record the row-block
        // synthetic column-0 stat, which `verify_zone_map` then rejects on the
        // columnar table.
        columnar_columns: Option<Vec<crate::table::zone_map::ColumnStats>>,
        comparator: &crate::SharedComparator,
    ) -> crate::Result<Option<crate::UserKey>> {
        // The copied bytes ARE the block; their length must equal what the index
        // entry will record (`register_written_block` sizes the handle and advances
        // `file_pos` by `header.on_disk_size_with(self.ecc)`). A mismatch means the
        // writer's ECC scheme was not mirrored from the source — refuse rather than
        // record a handle that over- or under-reads on reopen.
        let expected = header.on_disk_size_with(self.ecc) as usize;
        if raw.len() != expected {
            return Err(crate::Error::InvalidHeader(
                "verbatim block copy: raw length disagrees with the header on-disk size",
            ));
        }
        self.append_verbatim_extent(
            (raw, source),
            u64::from(header.uncompressed_length),
            layout,
            entries,
            columnar_columns,
            None,
            comparator,
        )
    }

    /// Counts each new key of `entries`, in order, against the partition of a
    /// prebuilt filter that covers it, and refuses a key none of them covers,
    /// before anything is written. Without a prebuilt filter it does nothing.
    /// A column batch is written whole, not entry by entry, so only it needs
    /// the count up front.
    #[cfg(feature = "columnar")]
    pub(crate) fn count_prebuilt_keys(
        &mut self,
        entries: &[InternalValue],
        comparator: &crate::SharedComparator,
    ) -> crate::Result<()> {
        if let Some(prebuilt) = self.prebuilt_filter.as_mut() {
            let mut previous =
                Self::last_key_in(&self.chunk, self.spilled_key.as_ref()).map(|k| &**k);
            for entry in entries {
                let key = &*entry.key.user_key;
                if previous.is_some_and(|p| crate::comparator::same_user_key(p, key)) {
                    continue;
                }
                prebuilt.count_key(key, &mut self.prebuilt_partition, comparator.as_ref())?;
                previous = Some(key);
            }
        }
        Ok(())
    }

    /// Appends a columnar row group by copying its raw on-disk bytes
    /// **verbatim** — directory and every page, as the source wrote them —
    /// under one index entry, while folding the same per-row accounting a
    /// freshly-encoded group gets.
    ///
    /// The row-group counterpart to [`Self::append_verbatim_data_block`]. The
    /// group's page offsets are relative to its own directory, so the copy
    /// stays addressable at its new file offset without rewriting a byte.
    ///
    /// `raw` MUST be every block of one group, back to back, each already
    /// proved verbatim-safe by the salvage walk; `uncompressed_length` is the
    /// sum over them, and `group_tag` and `head_zones_len` are the tag and
    /// head zone block length its directory carries.
    ///
    /// # Errors
    ///
    /// As [`Self::append_verbatim_data_block`], minus the single-header length
    /// check: a group has no one header to check against. Also
    /// [`crate::Error::InvalidHeader`] when [`Self::accepts_group_tag`] refuses
    /// `group_tag`, which a caller is expected to have asked first.
    #[cfg(feature = "columnar")]
    pub(crate) fn append_verbatim_row_group(
        &mut self,
        (raw, source): (&[u8], VerbatimSource),
        uncompressed_length: u64,
        (group_tag, head_zones_len): (u64, u32),
        entries: &[InternalValue],
        columnar_columns: Option<Vec<crate::table::zone_map::ColumnStats>>,
        comparator: &crate::SharedComparator,
    ) -> crate::Result<Option<crate::UserKey>> {
        if !self.accepts_group_tag(group_tag) {
            return Err(crate::Error::InvalidHeader(
                "columnar: a copied row group's tag does not follow the table's last",
            ));
        }
        // The group opens with its directory, whose header gives its length.
        let directory_len = carried_header(raw)?.on_disk_size_with(self.ecc);
        let first_key = self.append_verbatim_extent(
            (raw, source),
            uncompressed_length,
            Vec::new(),
            entries,
            columnar_columns,
            Some(row_group_ref(group_tag, directory_len, head_zones_len)?),
            comparator,
        )?;
        self.group_tags.insert(group_tag);
        Ok(first_key)
    }

    /// Appends `raw`, a row group of another table copied whole, whose rows are
    /// `rows` and whose values are stored as `layout`, under the tag and
    /// lengths `group` records for it; `columns` are its per-column zone-map
    /// statistics, as its own table recorded them.
    ///
    /// Returns `false`, writing nothing, when this table cannot take the group
    /// as it is: a row-major table, one whose values are stored the other way,
    /// one already holding the group's tag, one keeping a zone map the group
    /// has no statistics for, or one the group's rows do not continue in
    /// internal-key order. The caller then writes the rows instead.
    ///
    /// The group's transform must be this writer's: its pages compressed with
    /// this writer's codec and dictionary, and neither encrypted nor
    /// ECC-protected, which a caller establishes before offering it.
    ///
    /// # Errors
    ///
    /// As [`Self::append_verbatim_row_group`], plus
    /// [`crate::Error::InvalidHeader`] when `raw` does not tile into whole
    /// blocks.
    #[cfg(feature = "columnar")]
    pub(crate) fn append_carried_row_group(
        &mut self,
        (raw, source): (&[u8], VerbatimSource),
        group: crate::table::index_block::RowGroupRef,
        layout: crate::table::meta::ValueLayout,
        rows: &[InternalValue],
        columns: Option<Vec<crate::table::zone_map::ColumnStats>>,
        comparator: &crate::SharedComparator,
    ) -> crate::Result<bool> {
        // A table keeping a zone map needs the group's per-column statistics,
        // which only the source's zone map holds for bytes copied unread.
        if !self.use_columnar
            || self.value_layout.is_some_and(|fixed| fixed != layout)
            || !self.accepts_group_tag(group.tag.get())
            || (self.use_zone_map && columns.is_none())
        {
            return Ok(false);
        }
        // Two inputs may hold one key at one seqno (a write repeated at the
        // seqno it had), and the merge emits both; a table that took them so
        // holds a group a copy refuses as out of order. Its rows go through
        // the ordinary path, which keeps such a tie as the merge emitted it.
        if self.validate_direct_block_order(rows, comparator).is_err() {
            return Ok(false);
        }
        // What the group decodes to, summed over its blocks as an encoded
        // group sums it.
        let mut uncompressed = 0u64;
        let mut at = 0usize;
        while let Some(frame) = raw.get(at..).filter(|frame| !frame.is_empty()) {
            let header = carried_header(frame)?;
            let len = header.on_disk_size_with(self.ecc) as usize;
            if len == 0 || len > frame.len() {
                return Err(crate::Error::InvalidHeader(
                    "carried row group does not tile into blocks",
                ));
            }
            uncompressed += u64::from(header.uncompressed_length);
            at += len;
        }
        self.append_verbatim_row_group(
            (raw, source),
            uncompressed,
            (group.tag.get(), group.head_zones_len),
            rows,
            columns,
            comparator,
        )?;
        self.value_layout = Some(layout);
        Ok(true)
    }

    /// Appends `frame`, a data block of another table copied whole, whose rows
    /// are `rows`, read from `source`, a table written with `shape`.
    ///
    /// Returns `false`, writing nothing, when this table cannot take the block
    /// as it is: a columnar table; one with a prebuilt filter; one whose
    /// restart interval differs, since the locator names a key's restart head
    /// by the table's interval; one whose per-KV footer differs, since a
    /// table's reader strips the footer by its own descriptor; one whose hash
    /// index ratio differs, which the table records for all its blocks; a frame that is
    /// no data block; or rows that do not continue the table in internal-key
    /// order. The caller then writes the rows instead.
    ///
    /// The block's transform must be this writer's: compressed with this
    /// writer's codec and dictionary, and neither encrypted nor ECC-protected,
    /// which a caller establishes before offering it. A block split into
    /// several inner frames is not offered.
    ///
    /// # Errors
    ///
    /// As [`Self::append_verbatim_data_block`].
    pub(crate) fn append_carried_data_block(
        &mut self,
        (frame, source): (&[u8], VerbatimSource),
        shape: CarriedBlockShape,
        rows: &[InternalValue],
        comparator: &crate::SharedComparator,
    ) -> crate::Result<bool> {
        let footer = self.kv_checksum.and_then(|(policy, algo)| {
            policy
                .applies(self.initial_level, self.table_id)
                .then_some(algo)
        });
        if self.use_columnar
            || self.prebuilt_filter.is_some()
            || self.data_block_restart_interval != shape.restart_interval
            || footer != shape.kv_checksum
            || self.data_block_hash_ratio.to_bits() != shape.hash_ratio_bits
        {
            return Ok(false);
        }
        let Ok(header) = carried_header(frame) else {
            return Ok(false);
        };
        if header.block_type != crate::table::block::BlockType::Data
            || header.on_disk_size_with(self.ecc) as usize != frame.len()
        {
            return Ok(false);
        }
        // Two inputs may hold one key at one seqno, and the merge emits both;
        // rows written one by one keep such a tie as the merge emitted it.
        if self.validate_direct_block_order(rows, comparator).is_err() {
            return Ok(false);
        }
        self.append_verbatim_data_block(
            (frame, source),
            header,
            Vec::new(),
            rows,
            None,
            comparator,
        )?;
        Ok(true)
    }

    /// Whether a row group tagged `group_tag` may be copied in next. A copy
    /// keeps the tag its pages were stamped with, so no group of the table may
    /// have taken it already; a group encoded here takes a free one instead.
    ///
    /// A salvage or a compaction copying groups finds their tags free unless
    /// another group of the output drew the same tag, by chance: every tag is
    /// a hash of its own group's content ([`Self::next_group_tag`]), so a group
    /// re-encoded here and a group copied in collide only when their hashes do,
    /// and the one that clashes is then encoded rather than copied.
    #[cfg(feature = "columnar")]
    #[must_use]
    pub(crate) fn accepts_group_tag(&self, group_tag: u64) -> bool {
        group_tag != 0 && !self.group_tags.contains(&group_tag)
    }

    /// The tag for the row group `batch`, ending on `last_key`, encoded here: a
    /// hash of the table's id, its age and the group's content, in the lower
    /// half of `u64` and never zero, or the first tag above it no group of the
    /// table took.
    ///
    /// Deriving every tag from its own group is what makes a page's stamp name
    /// its table as well as its group: a page's values are apart from their
    /// keys, so a page taken for another table's would serve that table's
    /// value under this table's key, and two tables' tags meet only by chance
    /// unless their groups hold the same rows. Nothing outside the table enters
    /// the hash, so the same writes give the same tags wherever the folder is,
    /// and tables of one id and one age, as two trees on one fixed clock write,
    /// part wherever their content does.
    #[cfg(feature = "columnar")]
    fn next_group_tag(
        &mut self,
        batch: &crate::table::columnar::ColumnBatch,
        last_key: &[u8],
    ) -> u64 {
        let age = *self.group_tag_age.get_or_insert_with(|| {
            self.created_at
                .unwrap_or_else(|| crate::time::now_on(self.clock.as_ref()).as_nanos())
        });
        let mut seed = xxhash_rust::xxh3::Xxh3::default();
        seed.update(&self.table_id.to_le_bytes());
        seed.update(&age.to_le_bytes());
        seed.update(&batch.row_count.to_le_bytes());
        for column in &batch.columns {
            let (type_tag, width) = column.type_tag.to_wire();
            seed.update(&column.column_id.to_le_bytes());
            seed.update(&[type_tag, width, u8::from(column.validity.is_some())]);
            if let Some(validity) = &column.validity {
                seed.update(validity);
            }
            seed.update(&(column.data.len() as u64).to_le_bytes());
            seed.update(&column.data);
        }
        seed.update(last_key);
        // Never zero: the index names a group by a non-zero tag.
        let mut tag = (seed.digest() >> 1).max(1);
        // The probe starts below 2^63 and steps once per tag a group of this
        // table took, far fewer than 2^63, so it cannot overflow.
        while self.group_tags.contains(&tag) {
            tag += 1;
        }
        tag
    }

    /// `raw`, the blocks of one extent copied from `source`, with each block's
    /// stored checksum moved from its place there to the place it lands here.
    /// The payloads are not touched.
    ///
    /// # Errors
    ///
    /// Returns [`crate::Error::InvalidHeader`] when `raw` does not tile into
    /// whole blocks under this writer's ECC scheme.
    fn restamp_extent(&self, raw: &[u8], source: VerbatimSource) -> crate::Result<Vec<u8>> {
        use crate::table::block::{ChecksumAt, Header};

        let dest = self.file_writer.get_ref().position();
        let mut out = raw.to_vec();
        let mut at = 0usize;
        while at < out.len() {
            let frame = out.get_mut(at..).ok_or(crate::Error::InvalidHeader(
                "verbatim copy: block past the extent",
            ))?;
            let len = carried_header(frame)?.on_disk_size_with(self.ecc) as usize;
            if len == 0 || len > frame.len() {
                return Err(crate::Error::InvalidHeader(
                    "verbatim copy: extent does not tile into blocks",
                ));
            }
            let relative = at as u64;
            Header::rebind_verified_frame(
                frame,
                ChecksumAt::table(source.table_id, source.offset + relative),
                ChecksumAt::table(self.table_id, dest + relative),
            )?;
            at += len;
        }
        Ok(out)
    }

    /// Shared body of the verbatim copies: validates the entries' order,
    /// appends `raw` to the data region, and registers it as one index entry,
    /// naming the row group `row_group` when the copy is one.
    #[expect(
        clippy::too_many_arguments,
        reason = "cohesive per-copy fields, forwarded to `register_written_extent`"
    )]
    fn append_verbatim_extent(
        &mut self,
        (raw, source): (&[u8], VerbatimSource),
        uncompressed_length: u64,
        layout: alloc::vec::Vec<u32>,
        entries: &[InternalValue],
        columnar_columns: Option<Vec<crate::table::zone_map::ColumnStats>>,
        row_group: Option<crate::table::index_block::RowGroupRef>,
        comparator: &crate::SharedComparator,
    ) -> crate::Result<Option<crate::UserKey>> {
        let bytes_written = u32::try_from(raw.len())
            .map_err(|_| crate::Error::InvalidHeader("verbatim copy: extent length exceeds u32"))?;
        // The entries come from an UNTRUSTED (possibly tampered,
        // checksum-repatched) block; `account_direct_block` trusts their order,
        // so validate it before any state mutation.
        self.validate_direct_block_order(entries, comparator)?;
        // The extent is stamped for the file position, so whatever the writer
        // still holds goes out first; a refused extent then counts no rows.
        self.settle_pending_blocks()?;
        let restamped = self.restamp_extent(raw, source)?;
        let Some(inputs) = self.account_direct_block(entries)? else {
            return Ok(None);
        };
        {
            #[cfg(not(feature = "std"))]
            use crate::io::Write;
            #[cfg(feature = "std")]
            use std::io::Write;
            // Append straight into the active data region, exactly where
            // `Block::write_to` would have written a freshly-encoded block's bytes.
            self.file_writer.write_all(&restamped)?;
        }
        self.register_written_extent(
            bytes_written,
            uncompressed_length,
            layout,
            inputs.last_key.clone(),
            inputs.last_seqno,
            inputs.seqno_bounds,
            inputs.item_count,
            inputs.zone_block_min,
            columnar_columns,
            row_group,
        )?;
        self.locator_block_done();
        Ok(Some(inputs.last_key))
    }

    /// Validates a direct (verbatim / re-emitted) block's entries against the
    /// internal-key ordering invariant BEFORE any writer state mutates: user
    /// keys non-decreasing (duplicate user keys are MVCC versions and are
    /// allowed, unlike bulk ingest), equal user keys carry strictly decreasing
    /// seqnos, and the block must not sort before the previously written key.
    /// The salvage walk feeds these paths with entries decoded from untrusted
    /// (possibly tampered, checksum-repatched) blocks; emitting an out-of-order
    /// block would register a wrong last-key in the index and corrupt binary
    /// search / scan order in the recovered SST. `pub(crate)` so the salvage
    /// walk can also pre-validate its row-by-row re-emit path.
    pub(crate) fn validate_direct_block_order(
        &self,
        entries: &[InternalValue],
        comparator: &crate::SharedComparator,
    ) -> crate::Result<()> {
        let err = || crate::Error::InvalidHeader("direct block entries out of internal-key order");
        // A key's versions descend by seqno, with ONE legitimate tie: a single
        // write batch may add several merge operands for a key, and they share
        // the batch's seqno, so a flush stores them all at it. Rejecting that
        // would make salvage drop an otherwise clean block — losing valid
        // operands and changing the merged value the repaired tree serves.
        // Every other equal-seqno pair stays a violation: for those kinds the
        // internal key ties without a tie-breaker, so their order (and so the
        // value a read serves) would not be reproducible.
        let tied_operands = |a: &crate::InternalValue, b: &crate::InternalValue| {
            a.key.value_type == crate::ValueType::MergeOperand
                && b.key.value_type == crate::ValueType::MergeOperand
        };
        if let (Some(prev), Some(first)) = (
            Self::last_key_in(&self.chunk, self.spilled_key.as_ref()),
            entries.first(),
        ) {
            match comparator.compare(prev.as_ref(), first.key.user_key.as_ref()) {
                core::cmp::Ordering::Greater => return Err(err()),
                // The same user key spans the block boundary: its versions must
                // keep descending across the edge, just like the in-block window
                // check below, and a tie is allowed only for the operand run
                // above — which the edge does not have to break. An untracked
                // prior seqno cannot happen once a key was written (both write
                // paths record it), so treat it as a violation, not a pass.
                core::cmp::Ordering::Equal
                    if self.current_key_seqno.is_none_or(|prev_seqno| {
                        first.key.seqno > prev_seqno
                            || (first.key.seqno == prev_seqno
                                && !(first.key.value_type == crate::ValueType::MergeOperand
                                    && self.previous_type == Some(crate::ValueType::MergeOperand)))
                    }) =>
                {
                    return Err(err());
                }
                _ => {}
            }
        }
        for pair in entries.windows(2) {
            let (Some(a), Some(b)) = (pair.first(), pair.get(1)) else {
                continue;
            };
            match comparator.compare(a.key.user_key.as_ref(), b.key.user_key.as_ref()) {
                core::cmp::Ordering::Less => {}
                core::cmp::Ordering::Equal if a.key.seqno > b.key.seqno => {}
                core::cmp::Ordering::Equal if a.key.seqno == b.key.seqno && tied_operands(a, b) => {
                }
                _ => return Err(err()),
            }
        }
        Ok(())
    }

    /// Builds the parallel compressor on first use, capturing the now-finalized
    /// transform params. No-op once built. Only reachable when
    /// `parallel_settings` is set.
    #[cfg(feature = "std")]
    fn ensure_parallel(&mut self) {
        if self.parallel.is_some() {
            return;
        }
        if let Some(parallel) = self.parallel_settings.clone() {
            let inline_below = parallel.inline_below.map_or_else(
                || parallel_compressor::default_inline_below(self.data_block_compression),
                u64::from,
            );
            self.parallel = Some(BlockCompressor::new(
                parallel.spawner,
                parallel.threads,
                self.parallel_cap,
                inline_below,
                self.table_id,
                self.data_block_compression,
                self.encryption.clone(),
                #[cfg(zstd_any)]
                self.zstd_dictionary.clone(),
                #[cfg(zstd_any)]
                self.zstd_two_pass_seed,
                self.ecc,
            ));
        }
    }

    /// Drains one prepared block from the parallel pipeline in submission order,
    /// writes it, and registers its index entry. No-op when nothing is in
    /// flight; the per-block error is propagated.
    #[cfg(feature = "std")]
    fn drain_one_parallel(&mut self) -> crate::Result<()> {
        let Some(prepared) = self.parallel.as_mut().and_then(BlockCompressor::take_next) else {
            return Ok(());
        };
        let mut prepared = prepared?;
        let Some(meta) = self.pending_meta.pop_front() else {
            return Err(crate::Error::Io(crate::io::Error::other(
                "parallel block pipeline meta desync",
            )));
        };
        // Take the inner-block layout before `write_to` consumes the block.
        let layout = core::mem::take(&mut prepared.layout);
        // Bound here, where its place is known: the workers only prepare.
        let at = next_block_at(self.table_id, &self.file_writer);
        let header = prepared.write_to(&mut self.file_writer, at)?;
        // The block's bound was added to the in-flight total when it was
        // queued, so the total holds at least this much.
        debug_assert!(self.parallel_pending_bytes >= meta.frame_bound);
        self.parallel_pending_bytes -= meta.frame_bound;
        debug_assert!(self.parallel_pending_heap >= meta.heap_bound);
        self.parallel_pending_heap -= meta.heap_bound;
        self.register_written_block(
            header,
            layout,
            meta.last_key,
            meta.last_seqno,
            meta.seqno_bounds,
            meta.item_count,
            meta.zone_block_min,
            None, // parallel pipeline handles row (Data) blocks only
        )
    }

    /// Finishes the table, making it durable: its file is synced and so is the
    /// directory entry that names it.
    ///
    /// # Errors
    ///
    /// Will return `Err` if an IO error occurs.
    pub fn finish(self) -> crate::Result<Option<(TableId, Checksum)>> {
        let fs = Arc::clone(&self.fs);
        let sync_mode = self.sync_mode;
        let folder = crate::file::entry_directory(&self.path).to_path_buf();
        let finished = self.finish_deferring_dir_sync(None)?;
        if finished.is_some() {
            crate::file::fsync_directory(&folder, &*fs, sync_mode)?;
        }
        Ok(finished)
    }

    /// [`Self::finish`] without the directory sync, for a caller that syncs
    /// the folder itself once it is done with it: the version install syncs
    /// each folder its new tables went to once for the whole transition.
    ///
    /// A compaction output takes its `created_at` from `ages`: the newest age
    /// among the compaction inputs its key range meets, so a rewrite keeps the
    /// age of the data it carries instead of the time it was written, and an
    /// age-based policy (FIFO's TTL and drop order) sees the same data as old
    /// after a compaction as before it. Without `ages` (a flush, an ingest, a
    /// salvage copy) the clock is read.
    // TODO: split meta writing into new function
    #[expect(clippy::too_many_lines)]
    pub(crate) fn finish_deferring_dir_sync(
        mut self,
        ages: Option<&mut AgeSweep>,
    ) -> crate::Result<Option<(TableId, Checksum)>> {
        #[cfg(not(feature = "std"))]
        use crate::io::Write;
        #[cfg(feature = "std")]
        use std::io::Write;

        // Every block written before item_count is read or the index, filter
        // and metadata are written.
        self.settle_pending_blocks()?;

        // No items and no range tombstones — delete the empty table file.
        if self.meta.item_count == 0 && self.range_tombstones.is_empty() {
            self.fs.remove_file(&self.path)?;
            return Ok(None);
        }

        // If we have range tombstones but no KV items, write a synthetic
        // weak tombstone at the first RT's start key to produce a valid index.
        // Preserve seqno bounds for real entries by saving/restoring metadata
        // around the sentinel write. The sentinel uses the table's lowest RT
        // seqno and should not influence user-visible metadata.
        // Also ensure the table metadata key range covers all range tombstones.
        if self.meta.item_count == 0 {
            // The coverage of all range tombstones: as the caller ordered it
            // under its comparator, else in byte order.
            let (mut min_start, mut max_end) = match self.range_tombstone_coverage.take() {
                Some((start, end)) => (Some(start), Some(end)),
                None => (None, None),
            };
            if min_start.is_none() {
                for rt in &self.range_tombstones {
                    match &min_start {
                        None => min_start = Some(rt.start.clone()),
                        Some(cur_min) if rt.start < *cur_min => {
                            min_start = Some(rt.start.clone());
                        }
                        _ => {}
                    }
                    match &max_end {
                        None => max_end = Some(rt.end.clone()),
                        Some(cur_max) if rt.end > *cur_max => max_end = Some(rt.end.clone()),
                        _ => {}
                    }
                }
            }
            // The sentinel uses the (seqno, start)-minimal tombstone's start and
            // seqno. Shared with the verify path so both agree on which entry is
            // the synthetic sentinel (excluded from the recorded KV seqno range).
            let (sentinel_start, sentinel_seqno) =
                match crate::range_tombstone::RangeTombstone::sentinel(&self.range_tombstones) {
                    Some((start, seqno)) => (Some(start.clone()), Some(seqno)),
                    None => (None, None),
                };

            if let (Some(start), Some(end), Some(sentinel_key), Some(sentinel_seqno)) =
                (min_start, max_end, sentinel_start, sentinel_seqno)
            {
                let saved_lo = self.meta.lowest_seqno;
                let saved_hi = self.meta.highest_seqno;
                let saved_kv_hi = self.meta.highest_kv_seqno;

                // Write a sentinel key to force index block creation in RT-only
                // tables. The sentinel must use the start key of the same
                // tombstone that contributes the lowest seqno; otherwise it can
                // become visible at a key that is not yet covered by any visible
                // range tombstone and incorrectly mask older values.
                self.write(InternalValue::new_weak_tombstone(
                    sentinel_key,
                    sentinel_seqno,
                ))?;
                // Written out now: its block, landing later, would set the key
                // range back to the sentinel's own key over the one set below.
                self.settle_pending_blocks()?;

                // Restore seqno bounds — sentinel seqno is derived from RT
                // metadata, not from user data, so it should not shift the
                // table's seqno range. Item/key counts are NOT decremented:
                // the sentinel IS an on-disk entry and counts must match
                // actual block contents for consistency with recovery/tests.
                self.meta.lowest_seqno = saved_lo;
                self.meta.highest_seqno = saved_hi;
                self.meta.highest_kv_seqno = saved_kv_hi;

                // Ensure the table's key range covers all range tombstones.
                self.meta.first_key = Some(start);
                self.meta.last_key = Some(end);
            }
        }
        // Reads skip a table by its key range before they look at its range
        // tombstones, so the range has to reach as far as they do.
        if let Some((start, end)) = self.key_range_cover.take() {
            self.meta.first_key = Some(start);
            self.meta.last_key = Some(end);
        }

        // The index and filter writers are taken out whole (their stand-ins
        // hold no buffers), so the writer stays whole for the meta below.
        log::trace!("Finishing index writer");
        let index_writer =
            core::mem::replace(&mut self.index_writer, Box::new(FullIndexWriter::new()));
        let (index_block_count, tli_bytes) = index_writer.finish(&mut self.file_writer)?;

        log::trace!("Finishing filter writer");
        // A filter built elsewhere is framed now, under the settings this
        // table finished with.
        if let Some(prebuilt) = self.prebuilt_filter.take() {
            self.filter_writer = match prebuilt {
                PrebuiltFilter::Full(filter) => Box::new(FullFilterWriter::prebuilt(filter))
                    .use_table_id(self.table_id)
                    .use_encryption(self.encryption.clone())
                    .use_ecc(self.ecc),
                PrebuiltFilter::Partitioned(partitions) => {
                    Box::new(filter::PartitionedFilterWriter::prebuilt(
                        partitions,
                        self.table_id,
                        self.encryption.clone(),
                        self.ecc,
                        self.index_block_compression,
                    )?)
                }
            };
        }
        let mut filter_writer = core::mem::replace(
            &mut self.filter_writer,
            Box::new(FullFilterWriter::new(self.bloom_policy)),
        );
        if let (Some(first), Some(last)) = (&self.meta.first_key, &self.meta.last_key) {
            filter_writer.set_key_range(first, last);
        }
        let filter = filter_writer.finish(&mut self.file_writer, self.spilled_key.as_ref())?;
        let filter_block_count = filter.blocks;
        self.meta.filter_hashes = filter.hashes;
        self.meta.filter_partition_hashes = filter.partition_hashes;

        // Write the optional inner-block layout section (only when at least one
        // data block split into >= 2 inner zstd blocks). Absent otherwise, so
        // default small-block tables gain no bytes.
        if !self.block_layouts.is_empty() {
            start_section(
                &mut self.file_writer,
                &mut self.written_back,
                &mut self.writeback_bytes,
                "block_layout",
            )?;

            self.block_buffer.clear();
            crate::table::block_layout::encode_block_layouts(
                &mut self.block_buffer,
                &self.block_layouts,
            );

            let at = next_block_at(self.table_id, &self.file_writer);
            Block::write_into(
                &mut self.file_writer,
                &self.block_buffer,
                crate::table::block::BlockIdentity {
                    table_id: self.table_id,
                    block_type: crate::table::block::BlockType::BlockLayout,
                    dict_id: 0,
                    window_log: 0,
                },
                // Layout metadata is small and read on the cold-block path;
                // store it uncompressed. Encryption / page_ecc still apply via
                // the configured providers, matching the other meta sections.
                &{
                    let t = match self.encryption.as_deref() {
                        Some(enc) => crate::table::block::BlockTransform::Encrypted(enc),
                        None => crate::table::block::BlockTransform::PLAIN,
                    };
                    if let Some(ecc) = self.ecc {
                        t.with_ecc(ecc)
                    } else {
                        t
                    }
                },
                at,
            )?;
        }

        // Write the optional seqno-bounds section (only when seqno_in_index is on
        // and at least one block was written). Parallel to the index, keyed by
        // data-block offset; absent otherwise, so the index stays legacy-sized.
        if !self.seqno_bounds_section.is_empty() {
            start_section(
                &mut self.file_writer,
                &mut self.written_back,
                &mut self.writeback_bytes,
                "seqno_bounds",
            )?;
            self.block_buffer.clear();
            crate::table::seqno_bounds::encode_seqno_bounds(
                &mut self.block_buffer,
                &self.seqno_bounds_section,
            )?;
            let at = next_block_at(self.table_id, &self.file_writer);
            Block::write_into(
                &mut self.file_writer,
                &self.block_buffer,
                crate::table::block::BlockIdentity {
                    table_id: self.table_id,
                    block_type: crate::table::block::BlockType::SeqnoBounds,
                    dict_id: 0,
                    window_log: 0,
                },
                &{
                    let t = match self.encryption.as_deref() {
                        Some(enc) => crate::table::block::BlockTransform::Encrypted(enc),
                        None => crate::table::block::BlockTransform::PLAIN,
                    };
                    if let Some(ecc) = self.ecc {
                        t.with_ecc(ecc)
                    } else {
                        t
                    }
                },
                at,
            )?;
        }

        // Write the optional zone-map section (only when the policy is on and at
        // least one block was written). Parallel to the index, keyed by
        // data-block offset; absent otherwise, so a point read pays nothing.
        if !self.zone_map_section.is_empty() {
            start_section(
                &mut self.file_writer,
                &mut self.written_back,
                &mut self.writeback_bytes,
                "zone_map",
            )?;
            self.block_buffer.clear();
            crate::table::zone_map::encode_zone_map(
                &mut self.block_buffer,
                &self.zone_map_section,
            )?;
            let at = next_block_at(self.table_id, &self.file_writer);
            Block::write_into(
                &mut self.file_writer,
                &self.block_buffer,
                crate::table::block::BlockIdentity {
                    table_id: self.table_id,
                    block_type: crate::table::block::BlockType::ZoneMap,
                    dict_id: 0,
                    window_log: 0,
                },
                &{
                    let t = match self.encryption.as_deref() {
                        Some(enc) => crate::table::block::BlockTransform::Encrypted(enc),
                        None => crate::table::block::BlockTransform::PLAIN,
                    };
                    if let Some(ecc) = self.ecc {
                        t.with_ecc(ecc)
                    } else {
                        t
                    }
                },
                at,
            )?;
        }

        // Write the optional positional delete-bitmap section (only when the
        // segment has materialized deletes AND the strategy keeps a bitmap;
        // copy-on-write drops rows instead). Applied as a row mask at scan time;
        // absent otherwise, so a read without deletes pays nothing. Compute the
        // decision ONCE: the recorded `delete_bitmap_len` below must reflect the
        // exact same predicate as the section actually written here, or the
        // reader's "count > 0 requires a section" cross-check would misfire.
        let writes_delete_bitmap =
            !self.delete_bitmap.is_empty() && self.delete_strategy.writes_bitmap();
        // Encode ONCE and reuse the bytes for both the section content and the
        // `delete_bitmap_hash` below, so the persisted hash provably covers the
        // exact bytes written (and the encode does not run twice per finish).
        let delete_bitmap_bytes = writes_delete_bitmap.then(|| self.delete_bitmap.encode());
        if writes_delete_bitmap {
            // Co-write invariant: the positional mask resolves each block's start
            // row from the zone map, and recovery rejects a delete-bitmap SST that
            // lacks one. Fail here at the write site (a clear misconfiguration)
            // rather than letting the SST be written and then fail to open.
            if self.zone_map_section.is_empty() {
                return Err(crate::Error::InvalidHeader(
                    "delete-bitmap requires the zone map (use_zone_map(true))",
                ));
            }
            start_section(
                &mut self.file_writer,
                &mut self.written_back,
                &mut self.writeback_bytes,
                "delete_bitmap",
            )?;
            self.block_buffer.clear();
            self.block_buffer
                .extend_from_slice(delete_bitmap_bytes.as_deref().unwrap_or_default());
            let at = next_block_at(self.table_id, &self.file_writer);
            Block::write_into(
                &mut self.file_writer,
                &self.block_buffer,
                crate::table::block::BlockIdentity {
                    table_id: self.table_id,
                    block_type: crate::table::block::BlockType::DeleteBitmap,
                    dict_id: 0,
                    window_log: 0,
                },
                &{
                    let t = match self.encryption.as_deref() {
                        Some(enc) => crate::table::block::BlockTransform::Encrypted(enc),
                        None => crate::table::block::BlockTransform::PLAIN,
                    };
                    if let Some(ecc) = self.ecc {
                        t.with_ecc(ecc)
                    } else {
                        t
                    }
                },
                at,
            )?;
        }

        // Write the optional owned-blob-objects section: the cell-row objects
        // this table owns, which dropping it releases. Absent when it owns none.
        if !self.owned_blob_objects.is_empty() {
            self.owned_blob_objects.sort_unstable();
            self.owned_blob_objects.dedup();
            start_section(
                &mut self.file_writer,
                &mut self.written_back,
                &mut self.writeback_bytes,
                "owned_blob_objects",
            )?;
            self.block_buffer.clear();
            encode_owned_blob_objects(&mut self.block_buffer, &self.owned_blob_objects)?;
            let at = next_block_at(self.table_id, &self.file_writer);
            Block::write_into(
                &mut self.file_writer,
                &self.block_buffer,
                crate::table::block::BlockIdentity {
                    table_id: self.table_id,
                    block_type: crate::table::block::BlockType::OwnedBlobObjects,
                    dict_id: 0,
                    window_log: 0,
                },
                &{
                    let t = match self.encryption.as_deref() {
                        Some(enc) => crate::table::block::BlockTransform::Encrypted(enc),
                        None => crate::table::block::BlockTransform::PLAIN,
                    };
                    if let Some(ecc) = self.ecc {
                        t.with_ecc(ecc)
                    } else {
                        t
                    }
                },
                at,
            )?;
        }

        // Write the optional retrieval-ribbon locator section. Emitted only when
        // the level's locator policy is enabled AND the per-SST widths fit the
        // actual layout (`build_locator_section` returns `None` to skip
        // gracefully otherwise). Absent for a default table, so no bytes added.
        let locator_section = match self.prebuilt_locator.take() {
            Some(section) => Some(section),
            None => self.locator.and_then(|spec| {
                crate::table::locator::build_locator_section(&self.locators, spec)
            }),
        };
        if let Some(section) = locator_section {
            start_section(
                &mut self.file_writer,
                &mut self.written_back,
                &mut self.writeback_bytes,
                "locator",
            )?;
            let at = next_block_at(self.table_id, &self.file_writer);
            Block::write_into(
                &mut self.file_writer,
                &section,
                crate::table::block::BlockIdentity {
                    table_id: self.table_id,
                    block_type: crate::table::block::BlockType::Locator,
                    dict_id: 0,
                    window_log: 0,
                },
                // Same protection envelope as the other meta sections: optional
                // encryption + page ECC via the configured providers.
                &{
                    let t = match self.encryption.as_deref() {
                        Some(enc) => crate::table::block::BlockTransform::Encrypted(enc),
                        None => crate::table::block::BlockTransform::PLAIN,
                    };
                    if let Some(ecc) = self.ecc {
                        t.with_ecc(ecc)
                    } else {
                        t
                    }
                },
                at,
            )?;
        }

        // Write range tombstones block (if any)
        if !self.range_tombstones.is_empty() {
            use crate::io::{LE, WriteBytesExt};

            start_section(
                &mut self.file_writer,
                &mut self.written_back,
                &mut self.writeback_bytes,
                "range_tombstones",
            )?;

            // Wire format (repeated): [start_len:u16_le][start][end_len:u16_le][end][seqno:u64_le]
            self.block_buffer.clear();
            for rt in &self.range_tombstones {
                let start_len = u16::try_from(rt.start.len()).map_err(|_| {
                    crate::io::Error::new(
                        crate::io::ErrorKind::InvalidData,
                        "range tombstone start key length exceeds u16::MAX",
                    )
                })?;
                let end_len = u16::try_from(rt.end.len()).map_err(|_| {
                    crate::io::Error::new(
                        crate::io::ErrorKind::InvalidData,
                        "range tombstone end key length exceeds u16::MAX",
                    )
                })?;

                self.block_buffer.write_u16::<LE>(start_len)?;
                self.block_buffer.extend_from_slice(&rt.start);
                self.block_buffer.write_u16::<LE>(end_len)?;
                self.block_buffer.extend_from_slice(&rt.end);
                self.block_buffer.write_u64::<LE>(rt.seqno)?;
            }

            let at = next_block_at(self.table_id, &self.file_writer);
            Block::write_into(
                &mut self.file_writer,
                &self.block_buffer,
                crate::table::block::BlockIdentity {
                    table_id: self.table_id,
                    block_type: crate::table::block::BlockType::RangeTombstone,
                    dict_id: 0,
                    window_log: 0,
                },
                // Range-tombstone blocks are always uncompressed; the
                // transform is Plain or Encrypted depending on the
                // configured provider. page_ecc upgrades to the
                // matching `*Ecc` variant when the tree opted in.
                &{
                    let t = match self.encryption.as_deref() {
                        Some(enc) => crate::table::block::BlockTransform::Encrypted(enc),
                        None => crate::table::block::BlockTransform::PLAIN,
                    };
                    if let Some(ecc) = self.ecc {
                        t.with_ecc(ecc)
                    } else {
                        t
                    }
                },
                at,
            )?;
        }

        // Snapshot all meta fields once — reused for both MID and TAIL
        // copies. Borrowing `first_key` / `last_key` from `self.meta`
        // here is fine because neither the MID write nor any of the
        // intermediate tail sections (`linked_blob_files`,
        // `table_version`, `meta_separator`) touch `self.meta`'s key
        // buffers — those stay owned by `self` through to the TAIL
        // write. Note that `self.index_writer` and `self.filter_writer`
        // are already moved by the `.finish()` calls above; we don't
        // touch them again here.
        #[expect(clippy::expect_used, reason = "non-empty table guaranteed earlier")]
        let first_key = self
            .meta
            .first_key
            .as_ref()
            .expect("first_key should exist");
        #[expect(clippy::expect_used, reason = "non-empty table guaranteed earlier")]
        let last_key = self.meta.last_key.as_ref().expect("last_key should exist");
        let range_tombstone_count = self.range_tombstones.len() as u64;
        // Decided once — both MID and TAIL copies must report the SAME
        // created_at so MID-fallback recovery produces the same timestamp as
        // a clean TAIL recovery. A compaction output inherits the age of the
        // inputs it carries data from, an imported table the age it was given;
        // anything else reads the clock.
        let created_at_nanos = self
            .created_at
            .or_else(|| ages.and_then(|ages| ages.age_of(first_key, last_key)))
            .unwrap_or_else(|| self.now_nanos());
        let mut meta_params = self.meta_section_params(
            first_key,
            last_key,
            (index_block_count, filter_block_count),
            range_tombstone_count,
            // Record the count that corresponds to the delete_bitmap section
            // ACTUALLY written above, via the single `writes_delete_bitmap`
            // decision. A strategy that keeps no bitmap (or a compaction that
            // applied its deletes, leaving the bitmap empty) writes no section
            // and records 0, so the reader's "count > 0 requires a section"
            // cross-check never fires on a legitimate table.
            //
            // Bind the delete-bitmap CONTENTS, not just its count, into the meta:
            // an equal-cardinality but checksum-valid bitmap substituted for the
            // real one would otherwise pass the count cross-check and, during
            // manifest repair (no original whole-file digest to compare against),
            // silently resurrect deleted rows / drop live ones. The hash is over
            // the exact encoded bytes written to the section above; the meta block
            // is itself checksum- and mirror-protected, so this authenticates the
            // section content transitively.
            if writes_delete_bitmap {
                (
                    self.delete_bitmap.len(),
                    crate::hash::hash128(delete_bitmap_bytes.as_deref().unwrap_or_default()),
                )
            } else {
                (0, 0)
            },
            created_at_nanos,
        );
        meta_params.prefix_extractor = indexed_extractor(
            self.prefix_extractor.as_deref(),
            self.carries_filter,
            self.partitioned_filter,
        );

        // MID meta copy — defends against torn-write at the file tail
        // (incomplete fsync). Written at the current writer cursor,
        // after `range_tombstones` and before
        // `linked_blob_files` / `table_version` / `meta_separator` /
        // `meta`. The 4 KiB `meta_separator` between MID and TAIL
        // guarantees the two copies land on different filesystem
        // sectors. `file_size` is `*self.meta.file_pos` so MID and
        // TAIL encode identical values: `file_pos` is only ever bumped
        // inside `spill_block`, so it doesn't change between the two
        // writes.
        write_meta_section(
            &mut self.file_writer,
            &mut self.block_buffer,
            self.encryption.as_deref(),
            self.ecc,
            self.table_id,
            &meta_params,
        )?;

        if !self.linked_blob_files.is_empty() {
            use crate::io::{LE, WriteBytesExt};

            start_section(
                &mut self.file_writer,
                &mut self.written_back,
                &mut self.writeback_bytes,
                "linked_blob_files",
            )?;

            #[expect(
                clippy::cast_possible_truncation,
                reason = "there are never 4 billion blob files linked to a single table"
            )]
            self.file_writer
                .write_u32::<LE>(self.linked_blob_files.len() as u32)?;

            for file in &self.linked_blob_files {
                self.file_writer.write_u64::<LE>(file.blob_file_id)?;
                self.file_writer.write_u64::<LE>(file.len as u64)?;
                self.file_writer.write_u64::<LE>(file.bytes)?;
                self.file_writer.write_u64::<LE>(file.on_disk_bytes)?;
                for key in [&file.first_key, &file.last_key] {
                    // A user key is at most `u16::MAX` bytes: the writer
                    // rejects longer keys before they reach a table.
                    let len = u16::try_from(key.len()).map_err(|_| {
                        crate::Error::InvalidHeader("linked_blob_files: key exceeds u16::MAX bytes")
                    })?;
                    self.file_writer.write_u16::<LE>(len)?;
                    self.file_writer.write_all(key)?;
                }
            }
        }

        start_section(
            &mut self.file_writer,
            &mut self.written_back,
            &mut self.writeback_bytes,
            "table_version",
        )?;
        self.file_writer.write_all(&[0x3])?;

        // 4 KiB padding section so MID copy and TAIL copy live on
        // different 4 KiB filesystem sectors. Without this, a single
        // bad sector at the tail could take out both copies (only
        // ~tens of bytes separate the linked_blob_files / table_version
        // sections between them).
        start_section(
            &mut self.file_writer,
            &mut self.written_back,
            &mut self.writeback_bytes,
            "meta_separator",
        )?;
        self.file_writer.write_all(&[0u8; META_SEPARATOR_LEN])?;

        // TLI mirror near the file tail. The head `tli` section was
        // emitted earlier in `finish()` by `index_writer.finish()` —
        // after `data` and the partitioned `index` sub-blocks (if any)
        // but before `filter` / `filter_tli` / `range_tombstones` /
        // `meta_mid` / `linked_blob_files` / `table_version` /
        // `meta_separator`. So the head copy lives in the middle of
        // the file, and `tli_tail` lives after `meta_separator` and
        // before the canonical `meta` section. Several KiB of
        // unrelated sections plus the 4 KiB `meta_separator`
        // guarantee the two copies land in different 4 KiB
        // filesystem sectors (for any typical table they are
        // hundreds of KiB to MiB apart). A torn-write or bad sector
        // at either end leaves the other copy intact. Reader prefers the tail
        // copy on open and transparently falls back to the head copy
        // on decode/checksum/decrypt failure. Re-encodes the same
        // `tli_bytes` returned by `index_writer.finish()` so the two
        // sections decode to the same logical TLI. Both copies MUST
        // be written with the same `CompressionType` (this call uses
        // `self.index_block_compression`, identical to what the
        // partitioned/full index writer used for the head): the block
        // header does not record the compression tag, so the reader
        // supplies a single `CompressionType` from table metadata
        // when decoding either copy. If the two were written under
        // different codecs, at least one copy would be undecodable.
        // The encryption nonce differs per `Block::write_into` call,
        // so the resulting ciphertext differs byte-for-byte across
        // the two copies, but both decrypt to the same plaintext
        // IndexBlock.
        start_section(
            &mut self.file_writer,
            &mut self.written_back,
            &mut self.writeback_bytes,
            "tli_tail",
        )?;
        let at = next_block_at(self.table_id, &self.file_writer);
        Block::write_into(
            &mut self.file_writer,
            &tli_bytes,
            crate::table::block::BlockIdentity {
                table_id: self.table_id,
                block_type: crate::table::block::BlockType::Index,
                dict_id: 0,
                window_log: 0,
            },
            // TLI tail mirror uses the same codec and the same zstd
            // seed strategy as the head, and never carries a zstd
            // dict. page_ecc upgrades the transform when the tree
            // opted in.
            &{
                let t = crate::table::block::BlockTransform::from_parts(
                    self.index_block_compression,
                    self.encryption.as_deref(),
                    #[cfg(zstd_any)]
                    None,
                )?;
                #[cfg(zstd_any)]
                let t = t.with_two_pass_seed(self.zstd_two_pass_seed);
                if let Some(ecc) = self.ecc {
                    t.with_ecc(ecc)
                } else {
                    t
                }
            },
            at,
        )?;

        // TAIL meta — the canonical, authoritative copy. `file_size`
        // is the LOGICAL size up to the end of the data blocks
        // (`self.meta.file_pos`), NOT the on-disk offset of this
        // block. `file_pos` is bumped only inside `spill_block`, so
        // it does not track the absolute file cursor here; the value
        // is the same one MID encoded above, which is the contract
        // downstream consumers (`Table::file_size`, compaction window
        // picking, checkpoint sizing — see the comment in
        // `checkpoint.rs:170-185`) read.
        meta_params.section_name = "meta";
        // file_size already carries `*self.meta.file_pos` from the MID write;
        // `file_pos` hasn't moved since (no intermediate write touches it), so
        // re-assigning is a no-op. Kept explicit for readability.
        meta_params.file_size = *self.meta.file_pos;
        // The index mirror above can outgrow a writeback step on a large
        // partitioned index; what follows it is a few KiB.
        write_back_tail(
            &mut self.file_writer,
            &mut self.written_back,
            &mut self.writeback_bytes,
        )?;
        write_meta_section(
            &mut self.file_writer,
            &mut self.block_buffer,
            self.encryption.as_deref(),
            self.ecc,
            self.table_id,
            &meta_params,
        )?;

        // Write fixed-size trailer and make the table durable. Data and size
        // are all a read of it needs (POSIX synchronized I/O data integrity
        // completion), so `sync_data`; its name is the directory sync's. A
        // deferred table is synced by the install that persists it.
        let checksum = self.file_writer.into_inner()?;
        {
            let file = checksum.get_ref().get_ref();
            if self.durability == crate::config::FlushDurability::Synced {
                FsFile::sync_data_with(&**file, self.sync_mode)?;
            }
            // Cold-level output (L1+) is compaction product: it isn't read
            // back immediately, so drop its just-written pages from the OS
            // page cache (advisory POSIX_FADV_DONTNEED) instead of letting
            // them evict hot pages of files we are still serving reads from.
            // L0 (flush output) is left cached — freshly written keys are
            // often read back soon. Best-effort: hint() is advisory, so a
            // failure just leaves the kernel on its default policy.
            if self.initial_level >= 1 {
                let _ = FsFile::hint(&**file, FileHint::WriteOnce);
            }
        }
        let checksum = checksum.checksum();

        // The folder is not synced here; see `finish`.

        log::debug!(
            "Written {} items in {} blocks into new table file #{}, written {} MiB",
            self.meta.item_count,
            self.meta.data_block_count,
            self.table_id,
            *self.meta.file_pos / 1_024 / 1_024,
        );

        Ok(Some((self.table_id, checksum)))
    }
}

/// Starts the section `name` of a table's tail on `file_writer`, first
/// writing back what was written before it ([`write_back_tail`]).
fn start_section(
    file_writer: &mut crate::sfa::Writer<ChecksummedWriter<BufWriter<Box<dyn FsFile>>>>,
    written_back: &mut u64,
    writeback_bytes: &mut u64,
    name: &str,
) -> crate::Result<()> {
    write_back_tail(file_writer, written_back, writeback_bytes)?;
    file_writer.start(name)?;
    Ok(())
}

/// Writes back what a table's tail wrote on `file_writer` since `written_back`
/// once at least `writeback_bytes` gathered (`0` writes nothing back).
/// Free-standing so `finish` can call it while it borrows other fields of the
/// writer.
fn write_back_tail(
    file_writer: &mut crate::sfa::Writer<ChecksummedWriter<BufWriter<Box<dyn FsFile>>>>,
    written_back: &mut u64,
    writeback_bytes: &mut u64,
) -> crate::Result<()> {
    if *writeback_bytes > 0 {
        #[cfg(not(feature = "std"))]
        use crate::io::Write;
        #[cfg(feature = "std")]
        use std::io::Write;
        let checksummed = file_writer.get_mut();
        let position = checksummed.position();
        // `written_back` is a position the file already reached.
        let pending = position - *written_back;
        if pending >= *writeback_bytes {
            // The range is in the buffer until it is flushed to the file.
            checksummed.flush()?;
            crate::fs::hint_writeback(
                &**checksummed.get_ref().get_ref(),
                *written_back,
                pending,
                writeback_bytes,
            );
            *written_back = position;
        }
    }
    Ok(())
}

/// The name of the extractor whose prefixes a table's filter indexes. Only a
/// full filter built by the table's own writer does: a partitioned one never
/// indexes prefixes, and a carried one indexes whatever its source's extractor
/// emitted, which the writer cannot name.
fn indexed_extractor(
    extractor: Option<&dyn PrefixExtractor>,
    carries_filter: bool,
    partitioned: bool,
) -> Option<&str> {
    extractor
        .filter(|_| !carries_filter && !partitioned)
        .map(PrefixExtractor::name)
}

/// Parameters bundle for [`write_meta_section`]. Free-standing struct
/// (not a `Writer` method) because `finish()` calls this AFTER
/// `index_writer.finish()` / `filter_writer.finish()` have consumed
/// those two fields by-value, leaving `self` partially-moved and so
/// unable to dispatch through `&mut self` methods.
#[cfg_attr(
    test,
    expect(
        clippy::struct_excessive_bools,
        reason = "the test-only legacy-meta flag pushes the mirror of the \
                  Writer's independent on-disk format toggles past the \
                  lint's threshold"
    )
)]
struct MetaSectionParams<'a> {
    section_name: &'static str,
    index_block_count: usize,
    filter_block_count: usize,
    file_size: u64,
    table_id: TableId,
    data_block_count: u64,
    item_count: u64,
    tombstone_count: u64,
    weak_tombstone_count: u64,
    weak_tombstone_reclaimable: u64,
    key_count: u64,
    /// Hashes the filter holds: `> 0` writes `filter_hashes`, zero (no
    /// filter) omits it.
    filter_hashes: u64,
    /// Width in bits per key of the static filter policy the table was
    /// written under, written with `filter_hashes`; `None` for a filter built
    /// elsewhere.
    filter_bits: Option<u8>,
    /// Hashes of the largest filter partition, zero for a full filter.
    filter_partition_hashes: u64,
    /// Name of the extractor whose prefixes the table's full filter indexes,
    /// written with `filter_hashes`; `None` when it indexes none.
    prefix_extractor: Option<&'a str>,
    sum_user_key_bytes: u64,
    sum_value_bytes: u64,
    uncompressed_size: u64,
    first_key: &'a [u8],
    last_key: &'a [u8],
    lowest_seqno: crate::SeqNo,
    highest_seqno: crate::SeqNo,
    highest_kv_seqno: crate::SeqNo,
    data_block_compression: CompressionType,
    index_block_compression: CompressionType,
    /// Effective per-SST per-KV-footer algorithm: `Some(algo)` when this
    /// table's `(level, table_id)` satisfies the kv-checksum policy (so
    /// every data block carries a footer under `algo`), `None` otherwise.
    /// An SST is homogeneous, so this single value describes the whole
    /// table. Recorded in meta as the `descriptor#kv_checksum` byte.
    kv_checksum_algo: Option<crate::runtime_config::ChecksumAlgorithm>,
    data_block_hash_ratio: f32,
    data_block_restart_interval: u8,
    index_block_restart_interval: u8,
    initial_level: u8,
    use_columnar: bool,
    /// How the columnar blocks store each row's value; written only for a
    /// columnar table.
    value_layout: crate::table::meta::ValueLayout,
    /// How the row groups were cut and encoded; written only for a columnar
    /// table.
    group_shape: crate::table::meta::GroupShape,
    /// Bulk-ingest provenance: `Some(_)` writes `descriptor#bulk_ingested`,
    /// `None` omits it (unknown provenance, preserving a legacy SST's absence).
    bulk_ingested: Option<bool>,
    /// L0 recency key, written as `recency` on every table.
    recency: TableId,
    /// Compaction lineage: `Some(_)` writes `lineage` (the sorted input ids a
    /// compaction output merges), `None` omits it.
    lineage: Option<Vec<TableId>>,
    /// Previous output of the same run: `Some(_)` writes `lineage_prev` (only
    /// alongside `lineage`), proving adjacency for the sibling-union rule.
    lineage_prev: Option<TableId>,
    /// `true` writes the `lineage_transformed` marker (only alongside
    /// `lineage`): a filter transformed this output's window.
    lineage_transformed: bool,
    /// `true` writes the `lineage_last` marker (only alongside `lineage`):
    /// this output closed its compaction run.
    lineage_last: bool,
    /// The ids below this table's of the files its folder was still writing
    /// when it finished: written as `writing_below` when any.
    writing_below: Vec<TableId>,
    range_tombstone_count: u64,
    /// Number of positions in this table's delete bitmap. Recorded so a reader
    /// can AUTHENTICATE the bitmap's presence: the section itself is optional
    /// (omitted when empty), so without this count a re-stamped TOC could hide a
    /// non-empty `delete_bitmap` (rename it away, or replace it with another
    /// valid optional section) and resurrect every positionally-deleted row. A
    /// `> 0` count with no readable delete-bitmap section is then a forgery.
    delete_bitmap_len: u64,
    /// XXH3-128 of the delete-bitmap section's encoded bytes (0 when no section
    /// is written), binding its CONTENTS into the meta so an equal-cardinality
    /// substitution cannot pass the count-only cross-check.
    delete_bitmap_hash: u128,
    /// `created_at` snapshot taken once in `finish()`. Both MID and
    /// TAIL writes consume this same value; generating it inside
    /// `write_meta_section` per call would stamp the two copies with
    /// different wall-clock readings and shift TTL/FIFO ordering on
    /// MID-fallback recovery.
    created_at_nanos: u128,
    /// Test-only: simulate a LEGACY table written before
    /// `descriptor#delete_bitmap_hash` existed, so gates that must accept (or
    /// reject) an unauthenticatable bitmap can be exercised against a real
    /// file. Never settable in production.
    #[cfg(test)]
    omit_delete_bitmap_hash: bool,
}

/// Resolves a `(page_ecc, EccScheme)` config pair into the per-block
/// [`EccParams`](crate::table::block::EccParams) the writer emits.
///
/// `page_ecc == false` → `None` (no parity). `Secded` maps to the per-word
/// SEC-DED scheme; a shard scheme (`Xor` / `ReedSolomon`) maps to its
/// `(data_shards, parity_shards)`. There is no implicit RS(4,2) fallback.
#[must_use]
#[expect(
    clippy::expect_used,
    reason = "a zero shard count here is an upstream config-invariant violation \
              (try_update rejects it on the live path), so it is a programming \
              error that must fail loud, like the writer's other contract asserts"
)]
pub(crate) fn resolve_ecc(
    page_ecc: bool,
    scheme: crate::runtime_config::EccScheme,
) -> Option<crate::table::block::EccParams> {
    // Without the `page_ecc` feature the parity codecs are not compiled, so no
    // block can actually carry a trailer; returning `None` keeps the stored
    // writer state (`self.ecc`) aligned with the real on-disk layout instead of
    // pushing a compensating feature guard into every consumer.
    if !page_ecc || !cfg!(feature = "page_ecc") {
        return None;
    }
    if matches!(scheme, crate::runtime_config::EccScheme::Secded) {
        return Some(crate::table::block::EccParams::SECDED);
    }
    scheme.shard_params().map(|(data, parity)| {
        #[expect(
            clippy::cast_possible_truncation,
            reason = "shard counts originate as u8 in EccScheme"
        )]
        let (data, parity) = (data as u8, parity as u8);
        // Shard counts come from a runtime-validated `EccScheme`
        // (`try_update` rejects zero counts), so `try_new` cannot fail on
        // the live path. The `expect` localizes the invariant for the one
        // remaining injection point — a hand-built `Config::ecc_scheme`
        // with a zero count — to a loud config error rather than a corrupt
        // SST descriptor.
        crate::table::block::EccParams::try_new(data, parity)
            .expect("EccScheme shard counts are non-zero")
    })
}

fn meta_kv(key: &str, value: &[u8]) -> InternalValue {
    InternalValue::from_components(key, value, 0, crate::ValueType::Value)
}

fn write_meta_section<W: crate::io::Write + crate::io::Seek>(
    file_writer: &mut crate::sfa::Writer<ChecksummedWriter<W>>,
    block_buffer: &mut Vec<u8>,
    encryption: Option<&dyn EncryptionProvider>,
    ecc: Option<crate::table::block::EccParams>,
    table_id: TableId,
    p: &MetaSectionParams<'_>,
) -> crate::Result<()> {
    file_writer.start(p.section_name)?;
    encode_meta_payload(ecc, p, block_buffer)?;
    write_meta_block(file_writer, block_buffer, encryption, ecc, table_id)
}

/// Encodes the meta block payload for `p` into `block_buffer`.
fn encode_meta_payload(
    ecc: Option<crate::table::block::EccParams>,
    p: &MetaSectionParams<'_>,
    block_buffer: &mut Vec<u8>,
) -> crate::Result<()> {
    // Record the EFFECTIVE Page-ECC scheme, not the requested one. On
    // builds without the `page_ecc` cargo feature, `with_ecc()` compiles to
    // the identity and no parity trailer is ever emitted, so the descriptor
    // must read "off" to stay consistent with the on-disk blocks (otherwise
    // an SST advertises parity that isn't there).
    let effective_ecc = if cfg!(feature = "page_ecc") {
        ecc
    } else {
        None
    };

    // Per-SST ECC descriptor: 4 bytes [kind, data_shards, parity_shards,
    // granularity], recording the exact scheme the blocks were written
    // with so the read path re-derives the parity layout. Map the
    // resolved EccParams back to a scheme (parity_shards == 1 => XOR,
    // else Reed-Solomon); granularity is Block (the only level today).
    let ecc_descriptor = {
        use crate::runtime_config::{EccGranularity, EccScheme};
        use crate::table::block::EccParams;
        let cfg = effective_ecc.map(|p| {
            // Branch on the scheme: the shard accessors panic on the per-word
            // SEC-DED variant, so match it explicitly. (SEC-DED carries no
            // shard counts; its descriptor is kind=1.)
            let scheme = match p {
                EccParams::Secded => EccScheme::Secded,
                EccParams::Shard {
                    data_shards,
                    parity_shards: 1,
                } => EccScheme::Xor { data_shards },
                EccParams::Shard {
                    data_shards,
                    parity_shards,
                } => EccScheme::ReedSolomon {
                    data_shards,
                    parity_shards,
                },
            };
            (scheme, EccGranularity::Block)
        });
        crate::runtime_config::ecc_descriptor_bytes(cfg)
    };

    let meta = meta_kv;
    let mut meta_items = vec![
        meta("block_count#data", &p.data_block_count.to_le_bytes()),
        meta(
            "block_count#filter",
            &(p.filter_block_count as u64).to_le_bytes(),
        ),
        meta(
            "block_count#index",
            &(p.index_block_count as u64).to_le_bytes(),
        ),
        meta("checksum_type", &[u8::from(ChecksumType::Xxh3)]),
        meta(
            "compression#data",
            &p.data_block_compression.encode_into_vec(),
        ),
        meta(
            "compression#index",
            &p.index_block_compression.encode_into_vec(),
        ),
        meta("crate_version", env!("CARGO_PKG_VERSION").as_bytes()),
        meta("created_at", &p.created_at_nanos.to_le_bytes()),
        meta(
            "data_block_hash_ratio",
            &p.data_block_hash_ratio.to_le_bytes(),
        ),
        // Per-SST layout descriptor: whether every data block in this table is
        // column-organized (PAX) rather than row-major. One byte for the whole
        // homogeneous SST, so the read path learns the layout from the
        // descriptor instead of inspecting a block header.
        meta("descriptor#columnar", &[u8::from(p.use_columnar)]),
        // Which columnar layout this table's row groups use, stamped for the
        // same reason as the filter format below: a reader has to learn it
        // from the metadata, because a row group read under the wrong layout
        // fails as a block-role mismatch deep inside a read, and salvage would
        // grade that as damage and drop the data. A table written before the
        // stamp existed stores each row group as one block.
        meta(
            "descriptor#columnar_format",
            &[crate::table::meta::COLUMNAR_FORMAT_VERSION],
        ),
        // Which BuRR wire format this table's filter and locator sections
        // carry. Stamped so a reader learns it from the metadata instead of
        // discovering it deep inside a point read, where the only honest
        // answer left is a parse error. A table written before the stamp
        // existed carries the v1 layout, which this release cannot read.
        meta(
            "descriptor#filter_format",
            &[crate::table::filter::ribbon::burr::FORMAT_VERSION],
        ),
        // Authenticated positional-delete count: the `delete_bitmap` section is
        // optional (omitted when empty), so this count lets a reader detect a
        // re-stamped TOC that hid a non-empty bitmap. A `> 0` count with no
        // readable bitmap section is a forgery that would resurrect deleted rows.
        meta(
            "descriptor#delete_bitmap_hash",
            &p.delete_bitmap_hash.to_le_bytes(),
        ),
        meta(
            "descriptor#delete_bitmap_len",
            &p.delete_bitmap_len.to_le_bytes(),
        ),
        // Per-SST transform descriptor: per-KV-footer presence + algorithm
        // as one byte (0 = no footer, else 1 + algo wire tag). Lets the
        // reader know the whole table's footer state without inspecting any
        // data block — the foundation for descriptor-driven block layout.
        meta(
            "descriptor#kv_checksum",
            &[crate::table::block::kv_checksum::descriptor_byte(
                p.kv_checksum_algo,
            )],
        ),
        // Per-SST transform descriptor: whether every block in this table
        // carries a Reed-Solomon parity trailer (Page ECC). One byte for the
        // whole homogeneous SST, so the read path learns ECC presence from
        // the descriptor instead of a per-block header field.
        meta("descriptor#page_ecc", &ecc_descriptor),
        meta("file_size", &p.file_size.to_le_bytes()),
        meta("filter_hash_type", &[u8::from(ChecksumType::Xxh3)]),
        meta("index_keys_have_seqno", &[0x1]),
        meta("initial_level", &p.initial_level.to_le_bytes()),
        meta("item_count", &p.item_count.to_le_bytes()),
        meta("key#max", p.last_key),
        meta("key#min", p.first_key),
        meta("key_bytes#sum", &p.sum_user_key_bytes.to_le_bytes()),
        meta("key_count", &p.key_count.to_le_bytes()),
        meta("prefix_truncation#data", &[1]),
        meta("prefix_truncation#index", &[1]),
        meta(
            "range_tombstone_count",
            &p.range_tombstone_count.to_le_bytes(),
        ),
        meta(
            "restart_interval#data",
            &p.data_block_restart_interval.to_le_bytes(),
        ),
        meta(
            "restart_interval#index",
            &p.index_block_restart_interval.to_le_bytes(),
        ),
        meta("seqno#kv_max", &p.highest_kv_seqno.to_le_bytes()),
        meta("seqno#max", &p.highest_seqno.to_le_bytes()),
        meta("seqno#min", &p.lowest_seqno.to_le_bytes()),
        meta("table_id", &p.table_id.to_le_bytes()),
        meta("table_version", &[3u8]),
        meta("tombstone_count", &p.tombstone_count.to_le_bytes()),
        meta("user_data_size", &p.uncompressed_size.to_le_bytes()),
        meta("value_bytes#sum", &p.sum_value_bytes.to_le_bytes()),
        meta(
            "weak_tombstone_count",
            &p.weak_tombstone_count.to_le_bytes(),
        ),
        meta(
            "weak_tombstone_reclaimable",
            &p.weak_tombstone_reclaimable.to_le_bytes(),
        ),
    ];

    // Per-SST provenance: whether this table was bulk-ingested (every entry at
    // local seqno 0, relying on a manifest-only global_seqno offset). Emitted
    // ONLY when the provenance is KNOWN — a `None` (legacy SST, or one salvaged
    // from a legacy source) omits the key so manifest repair reads it as unknown
    // and applies its seqno heuristic, rather than a stamped "not ingested" that
    // would wrongly keep a bulk-ingested table with global_seqno 0. Inserted out
    // of order, then the whole list is sorted below.
    if let Some(flag) = p.bulk_ingested {
        meta_items.push(meta("descriptor#bulk_ingested", &[u8::from(flag)]));
    }

    // How a columnar table stores each row's value: the whole value and a
    // caller's first field share a column id, and only the table can say
    // which one its column is. A row-major table has no value columns.
    if p.use_columnar {
        meta_items.push(meta("descriptor#value_layout", &[p.value_layout.to_byte()]));
        // How its row groups were cut and encoded: a group copied into
        // another table keeps all of it, so a copy goes only into a table
        // that would write it the same way.
        meta_items.push(meta("descriptor#group_shape", &p.group_shape.to_bytes()));
    }

    // L0 recency key, on every table: the reader requires it.
    meta_items.push(meta("recency", &p.recency.to_le_bytes()));

    // Hashes the filter holds, which a prefix extractor makes more than the
    // keys: what a filter's size follows. Emitted only when there is a filter.
    if p.filter_hashes > 0 {
        meta_items.push(meta("filter_hashes", &p.filter_hashes.to_le_bytes()));
        // How the filter was built, which a later filter budget prices it
        // by: the width of the static policy it was written under, and the
        // hashes of a full partition when it is partitioned. The level it lies
        // in does not tell either: a table keeps its filter through a move.
        if let Some(bits) = p.filter_bits {
            meta_items.push(meta("filter_bits", &[bits]));
        }
        // Whose prefixes the filter indexes: a prefix scan consults it only
        // under that extractor, and reads the table under any other.
        if let Some(name) = p.prefix_extractor {
            meta_items.push(meta("prefix_extractor", name.as_bytes()));
        }
        if p.filter_partition_hashes > 0 {
            meta_items.push(meta(
                "filter_partition_hashes",
                &p.filter_partition_hashes.to_le_bytes(),
            ));
        }
    }

    // Compaction lineage: the sorted input ids this output merges, as
    // consecutive little-endian u64s. Absence means "not a compaction
    // output" (or a table from before this key existed). Manifest repair
    // uses it to exclude a crashed compaction's DERIVED output when every
    // input survived — publishing both histories would apply the same merge
    // operands twice on read.
    if let Some(lineage) = &p.lineage {
        let mut bytes = Vec::with_capacity(lineage.len() * 8);
        for id in lineage {
            bytes.extend_from_slice(&id.to_le_bytes());
        }
        meta_items.push(meta("lineage", &bytes));
        // Adjacency proof for the sibling-union rule: repair may only union
        // output ranges whose previous-output chain is unbroken.
        if let Some(prev) = p.lineage_prev {
            meta_items.push(meta("lineage_prev", &prev.to_le_bytes()));
        }
        // Presence markers (empty payload): a transformed window, and the
        // run's closing output. Both qualify the lineage, so neither is
        // written without it.
        if p.lineage_transformed {
            meta_items.push(meta("lineage_transformed", &[]));
        }
        if p.lineage_last {
            meta_items.push(meta("lineage_last", &[]));
        }
    }

    // The files of this table's folder still being written below it, as
    // consecutive little-endian u64s; absent when none was. A repair without
    // a manifest reads an empty file under such an id as an output a power
    // loss cut short, not as a table that lost its bytes.
    if !p.writing_below.is_empty() {
        let mut bytes = Vec::with_capacity(p.writing_below.len() * 8);
        for id in &p.writing_below {
            bytes.extend_from_slice(&id.to_le_bytes());
        }
        meta_items.push(meta("writing_below", &bytes));
    }

    // Test-only legacy simulation: drop the bitmap-content hash so the file
    // matches what a pre-hash writer produced.
    #[cfg(test)]
    if p.omit_delete_bitmap_hash {
        meta_items.retain(|kv| kv.key.user_key.as_ref() != b"descriptor#delete_bitmap_hash");
    }

    // The data-block encoder requires sorted keys; the entry above may be out of
    // position (and future optional entries likewise), so sort rather than assert.
    meta_items.sort_by(|a, b| a.key.cmp(&b.key));

    block_buffer.clear();
    DataBlock::encode_into(block_buffer, &meta_items, 1, 0.0)?;
    Ok(())
}

/// Writes the meta block payload in `block_buffer` as the open section.
fn write_meta_block<W: crate::io::Write + crate::io::Seek>(
    file_writer: &mut crate::sfa::Writer<ChecksummedWriter<W>>,
    block_buffer: &[u8],
    encryption: Option<&dyn EncryptionProvider>,
    ecc: Option<crate::table::block::EccParams>,
    table_id: TableId,
) -> crate::Result<()> {
    let at = crate::table::block::ChecksumAt::block(
        table_id,
        crate::table::block::BlockType::Meta,
        file_writer.get_ref().position(),
    );
    Block::write_into(
        file_writer,
        block_buffer,
        crate::table::block::BlockIdentity {
            // Seal the Meta block under the writer's own (durable) table id.
            // The reader supplies the same id out-of-band (SST path / manifest
            // entry), so a Meta block transplanted from another SST fails AEAD
            // verification. The reader cannot take it from the meta payload
            // (that is what it is about to decrypt), hence the out-of-band
            // source on both sides.
            table_id,
            block_type: crate::table::block::BlockType::Meta,
            dict_id: 0,
            window_log: 0,
        },
        // Meta blocks are always written uncompressed; the transform
        // is Plain or Encrypted depending on whether the table is
        // keyed. page_ecc upgrades to the matching `*Ecc` variant
        // when the tree opted in.
        &{
            let t = match encryption {
                Some(enc) => crate::table::block::BlockTransform::Encrypted(enc),
                None => crate::table::block::BlockTransform::PLAIN,
            };
            // The Meta block is read at table-open BEFORE its own
            // descriptor is parsed (chicken-and-egg: the scheme is IN
            // the descriptor), so the reader cannot know a per-config
            // scheme for it. Self-describing blocks therefore use the
            // FIXED RS(4,2) layout (matching the reader's fallback);
            // the configurable scheme applies only to the SST data /
            // index / filter blocks, whose scheme the reader learns
            // from this descriptor. Without the `page_ecc` feature
            // `with_ecc` is the identity.
            if ecc.is_some() {
                t.with_ecc(crate::table::block::EccParams::RS_4_2)
            } else {
                t
            }
        },
        at,
    )?;

    Ok(())
}

#[cfg(test)]
mod tests;