coordinode-lsm-tree 7.6.0

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

use super::{Block, BlockHandle, DataBlock};
use crate::fs::FsFile;
use crate::{
    CompressionType, KeyRange, SeqNo, TableId, checksum::ChecksumType, coding::Decode,
    comparator::default_comparator, runtime_config::ChecksumAlgorithm, table::block::BlockType,
};
use alloc::vec::Vec;
use core::ops::Deref;

/// The columnar table layout this release writes and reads, stamped into
/// every table's descriptor as `descriptor#columnar_format`.
///
/// `2` is a row group laid out as a page directory followed by its pages. The
/// unstamped layout before it, which stored each row group as one block, is
/// format `1` by implication and is not read: a columnar table carrying it is
/// refused at open, so that it is converted rather than misread.
///
/// A property of the format rather than of the `columnar` feature, which is
/// why it lives here: a build without the feature still writes the stamp for
/// the tables it cannot make columnar, so every table records which layout a
/// reader should expect.
///
/// Format `2` is part of the storage `V6` contract. Until a release ships
/// `V6`, changes to its index entries, directory or pages are made in place
/// under the same number; the number moves with the first change after `V6`
/// is released.
pub(crate) const COLUMNAR_FORMAT_VERSION: u8 = 2;

/// How a columnar table stores each row's value, from the optional
/// `descriptor#value_layout` property.
///
/// The two layouts can hold a column of the same id: the whole value sits in
/// [`COL_VALUE`](crate::table::columnar::COL_VALUE), which is also the first id
/// a caller may give a field. Only the table can say which one its column
/// is, so it records it.
#[derive(Copy, Clone, Debug, Eq, PartialEq)]
pub enum ValueLayout {
    /// The whole value in one column: rows written through the row path and
    /// transposed at a flush or a compaction, whose value is read back as it
    /// was written. A table without the property stores this layout.
    Whole,
    /// The value split into caller fields, one column each: a columnar batch
    /// ingested as it was built. A field without a column is not in the table.
    Split,
    /// The table of a tree whose rows may be written as cells: a cell row's
    /// fields in the columns their ids name, its references beside them, and
    /// every other value whole, read back as it was written.
    Cells,
}

impl ValueLayout {
    /// The property's one-byte encoding.
    #[must_use]
    pub(crate) const fn to_byte(self) -> u8 {
        match self {
            Self::Whole => 0,
            Self::Split => 1,
            Self::Cells => 2,
        }
    }
}

/// How a columnar table cut and encoded its row groups, from the optional
/// `descriptor#group_shape` property: the column encoding of its pages, and
/// the row group and page sizes its writer cut at.
///
/// A group copied into another table keeps all three, so a table's groups
/// are copied only into a table that would write them the same way.
#[derive(Copy, Clone, Debug, Eq, PartialEq)]
pub struct GroupShape {
    /// The encoding its column pages were written with.
    pub encoding: crate::config::ColumnEncoding,
    /// The size its writer closed a row group at.
    pub row_group_size: u32,
    /// The size its writer cut a column page at.
    pub page_size: u32,
}

impl GroupShape {
    /// The property's encoding: the column encoding as one byte, then the
    /// row group and page sizes as little-endian `u32`s.
    pub(crate) fn to_bytes(self) -> [u8; 9] {
        let mut out = [0u8; 9];
        out[0] = match self.encoding {
            crate::config::ColumnEncoding::Plain => 0,
            crate::config::ColumnEncoding::Auto => 1,
        };
        out[1..5].copy_from_slice(&self.row_group_size.to_le_bytes());
        out[5..9].copy_from_slice(&self.page_size.to_le_bytes());
        out
    }

    /// Decodes [`Self::to_bytes`], or `None` for any other byte string and for
    /// sizes past [`crate::config::MAX_BLOCK_SIZE`], which no writer accepts.
    pub(crate) fn from_bytes(bytes: &[u8]) -> Option<Self> {
        let (&tag, rest) = bytes.split_first()?;
        let encoding = match tag {
            0 => crate::config::ColumnEncoding::Plain,
            1 => crate::config::ColumnEncoding::Auto,
            _ => return None,
        };
        let (row_group, page) = rest.split_at_checked(4)?;
        let shape = Self {
            encoding,
            row_group_size: u32::from_le_bytes(row_group.try_into().ok()?),
            page_size: u32::from_le_bytes(page.try_into().ok()?),
        };
        (shape.row_group_size <= crate::config::MAX_BLOCK_SIZE
            && shape.page_size <= crate::config::MAX_BLOCK_SIZE)
            .then_some(shape)
    }
}

/// Nanosecond timestamp.
#[derive(Copy, Clone, Debug, PartialEq, Eq, Ord, PartialOrd)]
pub struct Timestamp(u128);

impl Deref for Timestamp {
    type Target = u128;

    fn deref(&self) -> &Self::Target {
        &self.0
    }
}

impl From<Timestamp> for u128 {
    fn from(val: Timestamp) -> Self {
        val.0
    }
}

impl From<u128> for Timestamp {
    fn from(value: u128) -> Self {
        Self(value)
    }
}

// `PartialEq`: the out-of-band verifier compares the two decoded meta
// mirrors: a tail re-stamped to another internally-consistent payload is
// detectable only by disagreeing with `meta_mid`. `Clone` lets that verifier
// compare an ECC-masked copy ([`Self::without_ecc`]) without consuming the
// original.
#[derive(Debug, Clone, PartialEq, Eq)]
#[expect(
    clippy::struct_excessive_bools,
    reason = "decoded on-disk meta: each bool mirrors an independent \
              presence-marker key (bulk-ingest provenance, lineage \
              transformed / last); enums would misrepresent the format"
)]
pub struct ParsedMeta {
    pub id: TableId,
    pub created_at: Timestamp,
    pub data_block_count: u64,
    pub index_block_count: u64,
    pub key_range: KeyRange,
    pub(super) seqnos: (SeqNo, SeqNo),

    /// Highest seqno from KV entries only (excludes range tombstones).
    ///
    /// Falls back to `seqnos.1` (overall max) for tables written before
    /// this field was introduced, which is conservative but correct.
    pub(super) highest_kv_seqno: SeqNo,
    pub file_size: u64,
    pub item_count: u64,

    /// Number of DISTINCT user keys the writer counted (the `key_count` meta
    /// key), or `None` for tables written before this field was persisted.
    /// Together with [`Self::item_count`] it proves key uniqueness:
    /// `key_count == item_count` means no key carries more than one version,
    /// so a scan can skip per-key newest-visible dedup. `None` is conservative
    /// (duplicate versions assumed).
    pub key_count: Option<u64>,

    /// Hashes the table's filter holds (the `filter_hashes` meta key): one
    /// per distinct key, and under a prefix extractor one per distinct
    /// prefix. `None` for a table without a filter.
    pub filter_hashes: Option<u64>,

    /// Hashes of the largest partition of the table's filter (the
    /// `filter_partition_hashes` meta key): how its filter was split, which
    /// the level it lies in does not tell. `None` for a full filter or none.
    pub filter_partition_hashes: Option<u64>,

    /// Width in bits per key of the static filter policy the table was
    /// written under (the `filter_bits` meta key). `None` without a filter.
    pub filter_bits: Option<u8>,

    /// Name of the [`PrefixExtractor`](crate::PrefixExtractor) whose prefixes
    /// the table's full filter indexes (the `prefix_extractor` meta key).
    /// `None` when it indexes none: no extractor, a partitioned or carried
    /// filter, or no filter.
    pub prefix_extractor: Option<crate::Slice>,

    pub tombstone_count: u64,

    /// Number of RANGE tombstones the writer emitted into the
    /// `range_tombstones` section. Distinct from `tombstone_count` (point
    /// deletes). Salvage cross-checks it against the parsed section: a `> 0`
    /// count with no readable section means the section was hidden (renamed /
    /// re-roled), so salvaging would resurrect the rows it masked.
    pub range_tombstone_count: u64,

    /// Number of positions in this table's `delete_bitmap` section, or `None`
    /// for tables written before this field existed. The section is optional
    /// (omitted when empty), so this authenticated count lets a reader detect a
    /// re-stamped TOC that hid a non-empty bitmap (renamed it, or replaced it
    /// with another valid optional section): a `Some(n)` with `n > 0` and no
    /// readable delete-bitmap section is a forgery that would resurrect every
    /// positionally-deleted row.
    pub delete_bitmap_len: Option<u64>,

    /// XXH3-128 of the delete-bitmap section's encoded bytes, binding its
    /// CONTENTS (not just its cardinality) into the meta. `None` on tables
    /// written before this field. A count-only cross-check would accept an
    /// equal-cardinality checksum-valid substitution that, during manifest repair
    /// (no original whole-file digest), resurrects deleted rows / drops live ones.
    pub delete_bitmap_hash: Option<u128>,

    pub weak_tombstone_count: u64,
    pub weak_tombstone_reclaimable: u64,

    /// Sum of user-key byte lengths across all entries in this table (every
    /// version), or `None` for tables written before this field existed.
    /// Used by storage introspection to report the average key size; callers
    /// fall back to a file-size estimate when absent.
    pub sum_user_key_bytes: Option<u64>,

    /// Sum of value byte lengths across all entries in this table (every
    /// version), or `None` for tables written before this field existed.
    pub sum_value_bytes: Option<u64>,

    pub data_block_compression: CompressionType,
    pub index_block_compression: CompressionType,

    /// Per-SST per-KV-footer descriptor: `Some(algo)` when every data block
    /// in this table carries a per-KV checksum footer under `algo`, `None`
    /// when the table was written with no per-KV footers. Sourced from the
    /// `descriptor#kv_checksum` meta byte. Because an SST is homogeneous,
    /// this single value lets the read / scrub path know the footer state
    /// of the whole table without inspecting any data block header.
    pub kv_checksum_algo: Option<ChecksumAlgorithm>,

    /// `true` only when this table was written with a RECOGNIZED + applicable
    /// Page ECC scheme (`page_ecc == ecc_params.is_some()`). Under the
    /// three-state contract this is NOT a blanket "ECC present" flag: a table
    /// whose descriptor decodes to an unsupported scheme has `page_ecc = false`
    /// but [`Self::ecc_unrecognized`] `= true`.
    ///
    /// Callers MUST size / verify parity trailers from [`Self::ecc_params`]
    /// (the per-SST scheme), NOT from this boolean — for SST block types
    /// (`Data` / `Index` / `Filter` / `RangeTombstone`) decoded headers zero
    /// `block_flags`, so trailer sizing comes from `ecc_params` via
    /// `Header::on_disk_size_with(ecc)`. Self-describing block types (`Meta` /
    /// `Manifest` / `ManifestFooter`) keep the `block_flags` byte and still
    /// carry a per-block `ECC_PARITY` flag (fixed RS(4,2) layout).
    pub page_ecc: bool,

    /// Per-SST Page-ECC shard scheme decoded from the
    /// `descriptor#page_ecc` value: `Some(params)` when this table's
    /// blocks carry a parity trailer under a recognized + applicable scheme
    /// (the read path sizes + recovers with it), `None` otherwise. Agrees
    /// with [`Self::page_ecc`] (`page_ecc == ecc_params.is_some()`).
    pub ecc_params: Option<crate::table::block::EccParams>,

    /// `true` when the `descriptor#page_ecc` value decoded to an ECC scheme
    /// this build cannot apply: page granularity, an unknown kind, or a
    /// non-canonical descriptor (recognized block-granularity `Secded` /
    /// `Xor` / `ReedSolomon` are applicable, not flagged here). The
    /// per-block read still returns the payload (framed by `data_length`,
    /// checksum-verified) with `EccStatus::Unrecognized`, but the trailer
    /// length is not derivable from a scheme — so the sequential scrub walk
    /// cannot size it and must skip ECC verification for this table, warning
    /// that recompaction is needed. Mutually exclusive with a `Some`
    /// [`Self::ecc_params`].
    pub ecc_unrecognized: bool,

    /// Restart interval the data blocks were encoded with. Needed to rebuild a
    /// positional restart index when partial-decoding a block on the lazy read
    /// path (the restart heads sit every `data_block_restart_interval` entries).
    pub data_block_restart_interval: u8,

    /// Restart interval the index blocks were encoded with. Not consumed on
    /// the read path (index blocks decode without it), but persisted so a
    /// layout-mirroring rewrite (salvage) reproduces the source's index
    /// encoding instead of the writer default.
    pub index_block_restart_interval: u8,

    /// The bits of the `f32` hash index ratio the data blocks were encoded with
    /// (0 = none), as stored. Not consumed on the read path, where each block
    /// describes its own hash index, but a block copied into another table must
    /// have been built under that table's ratio.
    pub data_block_hash_ratio_bits: u32,

    /// Whether this SST's data blocks are column-organized (PAX) rather than
    /// row-major. Read from the optional `descriptor#columnar` property;
    /// defaults to `false` for SSTs written without it (row-major). The reader
    /// reconstructs row entries from a columnar block on load.
    pub columnar: bool,

    /// Which `BuRR` wire format this table's filter and locator sections
    /// carry, from the optional `descriptor#filter_format` property.
    ///
    /// `None` means the key is ABSENT, which is what a table written before
    /// the stamp existed looks like — and every such table carries the v1
    /// layout this release cannot read. The recover path refuses one that
    /// actually has a filter or locator section, naming the offline converter,
    /// rather than letting the mismatch surface as a parse error inside the
    /// first point read. A table with neither section is unaffected: there is
    /// nothing to misread.
    pub filter_format: Option<u8>,

    /// Which columnar layout this table's row groups use, from the optional
    /// `descriptor#columnar_format` property.
    ///
    /// `None` means the key is ABSENT: a table written before the stamp
    /// existed, whose row groups are each one block. The recover path refuses
    /// a COLUMNAR table that does not carry the current format, naming the
    /// offline converter; a row-major table is unaffected, since it has no
    /// row groups to misread.
    pub columnar_format: Option<u8>,

    /// How the table stores each row's value; [`ValueLayout::Whole`] for a
    /// table without the property, and for a row-major table, which has no
    /// value columns.
    pub value_layout: ValueLayout,

    /// How a columnar table cut and encoded its row groups; `None` for a
    /// row-major table, which has no row groups. Every columnar table records
    /// it, and one that does not is refused as it is read.
    pub group_shape: Option<GroupShape>,

    /// Bulk-ingest provenance from the optional `descriptor#bulk_ingested`
    /// property: `Some(true)` = bulk-ingested (every entry at LOCAL seqno 0, MVCC
    /// ordering carried by a manifest-only `global_seqno`), `Some(false)` = a
    /// normal flush / compaction table (offset 0), `None` = the key is ABSENT, a
    /// legacy SST of UNKNOWN provenance. Manifest repair fails closed on
    /// `Some(true)` and treats `None` as ambiguous (a legacy table may itself
    /// have been bulk-ingested), since it cannot reconstruct the offset from the
    /// SST alone.
    pub bulk_ingested: Option<bool>,

    /// L0 recency key from the required `recency` property: an id a flush
    /// takes when it starts, an ingested table's own id (its install records
    /// a later one in the manifest), a compaction output's highest recency
    /// among its INPUTS. A table's own id says nothing about how new its
    /// content is (a pooled file's id was taken with its batch, an output's
    /// at write start while a newer flush may still install), so L0 is
    /// ordered by this key. Every table carries it; one without it is refused
    /// as it is read.
    pub recency: TableId,

    /// Compaction lineage from the optional `lineage` property: the sorted
    /// ids of the INPUT tables a compaction output was merged from. 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. `None` — a flush / ingest table, or one
    /// written before the key existed.
    pub lineage: Option<Vec<TableId>>,

    /// The PREVIOUS output of the same compaction run, from the optional
    /// `lineage_prev` property (`None` for a run's first output, or a table
    /// without lineage). Proves ADJACENCY: 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.
    pub lineage_prev: Option<TableId>,

    /// A compaction filter TRANSFORMED this output's window (presence of the
    /// `lineage_transformed` marker): its content is not derivable from its
    /// inputs, so manifest repair must never trade it back for resurrected
    /// inputs — it supersedes the inputs it covers instead.
    pub lineage_transformed: bool,

    /// This output closed its compaction run (presence of the `lineage_last`
    /// marker). With an unbroken adjacency chain from the run's first output,
    /// it proves the surviving set is the run's COMPLETE output set.
    pub lineage_last: bool,

    /// The ids below this table's of the files its folder was still writing
    /// when it was finished, ascending, from the optional `writing_below`
    /// property (empty when none was). Every other id below this table's in
    /// its folder had its table written by then.
    pub writing_below: Vec<TableId>,
}

macro_rules! read_u8 {
    ($block:expr, $name:expr, $cmp:expr) => {{
        let item = $block
            .point_read($name, SeqNo::MAX, $cmp)?
            .ok_or(crate::Error::InvalidHeader("TableMeta"))?;
        // Single-byte meta value: reject an overlong / corrupt payload instead
        // of silently truncating to the first byte (read_u8 ignores trailing
        // bytes), which would weaken corruption detection on these
        // format-critical descriptor fields.
        match &item.value[..] {
            [b] => *b,
            _ => return Err(crate::Error::InvalidHeader("TableMeta")),
        }
    }};
}

macro_rules! read_u64 {
    ($block:expr, $name:expr, $cmp:expr) => {{
        let item = $block
            .point_read($name, SeqNo::MAX, $cmp)?
            .ok_or(crate::Error::InvalidHeader("TableMeta"))?;
        // Exactly eight little-endian bytes: an overlong / short payload is
        // corrupt meta, not silently truncatable.
        let bytes = <[u8; 8]>::try_from(&item.value[..])
            .map_err(|_| crate::Error::InvalidHeader("TableMeta"))?;
        u64::from_le_bytes(bytes)
    }};
}

macro_rules! read_u128 {
    ($block:expr, $name:expr, $cmp:expr) => {{
        let item = $block
            .point_read($name, SeqNo::MAX, $cmp)?
            .ok_or(crate::Error::InvalidHeader("TableMeta"))?;
        // Exactly sixteen little-endian bytes — same exact-width contract as
        // read_u64!, so an overlong / short payload is rejected as corrupt meta.
        let bytes = <[u8; 16]>::try_from(&item.value[..])
            .map_err(|_| crate::Error::InvalidHeader("TableMeta"))?;
        u128::from_le_bytes(bytes)
    }};
}

/// Validates that `kv_seqno` does not exceed `max_seqno`.
///
/// KV-only seqno must be ≤ overall max (which includes both KV and RT seqnos).
/// A value above `max_seqno` indicates on-disk corruption.
fn validated_kv_seqno(kv_seqno: SeqNo, max_seqno: SeqNo) -> crate::Result<SeqNo> {
    if kv_seqno > max_seqno {
        return Err(crate::io::Error::new(
            crate::io::ErrorKind::InvalidData,
            "seqno#kv_max exceeds seqno#max",
        )
        .into());
    }
    Ok(kv_seqno)
}

fn validated_restart_interval_index(restart_interval: u8) -> crate::Result<u8> {
    if restart_interval == 0 {
        return Err(crate::io::Error::new(
            crate::io::ErrorKind::InvalidData,
            "restart_interval#index must be greater than zero",
        )
        .into());
    }
    Ok(restart_interval)
}

impl ParsedMeta {
    /// Returns `self` with the ECC-descriptor fields normalized to their
    /// "no scheme" values (`page_ecc = false`, `ecc_params = None`,
    /// `ecc_unrecognized = false`).
    ///
    /// The out-of-band mirror comparison uses this so two `meta` / `meta_mid`
    /// copies diverge only on a NON-ECC field: an ECC-descriptor-only difference
    /// (a lone unrecognized sibling, or a descriptor-only forge) is arbitrated
    /// separately and must not by itself condemn a healthy table, while a change
    /// to a correctness field like `created_at` still surfaces even when it hides
    /// behind an unrecognized descriptor.
    #[cfg(feature = "std")]
    pub(crate) fn without_ecc(mut self) -> Self {
        self.page_ecc = false;
        self.ecc_params = None;
        self.ecc_unrecognized = false;
        self
    }

    #[expect(clippy::too_many_lines)]
    pub fn load_with_handle(
        file: &dyn FsFile,
        handle: &BlockHandle,
        expected_table_id: Option<crate::TableId>,
        encryption: Option<&dyn crate::encryption::EncryptionProvider>,
    ) -> crate::Result<Self> {
        let block = Block::from_file(
            file,
            *handle,
            crate::table::block::BlockIdentity {
                // The caller supplies the durable `table_id` from an
                // out-of-band source (the SST file path / manifest entry),
                // NOT from this meta payload — so the Meta block's AAD binds
                // it to its owning table and a Meta block transplanted from
                // another SST fails AEAD verification. `table_id` is durable
                // across reopen (persisted as `metadata.id` + the file name),
                // unlike the ephemeral tree id, so binding it is safe.
                // `None` (diagnostic readers / unencrypted opens) maps to 0:
                // the AAD identity is unused without a provider, and no
                // payload-id cross-check is enforced.
                table_id: expected_table_id.unwrap_or(0),
                block_type: BlockType::Meta,
                dict_id: 0,
                window_log: 0,
            },
            // Meta blocks are always written uncompressed; the
            // transform is therefore Plain (no encryption configured)
            // or Encrypted (provider supplied by the caller).
            &match encryption {
                Some(enc) => crate::table::block::BlockTransform::Encrypted(enc),
                None => crate::table::block::BlockTransform::PLAIN,
            },
        )?;

        if block.header.block_type != BlockType::Meta {
            return Err(crate::Error::InvalidTag((
                "BlockType",
                block.header.block_type.into(),
            )));
        }

        let block = DataBlock::new(block);

        // Metadata keys are always lexicographic, so use the default comparator.
        let cmp = default_comparator();

        {
            let table_version = block
                .point_read(b"table_version", SeqNo::MAX, &cmp)?
                .ok_or(crate::Error::InvalidHeader("TableMeta"))?
                .value;

            if *table_version != [3u8] {
                return Err(crate::Error::InvalidHeader("TableMeta"));
            }
        }

        {
            let hash_type = block
                .point_read(b"filter_hash_type", SeqNo::MAX, &cmp)?
                .ok_or(crate::Error::InvalidHeader("TableMeta"))?
                .value;

            if *hash_type != [u8::from(ChecksumType::Xxh3)] {
                return Err(crate::Error::InvalidHeader("TableMeta"));
            }
        }

        {
            let hash_type = block
                .point_read(b"checksum_type", SeqNo::MAX, &cmp)?
                .ok_or(crate::Error::InvalidHeader("TableMeta"))?
                .value;

            if *hash_type != [u8::from(ChecksumType::Xxh3)] {
                return Err(crate::Error::InvalidHeader("TableMeta"));
            }
        }

        let index_block_restart_interval =
            validated_restart_interval_index(read_u8!(block, b"restart_interval#index", &cmp))?;
        // Data-block restart interval: needed to rebuild a positional restart
        // index when partial-decoding a block (lazy read path).
        let data_block_restart_interval =
            validated_restart_interval_index(read_u8!(block, b"restart_interval#data", &cmp))?;
        let data_block_hash_ratio_bits = {
            let item = block
                .point_read(b"data_block_hash_ratio", SeqNo::MAX, &cmp)?
                .ok_or(crate::Error::InvalidHeader("TableMeta"))?;
            let bytes = <[u8; 4]>::try_from(&item.value[..])
                .map_err(|_| crate::Error::InvalidHeader("TableMeta"))?;
            u32::from_le_bytes(bytes)
        };

        // Optional layout descriptor: absent (older / row-major SSTs) means
        // row-major; otherwise exactly one byte, non-zero meaning columnar. An
        // empty or overlong payload is on-disk corruption, rejected here.
        let columnar = match block.point_read(b"descriptor#columnar", SeqNo::MAX, &cmp)? {
            None => false,
            Some(v) => match v.value.as_ref() {
                [b] => *b != 0,
                _ => return Err(crate::Error::InvalidHeader("TableMeta")),
            },
        };

        // Optional BuRR wire-format stamp. Absent = a table written before the
        // stamp existed, hence the v1 layout; see the field's doc comment.
        let filter_format = match block.point_read(b"descriptor#filter_format", SeqNo::MAX, &cmp)? {
            None => None,
            Some(v) => match v.value.as_ref() {
                [b] => Some(*b),
                _ => return Err(crate::Error::InvalidHeader("TableMeta")),
            },
        };

        // Optional columnar-layout stamp. Absent = a table written before the
        // stamp existed; see the field's doc comment.
        let columnar_format =
            match block.point_read(b"descriptor#columnar_format", SeqNo::MAX, &cmp)? {
                None => None,
                Some(v) => match v.value.as_ref() {
                    [b] => Some(*b),
                    _ => return Err(crate::Error::InvalidHeader("TableMeta")),
                },
            };

        // Optional value layout, one byte. Every columnar table this format
        // writes carries it, and a store of the previous format reaches this
        // one only through the offline conversion, which stamps it on each
        // columnar table; so it is absent only from a row table, and a row
        // table holds its values whole.
        let value_layout = match block.point_read(b"descriptor#value_layout", SeqNo::MAX, &cmp)? {
            None => ValueLayout::Whole,
            Some(v) => match v.value.as_ref() {
                [0] => ValueLayout::Whole,
                [1] => ValueLayout::Split,
                [2] => ValueLayout::Cells,
                _ => return Err(crate::Error::InvalidHeader("TableMeta")),
            },
        };

        // The row group shape, recorded by every columnar table and by no
        // row-major one. Its absence is not read as an unknown shape: every
        // writer of this format records it, and a store of the previous one
        // reaches this format only through the offline conversion, which
        // writes each table anew.
        let group_shape = match block.point_read(b"descriptor#group_shape", SeqNo::MAX, &cmp)? {
            None if columnar => return Err(crate::Error::InvalidHeader("TableMeta")),
            None => None,
            Some(v) => Some(
                GroupShape::from_bytes(v.value.as_ref())
                    .ok_or(crate::Error::InvalidHeader("TableMeta"))?,
            ),
        };

        // Optional bulk-ingest provenance. `None` = the key is ABSENT: a legacy
        // SST (written before the flag existed) whose provenance is UNKNOWN, so
        // manifest repair must treat it as AMBIGUOUS (it may be a legacy
        // bulk-ingested table that stored local seqno 0 under a manifest-only
        // offset). `Some(true/false)` = a newer SST that authoritatively records
        // whether it was bulk-ingested. One byte, non-zero meaning ingested.
        let bulk_ingested = match block.point_read(b"descriptor#bulk_ingested", SeqNo::MAX, &cmp)? {
            None => None,
            Some(v) => match v.value.as_ref() {
                [b] => Some(*b != 0),
                _ => return Err(crate::Error::InvalidHeader("TableMeta")),
            },
        };

        let id = read_u64!(block, b"table_id", &cmp);
        // Cross-check the payload's stored id against the caller's durable
        // expected id (manifest entry / SST path). A mismatch means a swapped
        // or wrong-id file: reject it here rather than recovering the table
        // under the wrong logical id (on encryption-OFF opens nothing else
        // catches it; on encryption-ON opens the AEAD already bound the
        // expected id, but the explicit check fails loudly + uniformly).
        // `None` skips the check for diagnostic readers (inspect / scrub).
        if let Some(expected) = expected_table_id
            && id != expected
        {
            return Err(crate::Error::InvalidHeader("TableMeta"));
        }
        let item_count = read_u64!(block, b"item_count", &cmp);
        let tombstone_count = read_u64!(block, b"tombstone_count", &cmp);
        let range_tombstone_count = read_u64!(block, b"range_tombstone_count", &cmp);
        let data_block_count = read_u64!(block, b"block_count#data", &cmp);
        let index_block_count = read_u64!(block, b"block_count#index", &cmp);
        let _filter_block_count = read_u64!(block, b"block_count#filter", &cmp);
        let file_size = read_u64!(block, b"file_size", &cmp);
        let weak_tombstone_count = read_u64!(block, b"weak_tombstone_count", &cmp);
        let weak_tombstone_reclaimable = read_u64!(block, b"weak_tombstone_reclaimable", &cmp);

        let created_at = read_u128!(block, b"created_at", &cmp).into();

        let key_range = KeyRange::new((
            block
                .point_read(b"key#min", SeqNo::MAX, &cmp)?
                .ok_or(crate::Error::InvalidHeader("TableMeta"))?
                .value,
            block
                .point_read(b"key#max", SeqNo::MAX, &cmp)?
                .ok_or(crate::Error::InvalidHeader("TableMeta"))?
                .value,
        ));

        let seqnos = {
            let min = read_u64!(block, b"seqno#min", &cmp);
            let max = read_u64!(block, b"seqno#max", &cmp);
            (min, max)
        };

        // Optional field introduced for table-skip optimization.
        // Old tables lack this key; fall back to overall max seqno
        // (conservative: table-skip compares rt.seqno > highest_kv_seqno,
        // so falling back to the higher overall max just disables the
        // optimization for legacy tables — correct but not optimal).
        // If the key exists but is truncated, propagate the I/O error to
        // surface metadata corruption rather than silently falling back.
        let highest_kv_seqno =
            if let Some(item) = block.point_read(b"seqno#kv_max", SeqNo::MAX, &cmp)? {
                // Present-but-wrong-width is corrupt meta — require exactly 8
                // bytes rather than truncating an overlong payload.
                let bytes = <[u8; 8]>::try_from(&item.value[..])
                    .map_err(|_| crate::Error::InvalidHeader("TableMeta"))?;
                validated_kv_seqno(u64::from_le_bytes(bytes), seqnos.1)?
            } else {
                seqnos.1
            };

        let data_block_compression = {
            let bytes = block
                .point_read(b"compression#data", SeqNo::MAX, &cmp)?
                .ok_or(crate::Error::InvalidHeader("TableMeta"))?;

            let mut bytes = &bytes.value[..];
            CompressionType::decode_from(&mut bytes)?
        };

        let index_block_compression = {
            let bytes = block
                .point_read(b"compression#index", SeqNo::MAX, &cmp)?
                .ok_or(crate::Error::InvalidHeader("TableMeta"))?;

            let mut bytes = &bytes.value[..];
            CompressionType::decode_from(&mut bytes)?
        };

        // The per-SST transform descriptor keys are REQUIRED, not optional.
        // They are new in the V5 on-disk format, which bumped the block magic
        // to `[L,S,M,4]` (pre-V5 used `[L,S,M,3]`). A pre-V5 table's blocks —
        // including this meta block — are already rejected at the magic check
        // in `Header::decode_from` before this parse runs, so requiring the
        // descriptor adds no incremental incompatibility: there is no
        // readable older table that reaches this point lacking the key.
        // (Treating it as optional-with-default would only mask a corrupt or
        // truncated V5 meta block.)
        let kv_checksum_algo = crate::table::block::kv_checksum::descriptor_from_byte(read_u8!(
            block,
            b"descriptor#kv_checksum",
            &cmp
        ))?;

        // Per-SST ECC descriptor: 4 bytes [kind, data_shards,
        // parity_shards, granularity]. `kind == 0` = no parity. A present
        // descriptor records the exact shard scheme so the read path
        // re-derives the parity layout (no implicit RS(4,2) fallback).
        let (page_ecc, ecc_params, ecc_unrecognized) = {
            let v = block
                .point_read(b"descriptor#page_ecc", SeqNo::MAX, &cmp)?
                .ok_or(crate::Error::InvalidHeader("TableMeta"))?;
            use crate::runtime_config::{EccDescriptor, EccGranularity};
            // Three-state ECC contract: a recognized + applicable scheme
            // (`Secded`, `Xor`/`ReedSolomon`, block granularity, valid shards)
            // yields the recovery params (`ecc_unrecognized = false`). Anything
            // else that still decodes (page granularity, an unknown kind, or a
            // non-canonical descriptor) is NOT a hard error: it resolves to "no
            // recovery scheme" (`None`) with `ecc_unrecognized = true`. The
            // per-block read then frames the payload by `data_length`, verifies
            // it by checksum, and reports `EccStatus::Unrecognized` (a WARN +
            // recompaction hint) instead of failing; the scrub skips ECC-walk of
            // such tables.
            use crate::runtime_config::EccScheme;
            match crate::runtime_config::ecc_descriptor_from_bytes(&v.value)? {
                EccDescriptor::Off => (false, None, false),
                // SEC-DED has no shard layout (`shard_params() == None`); it
                // sizes its parity from `block_parity_len` instead, so it is
                // mapped to its dedicated `EccParams::SECDED` rather than going
                // through the shard path. Mirrors the writer's `resolve_ecc`.
                EccDescriptor::Recognized(EccScheme::Secded, EccGranularity::Block) => {
                    (true, Some(crate::table::block::EccParams::SECDED), false)
                }
                EccDescriptor::Recognized(scheme, EccGranularity::Block) => {
                    let params = scheme
                        .shard_params()
                        .map(|(d, p)| {
                            #[expect(
                                clippy::cast_possible_truncation,
                                reason = "shard counts originate as u8 in the descriptor"
                            )]
                            crate::table::block::EccParams::try_new(d as u8, p as u8)
                        })
                        .transpose()?;
                    // A non-SECDED scheme with no shard layout is unimplemented.
                    let unrecognized = params.is_none();
                    (params.is_some(), params, unrecognized)
                }
                EccDescriptor::Recognized(_, EccGranularity::Page)
                | EccDescriptor::Unrecognized => (false, None, true),
            }
        };

        // Optional shape fields (absent on tables written before storage
        // introspection support; callers fall back to a file-size estimate).
        // Present-but-wrong-width is corrupt meta, so require exactly 8 bytes.
        let read_opt_u64 = |key: &[u8]| -> crate::Result<Option<u64>> {
            match block.point_read(key, SeqNo::MAX, &cmp)? {
                Some(item) => {
                    let bytes = <[u8; 8]>::try_from(&item.value[..])
                        .map_err(|_| crate::Error::InvalidHeader("TableMeta"))?;
                    Ok(Some(u64::from_le_bytes(bytes)))
                }
                None => Ok(None),
            }
        };
        // Present-but-wrong-width is corrupt meta, so require exactly 16 bytes.
        let read_opt_u128 = |key: &[u8]| -> crate::Result<Option<u128>> {
            match block.point_read(key, SeqNo::MAX, &cmp)? {
                Some(item) => {
                    let bytes = <[u8; 16]>::try_from(&item.value[..])
                        .map_err(|_| crate::Error::InvalidHeader("TableMeta"))?;
                    Ok(Some(u128::from_le_bytes(bytes)))
                }
                None => Ok(None),
            }
        };
        let sum_user_key_bytes = read_opt_u64(b"key_bytes#sum")?;
        let sum_value_bytes = read_opt_u64(b"value_bytes#sum")?;
        // Intentionally OPTIONAL, not a required `read_u64!`: tables written
        // before this field carry no key and must still decode (`None`). It also
        // does not need to be mandatory for the forgery cross-check — the whole
        // meta block is verified field-for-field against the recovery-time copy
        // and the mirror, so a stripped or forged key fails meta integrity long
        // before the `delete_bitmap_len` cross-check would run.
        let delete_bitmap_len = read_opt_u64(b"descriptor#delete_bitmap_len")?;
        let delete_bitmap_hash = read_opt_u128(b"descriptor#delete_bitmap_hash")?;
        let key_count = read_opt_u64(b"key_count")?;
        let filter_hashes = read_opt_u64(b"filter_hashes")?;
        let filter_partition_hashes = read_opt_u64(b"filter_partition_hashes")?;
        // One byte of width; present-but-wrong-width is corrupt meta.
        let filter_bits = match block.point_read(b"filter_bits", SeqNo::MAX, &cmp)? {
            Some(item) => match *item.value {
                [bits] => Some(bits),
                _ => return Err(crate::Error::InvalidHeader("TableMeta")),
            },
            None => None,
        };
        let prefix_extractor = block
            .point_read(b"prefix_extractor", SeqNo::MAX, &cmp)?
            .map(|item| item.value);
        // Required: L0 is ordered by it, and a table id is no stand-in for a
        // compaction output's age. No released version wrote V6 tables
        // without it, so there is no older V6 shape to fall back for; V5
        // stores reach V6 only through the offline converter, which stamps it.
        let recency = read_u64!(block, b"recency", &cmp);
        // Compaction lineage: consecutive little-endian input ids. A payload
        // that is not a whole number of ids is corrupt meta.
        let lineage = match block.point_read(b"lineage", SeqNo::MAX, &cmp)? {
            Some(item) => {
                if item.value.len() % 8 != 0 {
                    return Err(crate::Error::InvalidHeader("TableMeta"));
                }
                Some(
                    item.value
                        .as_chunks::<8>()
                        .0
                        .iter()
                        .map(|&id| u64::from_le_bytes(id))
                        .collect(),
                )
            }
            None => None,
        };
        let lineage_prev = read_opt_u64(b"lineage_prev")?;
        // Presence markers: the payload is empty, only existence matters.
        let lineage_transformed = block
            .point_read(b"lineage_transformed", SeqNo::MAX, &cmp)?
            .is_some();
        let lineage_last = block
            .point_read(b"lineage_last", SeqNo::MAX, &cmp)?
            .is_some();
        // Consecutive little-endian ids, as `lineage`; absent when the
        // folder was writing no file below this one.
        let writing_below = match block.point_read(b"writing_below", SeqNo::MAX, &cmp)? {
            Some(item) => {
                if item.value.len() % 8 != 0 {
                    return Err(crate::Error::InvalidHeader("TableMeta"));
                }
                item.value
                    .as_chunks::<8>()
                    .0
                    .iter()
                    .map(|&id| u64::from_le_bytes(id))
                    .collect()
            }
            None => Vec::new(),
        };

        Ok(Self {
            id,
            created_at,
            data_block_count,
            index_block_count,
            key_range,
            seqnos,
            highest_kv_seqno,
            file_size,
            item_count,
            key_count,
            filter_hashes,
            filter_partition_hashes,
            filter_bits,
            prefix_extractor,
            tombstone_count,
            range_tombstone_count,
            delete_bitmap_len,
            delete_bitmap_hash,
            weak_tombstone_count,
            weak_tombstone_reclaimable,
            sum_user_key_bytes,
            sum_value_bytes,
            data_block_compression,
            index_block_compression,
            kv_checksum_algo,
            page_ecc,
            ecc_params,
            ecc_unrecognized,
            data_block_restart_interval,
            index_block_restart_interval,
            data_block_hash_ratio_bits,
            columnar,
            filter_format,
            columnar_format,
            value_layout,
            group_shape,
            bulk_ingested,
            recency,
            lineage,
            lineage_prev,
            lineage_transformed,
            lineage_last,
            writing_below,
        })
    }
}

#[cfg(test)]
#[allow(
    clippy::unwrap_used,
    clippy::indexing_slicing,
    clippy::useless_vec,
    reason = "test code"
)]
mod tests;