moraine 0.6.4

//! Key layout.
//!
//! A key is a typed tree — subspace, kind, components — encoded by
//! `storekey` in order-preserving form: one discriminant byte per enum
//! level (assigned by declaration order), then fixed-width big-endian
//! `u64` components in field order. Variant order is permanent once
//! written; the golden-vector tests pin the exact bytes.

use std::ops::Bound;

use storekey::{Decode, Encode};

use crate::{
    error::{Error, Result},
    store::index_encoding::CanonicalKey,
};

/// Length in bytes of the encoded subspace prefix — one discriminant
/// byte. The segment extractor and the prefix builders must agree on
/// this.
pub(crate) const TAG_PREFIX_LEN: usize = 1;

/// A fully addressed store key. Each variant is a subspace, and each
/// subspace is also a SlateDB segment (the store is created with a
/// one-byte segment extractor over the leading discriminant).
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Encode, Decode)]
pub(crate) enum Key {
    /// Store-level singletons: format version, head pointer, migration
    /// marker, and maintenance status. Overwritten in place.
    Sys(SysKey),
    /// One record per snapshot — `ducklake_snapshot` +
    /// `ducklake_snapshot_changes`, merged. Append-only.
    Snapshot {
        /// Snapshot id.
        snapshot_id: u64,
    },
    /// Live catalog state.
    Current(CurrentKey),
    /// Ended entity versions; append-only.
    History(HistoryKey),
    /// Inlined data and its archive forms.
    Inline(InlineKey),
    /// Equality-index entries. Live-only; keyed by index and canonical
    /// value.
    Index(IndexKey),
    /// `ducklake_schema_versions`: one record per snapshot at which a
    /// table's shape changed, holding the catalog schema version that
    /// snapshot minted. Retained across snapshot expiry; removed only with
    /// the table.
    SchemaVersion {
        /// The created-or-altered table (or view).
        table_id: u64,
        /// The snapshot the change landed in.
        begin_snapshot: u64,
    },
    /// The `current` keys one commit's batch wrote — the changelog a reader
    /// replays to advance a held view across that commit. Only a sliding
    /// window of the most recent commits' records is retained; later
    /// commits delete the rest.
    Changelog {
        /// The snapshot whose batch wrote these keys.
        snapshot_id: u64,
    },
}

/// An equality-index entry key. The unique kind keys on the value alone,
/// so two commits inserting the same value collide in write-write
/// detection; the non-unique kind appends the row id.
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Encode, Decode)]
pub(crate) enum IndexKey {
    /// A unique-index entry; value is the holding row id.
    Unique {
        /// The index this entry belongs to.
        index_id: u64,
        /// The canonical indexed value.
        key: CanonicalKey,
    },
    /// A non-unique-index entry; value is empty.
    Multi {
        /// The index this entry belongs to.
        index_id: u64,
        /// The canonical indexed value.
        key: CanonicalKey,
        /// The holding row id, disambiguating rows that share a value.
        row_id: u64,
    },
}

/// Store-level singleton keys.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Encode, Decode)]
pub(crate) enum SysKey {
    /// Layout format version + moraine version that wrote the store.
    Format,
    /// Latest committed snapshot id.
    Head,
    /// Structural-migration marker. Reserved from format v1.
    Migration,
    /// Bounded history of completed maintenance passes.
    MaintenanceStatus,
}

/// A live record: a temporally versioned entity, or the `current`-only
/// gc-file bookkeeping (no begin/end lifecycle, so no `history` mirror).
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Encode, Decode)]
pub(crate) enum CurrentKey {
    /// A live entity version.
    Entity(EntityKey),
    /// `ducklake_files_scheduled_for_deletion`, keyed by the scheduled
    /// file's id.
    GcFile {
        /// The scheduled data or delete file's id.
        data_file_id: u64,
    },
}

/// An ended entity version; `end_snapshot` is the final key component,
/// so a single entity's versions sort by when they ended.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Encode, Decode)]
pub(crate) struct HistoryKey {
    /// The ended entity.
    pub(crate) entity: EntityKey,
    /// Snapshot at which the version ended.
    pub(crate) end_snapshot: u64,
}

/// A temporally versioned catalog entity. Lives in `current` while live; its
/// `history` mirror appends `end_snapshot`.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Encode, Decode)]
pub(crate) enum EntityKey {
    /// `ducklake_schema`.
    Schema {
        /// Global catalog id.
        schema_id: u64,
    },
    /// `ducklake_table`.
    Table {
        /// Global catalog id.
        table_id: u64,
    },
    /// `ducklake_view`.
    View {
        /// Global catalog id.
        view_id: u64,
    },
    /// `ducklake_column` — one record per row, nested fields included.
    Column {
        /// Owning table.
        table_id: u64,
        /// Per-table field id (never reused across the table's history).
        column_id: u64,
    },
    /// `ducklake_partition_info` (+ partition columns embedded).
    Partition {
        /// Owning table.
        table_id: u64,
        /// Partition spec id.
        partition_id: u64,
    },
    /// `ducklake_data_file`.
    File {
        /// Owning table.
        table_id: u64,
        /// Global file id.
        data_file_id: u64,
    },
    /// `ducklake_delete_file`.
    DeleteFile {
        /// Owning table.
        table_id: u64,
        /// Global file id.
        delete_file_id: u64,
    },
    /// `ducklake_file_column_stats`, keyed file-major.
    FileColumnStats {
        /// Owning table.
        table_id: u64,
        /// Data file the stats describe.
        data_file_id: u64,
        /// Column the stats describe.
        column_id: u64,
    },
    /// `ducklake_table_stats`.
    TableStats {
        /// Owning table.
        table_id: u64,
    },
    /// `ducklake_table_column_stats`.
    TableColumnStats {
        /// Owning table.
        table_id: u64,
        /// Column the stats describe.
        column_id: u64,
    },
    /// `ducklake_tag` — object ids are unique across entity types.
    Tag {
        /// The tagged object's id.
        object_id: u64,
    },
    /// `ducklake_metadata` / `set_option` scopes — one record per scope.
    Option {
        /// Scope kind: global = 0, schema = 1, table = 2.
        scope_kind: u64,
        /// Scope id (0 for global).
        scope_id: u64,
    },
    /// `ducklake_sort_info` (+ sort expressions embedded).
    Sort {
        /// Owning table.
        table_id: u64,
        /// Sort spec id.
        sort_id: u64,
    },
    /// `ducklake_macro` (+ impls and parameters embedded).
    Macro {
        /// Global catalog id.
        macro_id: u64,
    },
    /// `ducklake_column_mapping` (+ name-mapping rows embedded).
    /// Unversioned and immutable: written once, never overwritten, never
    /// mirrored to history.
    Mapping {
        /// Owning table.
        table_id: u64,
        /// Mapping id, allocated by DuckLake from the file-id counter.
        mapping_id: u64,
    },
    /// A moraine-native equality-index definition. No DuckLake analog.
    Index {
        /// Owning table.
        table_id: u64,
        /// Index id, allocated from the global catalog-id counter.
        index_id: u64,
    },
}

impl EntityKey {
    /// Whether records of this kind carry a begin/end lifecycle and mirror
    /// to history when ended. Unversioned kinds (statistics, options, tags,
    /// mappings) are overwritten in place — or written once — and never
    /// appear in the history subspace.
    pub(crate) const fn is_versioned(self) -> bool {
        !matches!(
            self,
            Self::FileColumnStats { .. }
                | Self::TableStats { .. }
                | Self::TableColumnStats { .. }
                | Self::Option { .. }
                | Self::Tag { .. }
                | Self::Mapping { .. }
        )
    }
}

/// One [`EntityKey`] variant, without the fields identifying a record.
/// Used to derive a prefix that selects one catalog kind without scanning
/// every entity in `current` or `history`.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub(crate) enum EntityKind {
    /// `ducklake_schema`.
    Schema,
    /// `ducklake_table`.
    Table,
    /// `ducklake_view`.
    View,
    /// `ducklake_column`.
    Column,
    /// `ducklake_partition_info`.
    Partition,
    /// `ducklake_data_file`.
    File,
    /// `ducklake_delete_file`.
    DeleteFile,
    /// `ducklake_file_column_stats`.
    FileColumnStats,
    /// `ducklake_table_stats`.
    TableStats,
    /// `ducklake_table_column_stats`.
    TableColumnStats,
    /// `ducklake_tag`.
    Tag,
    /// `ducklake_metadata`.
    Option,
    /// `ducklake_sort_info`.
    Sort,
    /// `ducklake_macro`.
    Macro,
    /// `ducklake_column_mapping`.
    Mapping,
}

impl EntityKind {
    /// A minimal key of this kind for prefix derivation.
    const fn sample(self) -> EntityKey {
        match self {
            Self::Schema => EntityKey::Schema { schema_id: 0 },
            Self::Table => EntityKey::Table { table_id: 0 },
            Self::View => EntityKey::View { view_id: 0 },
            Self::Column => EntityKey::Column {
                table_id: 0,
                column_id: 0,
            },
            Self::Partition => EntityKey::Partition {
                table_id: 0,
                partition_id: 0,
            },
            Self::File => EntityKey::File {
                table_id: 0,
                data_file_id: 0,
            },
            Self::DeleteFile => EntityKey::DeleteFile {
                table_id: 0,
                delete_file_id: 0,
            },
            Self::FileColumnStats => EntityKey::FileColumnStats {
                table_id: 0,
                data_file_id: 0,
                column_id: 0,
            },
            Self::TableStats => EntityKey::TableStats { table_id: 0 },
            Self::TableColumnStats => EntityKey::TableColumnStats {
                table_id: 0,
                column_id: 0,
            },
            Self::Tag => EntityKey::Tag { object_id: 0 },
            Self::Option => EntityKey::Option {
                scope_kind: 0,
                scope_id: 0,
            },
            Self::Sort => EntityKey::Sort {
                table_id: 0,
                sort_id: 0,
            },
            Self::Macro => EntityKey::Macro { macro_id: 0 },
            Self::Mapping => EntityKey::Mapping {
                table_id: 0,
                mapping_id: 0,
            },
        }
    }

    /// Whether this kind has ended versions in `history`.
    pub(crate) const fn is_versioned(self) -> bool {
        self.sample().is_versioned()
    }
}

/// An inlined-data key: the per-schema-version Arrow schema, a live
/// record, the archived (post-flush) form of a live record, a
/// delete-table existence marker, a live chunk's row-range locator, or a
/// deregistered schema version's marker. New variants append: the
/// discriminant is the encoded byte.
/// `Live` and `Arch` share [`InlineOperation`], so an archive key has exactly
/// the components of its live form.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Encode, Decode)]
pub(crate) enum InlineKey {
    /// Arrow IPC schema message, once per `(table, schema_version)`. Has
    /// no archive form (schemas are never flushed away).
    Schema {
        /// Owning table.
        table_id: u64,
        /// Schema version the layout is pinned to.
        schema_version: u64,
    },
    /// A live inlined-data record.
    Live(InlineOperation),
    /// The archived form of an inlined-data record.
    Arch(InlineOperation),
    /// Marks that a table's `ducklake_inlined_delete_<table_id>` exists,
    /// once per table; the table outlives its `FileDelete` records, which
    /// a flush clears.
    FileDeleteTable {
        /// Owning table.
        table_id: u64,
    },
    /// A superseded locator, keyed by range end alone. Two live chunks can
    /// end at one row id, so this could name only one of them; nothing
    /// writes it any more and [`ChunkLocator`](Self::ChunkLocator) replaces
    /// it. Retained so a store written before the change still decodes,
    /// and so the directory's repair can sweep what it left.
    ChunkRange {
        /// Owning table.
        table_id: u64,
        /// Inclusive end of the chunk's contiguous row-id range.
        row_id_end: u64,
    },
    /// Marks that `(table_id, schema_version)`'s
    /// `ducklake_inlined_data_<t>_<v>` has been deregistered by a flush.
    /// Its [`Schema`](Self::Schema) record is retained beside it: the
    /// name still resolves, and scans it empty.
    SchemaDropped {
        /// Owning table.
        table_id: u64,
        /// The deregistered schema version.
        schema_version: u64,
    },
    /// One live inline chunk's inclusive row-id end, then the chunk's own
    /// identity. Two live chunks can end at one row id — an UPDATE
    /// re-inlines a row whose id another chunk's range still ends on — so
    /// the end alone cannot name a locator. The end leads, so a scan still
    /// seeks by row id.
    ChunkLocator {
        /// Owning table.
        table_id: u64,
        /// Inclusive end of the chunk's contiguous row-id range.
        row_id_end: u64,
        /// Schema version the chunk decodes against.
        schema_version: u64,
        /// Snapshot the chunk was written in.
        begin_snapshot: u64,
        /// The chunk's sequence within its schema version and snapshot.
        chunk_seq: u64,
    },
    /// One deletion event for an inline row; later updates append events.
    RowTombstone {
        /// Owning table.
        table_id: u64,
        /// Deleted row.
        row_id: u64,
        /// Snapshot that ended the row's preceding version.
        end_snapshot: u64,
    },
}

/// An inlined-data record that exists in both live and archived form.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Encode, Decode)]
pub(crate) enum InlineOperation {
    /// One inlined-insert chunk (Arrow record-batch body).
    Insert {
        /// Owning table.
        table_id: u64,
        /// Schema version the chunk was written under.
        schema_version: u64,
        /// Commit snapshot of the insert.
        begin_snapshot: u64,
        /// Disambiguates multiple chunks in one commit.
        chunk_seq: u64,
    },
    /// Legacy row-level tombstone, read only by the format migration.
    InlineDelete {
        /// Owning table.
        table_id: u64,
        /// Deleted row.
        row_id: u64,
    },
    /// Inlined delete against a Parquet-file row.
    FileDelete {
        /// Owning table.
        table_id: u64,
        /// Targeted data file.
        data_file_id: u64,
        /// Deleted row.
        row_id: u64,
    },
}

impl Key {
    /// A live entity key.
    pub(crate) const fn current(entity: EntityKey) -> Self {
        Self::Current(CurrentKey::Entity(entity))
    }

    /// An ended entity-version key.
    pub(crate) const fn history(entity: EntityKey, end_snapshot: u64) -> Self {
        Self::History(HistoryKey {
            entity,
            end_snapshot,
        })
    }

    /// Encode to the on-disk byte form.
    pub(crate) fn encode(&self) -> Vec<u8> {
        // Infallible by construction: a `Vec` sink raises no io error and
        // the derived `Encode` raises no custom error.
        #[allow(clippy::expect_used)]
        storekey::encode_vec(self).expect("storekey encode into a Vec cannot fail")
    }

    /// Decode from the on-disk byte form. Fails as `Corruption` on an
    /// unknown discriminant, wrong arity, or trailing bytes
    /// (`storekey::decode` verifies the reader is exhausted).
    pub(crate) fn decode(bytes: &[u8]) -> Result<Self> {
        storekey::decode(bytes).map_err(|err| Error::Corruption(format!("key: {err}")))
    }
}

/// Encodes one equality-index entry from a borrowed canonical value,
/// byte-identical to [`Key::Index`] without cloning the payload.
pub(crate) fn encode_index_entry(
    index_id: u64,
    unique: bool,
    key: &CanonicalKey,
    row_id: u64,
) -> Vec<u8> {
    let mut bytes = Vec::new();
    let mut writer = storekey::Writer::new(&mut bytes);
    // Infallible by construction: a `Vec` sink raises no I/O error and these
    // primitive/canonical encoders raise no custom error.
    let result = (|| -> std::result::Result<(), storekey::EncodeError> {
        writer.write_u8(INDEX_SUBSPACE_TAG)?;
        writer.write_u8(index_kind_tag(unique))?;
        writer.write_u64(index_id)?;
        <CanonicalKey as Encode<()>>::encode(key, &mut writer)?;
        if !unique {
            writer.write_u64(row_id)?;
        }
        Ok(())
    })();
    #[allow(clippy::expect_used)]
    result.expect("storekey encode into a Vec cannot fail");
    bytes
}

/// A subspace, fieldless — for building scan prefixes without naming
/// discriminant bytes; prefixes are derived by encoding a sample key and
/// truncating.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum Subspace {
    /// Store-level singletons.
    System,
    /// Snapshot records.
    Snapshot,
    /// Live catalog state.
    Current,
    /// Ended entity versions.
    History,
    /// Inlined data.
    Inline,
    /// Equality-index entries.
    Index,
    /// Per-table schema-version records.
    SchemaVersion,
    /// Per-commit changelogs.
    Changelog,
}

impl Subspace {
    /// Every subspace, in discriminant order. A census walks this, so a
    /// subspace added to [`Key`] and left out here goes unreported.
    pub(crate) const ALL: [Self; 8] = [
        Self::System,
        Self::Snapshot,
        Self::Current,
        Self::History,
        Self::Inline,
        Self::Index,
        Self::SchemaVersion,
        Self::Changelog,
    ];

    /// A minimal key inside this subspace, for prefix derivation.
    const fn sample(self) -> Key {
        match self {
            Self::System => Key::Sys(SysKey::Format),
            Self::Snapshot => Key::Snapshot { snapshot_id: 0 },
            Self::Current => Key::current(EntityKey::Schema { schema_id: 0 }),
            Self::History => Key::history(EntityKey::Schema { schema_id: 0 }, 0),
            Self::Inline => Key::Inline(InlineKey::Schema {
                table_id: 0,
                schema_version: 0,
            }),
            Self::Index => Key::Index(IndexKey::Unique {
                index_id: 0,
                key: CanonicalKey::empty(),
            }),
            Self::SchemaVersion => Key::SchemaVersion {
                table_id: 0,
                begin_snapshot: 0,
            },
            Self::Changelog => Key::Changelog { snapshot_id: 0 },
        }
    }
}

/// The kinds whose live keys are scoped `table_id`-first — the only kinds
/// [`current_table_prefix`] accepts.
#[allow(dead_code)]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum TableScopedKind {
    /// `ducklake_column`.
    Column,
    /// `ducklake_partition_info`.
    Partition,
    /// `ducklake_data_file`.
    File,
    /// `ducklake_delete_file`.
    DeleteFile,
    /// `ducklake_file_column_stats`.
    FileColumnStats,
    /// `ducklake_table_stats`.
    TableStats,
    /// `ducklake_table_column_stats`.
    TableColumnStats,
    /// `ducklake_sort_info`.
    Sort,
    /// `ducklake_column_mapping`.
    Mapping,
    /// Equality-index definition.
    Index,
}

impl TableScopedKind {
    /// An entity of this kind with its table id set and every other
    /// component zeroed, for prefix derivation.
    #[allow(dead_code)]
    const fn sample(self, table_id: u64) -> EntityKey {
        match self {
            Self::Column => EntityKey::Column {
                table_id,
                column_id: 0,
            },
            Self::Partition => EntityKey::Partition {
                table_id,
                partition_id: 0,
            },
            Self::File => EntityKey::File {
                table_id,
                data_file_id: 0,
            },
            Self::DeleteFile => EntityKey::DeleteFile {
                table_id,
                delete_file_id: 0,
            },
            Self::FileColumnStats => EntityKey::FileColumnStats {
                table_id,
                data_file_id: 0,
                column_id: 0,
            },
            Self::TableStats => EntityKey::TableStats { table_id },
            Self::TableColumnStats => EntityKey::TableColumnStats {
                table_id,
                column_id: 0,
            },
            Self::Sort => EntityKey::Sort {
                table_id,
                sort_id: 0,
            },
            Self::Mapping => EntityKey::Mapping {
                table_id,
                mapping_id: 0,
            },
            Self::Index => EntityKey::Index {
                table_id,
                index_id: 0,
            },
        }
    }
}

/// Discriminant bytes preceding an entity's components in a live key:
/// subspace, `CurrentKey::Entity`, entity kind.
const CUR_KIND_PREFIX_LEN: usize = 3;

/// Discriminant bytes preceding an entity's components in a history key:
/// subspace, then entity kind.
const HISTORY_KIND_PREFIX_LEN: usize = 2;

/// The three live-record kinds inside the inline subspace — the only kinds
/// [`inline_live_table_prefix`] builds a prefix for.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum InlineOperationKind {
    /// `inline/insert`.
    Insert,
    /// `inline/inline_delete`.
    InlineDelete,
    /// `inline/file_delete`.
    FileDelete,
}

impl InlineOperationKind {
    /// An op of this kind with its table id set and every other component
    /// zeroed, for prefix derivation.
    const fn sample(self, table_id: u64) -> InlineOperation {
        match self {
            Self::Insert => InlineOperation::Insert {
                table_id,
                schema_version: 0,
                begin_snapshot: 0,
                chunk_seq: 0,
            },
            Self::InlineDelete => InlineOperation::InlineDelete {
                table_id,
                row_id: 0,
            },
            Self::FileDelete => InlineOperation::FileDelete {
                table_id,
                data_file_id: 0,
                row_id: 0,
            },
        }
    }
}

/// Discriminant bytes preceding a live inline op's components: subspace,
/// `InlineKey::Live`, op kind.
const INLINE_LIVE_KIND_PREFIX_LEN: usize = 3;

/// Encodes `key` and keeps its first `len` bytes.
fn prefix_of(key: &Key, len: usize) -> Vec<u8> {
    let mut bytes = key.encode();
    bytes.truncate(len);
    bytes
}

/// Byte prefix of every live `inline/*` key of `kind` scoped to
/// `table_id`.
pub(crate) fn inline_live_table_prefix(kind: InlineOperationKind, table_id: u64) -> Vec<u8> {
    prefix_of(
        &Key::Inline(InlineKey::Live(kind.sample(table_id))),
        INLINE_LIVE_KIND_PREFIX_LEN + size_of::<u64>(),
    )
}

/// Byte prefix of one legacy inline operation kind across all tables.
pub(crate) fn inline_live_kind_prefix(kind: InlineOperationKind) -> Vec<u8> {
    prefix_of(
        &Key::Inline(InlineKey::Live(kind.sample(0))),
        INLINE_LIVE_KIND_PREFIX_LEN,
    )
}

/// Byte prefix of every versioned inline tombstone for `table_id`.
pub(crate) fn inline_row_tombstone_table_prefix(table_id: u64) -> Vec<u8> {
    prefix_of(
        &Key::Inline(InlineKey::RowTombstone {
            table_id,
            row_id: 0,
            end_snapshot: 0,
        }),
        2 + size_of::<u64>(),
    )
}

/// Discriminant bytes preceding an `inline/schema` key's components:
/// subspace, `InlineKey::Schema`.
const INLINE_SCHEMA_PREFIX_LEN: usize = 2;

/// Byte prefix of every `inline/schema` key scoped to `table_id`, across
/// all schema versions.
pub(crate) fn inline_schema_table_prefix(table_id: u64) -> Vec<u8> {
    prefix_of(
        &Key::Inline(InlineKey::Schema {
            table_id,
            schema_version: 0,
        }),
        INLINE_SCHEMA_PREFIX_LEN + size_of::<u64>(),
    )
}

/// Byte prefix of every `inline/schema` key, across every table.
pub(crate) fn inline_schema_prefix() -> Vec<u8> {
    prefix_of(
        &Key::Inline(InlineKey::Schema {
            table_id: 0,
            schema_version: 0,
        }),
        INLINE_SCHEMA_PREFIX_LEN,
    )
}

/// Discriminant bytes preceding an `inline/schema_dropped` key's
/// components: subspace, `InlineKey::SchemaDropped`.
const INLINE_SCHEMA_DROPPED_PREFIX_LEN: usize = 2;

/// Byte prefix of every `inline/schema_dropped` key scoped to `table_id`,
/// across all schema versions.
pub(crate) fn inline_schema_dropped_table_prefix(table_id: u64) -> Vec<u8> {
    prefix_of(
        &Key::Inline(InlineKey::SchemaDropped {
            table_id,
            schema_version: 0,
        }),
        INLINE_SCHEMA_DROPPED_PREFIX_LEN + size_of::<u64>(),
    )
}

/// Byte prefix of every `inline/schema_dropped` key, across every table.
pub(crate) fn inline_schema_dropped_prefix() -> Vec<u8> {
    prefix_of(
        &Key::Inline(InlineKey::SchemaDropped {
            table_id: 0,
            schema_version: 0,
        }),
        INLINE_SCHEMA_DROPPED_PREFIX_LEN,
    )
}

/// Discriminant bytes preceding an `inline/chunk_range` key's components:
/// subspace, then `InlineKey::ChunkRange`.
const INLINE_CHUNK_RANGE_PREFIX_LEN: usize = 2;

/// Discriminant bytes preceding an `inline/chunk_locator` key's components:
/// subspace, then `InlineKey::ChunkLocator`.
const INLINE_CHUNK_LOCATOR_PREFIX_LEN: usize = 2;

/// Byte prefix of every superseded chunk-range key scoped to `table_id`,
/// for the repair that sweeps them.
pub(crate) fn inline_chunk_range_table_prefix(table_id: u64) -> Vec<u8> {
    prefix_of(
        &Key::Inline(InlineKey::ChunkRange {
            table_id,
            row_id_end: 0,
        }),
        INLINE_CHUNK_RANGE_PREFIX_LEN + size_of::<u64>(),
    )
}

/// Byte prefix of every chunk locator scoped to `table_id`.
pub(crate) fn inline_chunk_locator_table_prefix(table_id: u64) -> Vec<u8> {
    prefix_of(
        &Key::Inline(InlineKey::ChunkLocator {
            table_id,
            row_id_end: 0,
            schema_version: 0,
            begin_snapshot: 0,
            chunk_seq: 0,
        }),
        INLINE_CHUNK_LOCATOR_PREFIX_LEN + size_of::<u64>(),
    )
}

/// The range suffix that seeks to the first chunk ending at or after
/// `row_id` within [`inline_chunk_locator_table_prefix`].
pub(crate) fn inline_chunk_locator_suffix(table_id: u64, row_id: u64) -> Vec<u8> {
    Key::Inline(InlineKey::ChunkLocator {
        table_id,
        row_id_end: row_id,
        schema_version: 0,
        begin_snapshot: 0,
        chunk_seq: 0,
    })
    .encode()
    .split_off(INLINE_CHUNK_LOCATOR_PREFIX_LEN + size_of::<u64>())
}

/// The `index` subspace discriminant byte, as [`Key::Index`] encodes it.
pub(crate) const INDEX_SUBSPACE_TAG: u8 = 7;
/// The [`IndexKey::Unique`] discriminant byte.
pub(crate) const INDEX_UNIQUE_TAG: u8 = 2;
/// The [`IndexKey::Multi`] discriminant byte.
pub(crate) const INDEX_MULTI_TAG: u8 = 3;
/// Discriminant bytes preceding an index entry's components: the `index`
/// subspace byte and the [`IndexKey`] kind byte.
const INDEX_KIND_PREFIX_LEN: usize = 2;
/// Bytes preceding an index entry's value: the kind discriminants and the
/// eight-byte index id.
pub(crate) const INDEX_ENTRY_PREFIX_LEN: usize = INDEX_KIND_PREFIX_LEN + size_of::<u64>();

/// The [`IndexKey`] discriminant byte for a unique or non-unique entry.
pub(crate) const fn index_kind_tag(unique: bool) -> u8 {
    if unique {
        INDEX_UNIQUE_TAG
    } else {
        INDEX_MULTI_TAG
    }
}

/// The nominal bytes one index entry stages, key and value together: the
/// entry prefix, the framed value, and one row id (in the key or as the
/// value). Escaping can make the real entry larger, never smaller.
#[cfg(test)]
pub(crate) fn index_entry_bytes(
    values: &[Option<crate::store::index_encoding::IndexKeyValue>],
) -> u64 {
    let bytes = INDEX_ENTRY_PREFIX_LEN
        + crate::store::index_encoding::nominal_key_bytes(values)
        + size_of::<u64>();
    bytes as u64
}

/// The two entry kinds inside the `index` subspace. An index is exclusively
/// one kind, so its entries form one contiguous `(kind, index_id)` range.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum IndexKind {
    /// `index/unique`.
    Unique,
    /// `index/multi`.
    Multi,
}

/// Byte prefix of every entry of one kind, across all indexes.
pub(crate) fn index_kind_prefix(kind: IndexKind) -> Vec<u8> {
    let key = match kind {
        IndexKind::Unique => Key::Index(IndexKey::Unique {
            index_id: 0,
            key: CanonicalKey::empty(),
        }),
        IndexKind::Multi => Key::Index(IndexKey::Multi {
            index_id: 0,
            key: CanonicalKey::empty(),
            row_id: 0,
        }),
    };

    prefix_of(&key, INDEX_KIND_PREFIX_LEN)
}

/// Byte prefix of every entry of one index.
pub(crate) fn index_index_prefix(kind: IndexKind, index_id: u64) -> Vec<u8> {
    let key = match kind {
        IndexKind::Unique => Key::Index(IndexKey::Unique {
            index_id,
            key: CanonicalKey::empty(),
        }),
        IndexKind::Multi => Key::Index(IndexKey::Multi {
            index_id,
            key: CanonicalKey::empty(),
            row_id: 0,
        }),
    };
    prefix_of(&key, INDEX_ENTRY_PREFIX_LEN)
}

/// Byte prefix of every non-unique entry sharing one `(index_id, value)`:
/// the entry key without its fixed eight-byte row id.
pub(crate) fn index_multi_value_prefix(index_id: u64, value: &CanonicalKey) -> Vec<u8> {
    let mut bytes = Key::Index(IndexKey::Multi {
        index_id,
        key: value.clone(),
        row_id: 0,
    })
    .encode();
    bytes.truncate(bytes.len() - size_of::<u64>());
    bytes
}

/// The framed value bytes as they appear in an entry key of `kind`, after the
/// `(kind, index_id)` prefix — the suffix a subrange bounds against.
pub(crate) fn index_value_suffix(kind: IndexKind, index_id: u64, value: &CanonicalKey) -> Vec<u8> {
    let mut bytes = match kind {
        IndexKind::Unique => Key::Index(IndexKey::Unique {
            index_id,
            key: value.clone(),
        })
        .encode(),
        IndexKind::Multi => index_multi_value_prefix(index_id, value),
    };

    bytes.drain(..INDEX_ENTRY_PREFIX_LEN);
    bytes
}

/// The framed suffix without its trailing terminator: the escaped body that
/// every key extending the value — further columns, or the row id a
/// non-unique entry ends with — carries as its own prefix.
pub(crate) fn index_value_body(kind: IndexKind, index_id: u64, value: &CanonicalKey) -> Vec<u8> {
    let mut body = index_value_suffix(kind, index_id, value);
    body.pop();
    body
}

/// The exclusive byte bound just above every entry whose value starts with
/// `value`: the value's own entries and, when it names only a leading prefix
/// of the index's columns, every extension of it. `None` when the body is all
/// `0xff` and so has no increment.
pub(crate) fn index_value_above(
    kind: IndexKind,
    index_id: u64,
    value: &CanonicalKey,
) -> Option<Vec<u8>> {
    increment_prefix(index_value_body(kind, index_id, value))
}

/// The smallest byte string lexicographically greater than every string that
/// begins with `prefix`, or `None` when `prefix` is all `0xff`.
pub(crate) fn increment_prefix(mut bytes: Vec<u8>) -> Option<Vec<u8>> {
    while let Some(last) = bytes.last_mut() {
        if *last != u8::MAX {
            *last += 1;
            return Some(bytes);
        }
        bytes.pop();
    }
    None
}

/// The kinds whose record counts scale with the data rather than the
/// schema; a bulk scan walks each in split ranges.
pub(crate) const SPLIT_SCAN_KINDS: [EntityKind; 5] = [
    EntityKind::Column,
    EntityKind::File,
    EntityKind::DeleteFile,
    EntityKind::FileColumnStats,
    EntityKind::TableColumnStats,
];

/// How many bytes past a prefix [`even_points_between`] interpolates over:
/// two big-endian id components.
const SPLIT_WIDTH: usize = 16;

/// A range of one prefix's keyspace, as bounds on the bytes after the
/// prefix.
pub(crate) type SuffixRange = (Bound<Vec<u8>>, Bound<Vec<u8>>);

/// Whether `suffix` lies within `lower` and `upper`.
fn within(lower: &Bound<Vec<u8>>, upper: &Bound<Vec<u8>>, suffix: &[u8]) -> bool {
    let above = match lower {
        Bound::Unbounded => true,
        Bound::Included(start) => suffix >= start.as_slice(),
        Bound::Excluded(start) => suffix > start.as_slice(),
    };
    let below = match upper {
        Bound::Unbounded => true,
        Bound::Included(end) => suffix <= end.as_slice(),
        Bound::Excluded(end) => suffix < end.as_slice(),
    };
    above && below
}

/// Cuts the range `lower..upper` of a prefix's keyspace at every distinct
/// point in `points` (suffixes, any order) that lies strictly inside it:
/// the ranges come back in key order, disjoint, and together cover exactly
/// `lower..upper`. No usable points yields that one range.
pub(crate) fn split_between(
    lower: Bound<Vec<u8>>,
    upper: Bound<Vec<u8>>,
    mut points: Vec<Vec<u8>>,
) -> Vec<SuffixRange> {
    points.retain(|point| {
        within(&lower, &upper, point) && !matches!(&lower, Bound::Included(start) if start == point)
    });
    points.sort_unstable();
    points.dedup();

    let mut ranges = Vec::with_capacity(points.len() + 1);
    let mut start = lower;
    for point in points {
        ranges.push((start, Bound::Excluded(point.clone())));
        start = Bound::Included(point);
    }
    ranges.push((start, upper));
    ranges
}

/// Up to `parts - 1` points spaced evenly between the suffixes `low` and
/// `high`, over their first [`SPLIT_WIDTH`] bytes read as a big-endian
/// integer (shorter suffixes are zero-padded). Fewer, or none, when the
/// extremes are too close to spread that many apart.
pub(crate) fn even_points_between(low: &[u8], high: &[u8], parts: usize) -> Vec<Vec<u8>> {
    let widen = |suffix: &[u8]| {
        let mut bytes = [0_u8; SPLIT_WIDTH];
        let taken = suffix.len().min(SPLIT_WIDTH);
        bytes[..taken].copy_from_slice(&suffix[..taken]);
        u128::from_be_bytes(bytes)
    };
    let (low, high) = (widen(low), widen(high));
    let Some(step) = high
        .checked_sub(low)
        .map(|span| span / u128::try_from(parts.max(1)).unwrap_or(u128::MAX))
    else {
        return Vec::new();
    };
    if step == 0 {
        return Vec::new();
    }

    (1..parts)
        .map(|part| low + step * u128::try_from(part).unwrap_or(u128::MAX))
        .map(|point| point.to_be_bytes().to_vec())
        .collect()
}

/// Byte prefix of every key in a subspace (exactly `TAG_PREFIX_LEN`
/// bytes — the same prefix the segment extractor derives).
pub(crate) fn subspace_prefix(subspace: Subspace) -> Vec<u8> {
    prefix_of(&subspace.sample(), TAG_PREFIX_LEN)
}

/// Byte prefix of every live key of `kind` scoped to `table_id`.
#[allow(dead_code)]
pub(crate) fn current_table_prefix(kind: TableScopedKind, table_id: u64) -> Vec<u8> {
    prefix_of(
        &Key::current(kind.sample(table_id)),
        CUR_KIND_PREFIX_LEN + size_of::<u64>(),
    )
}

/// Byte prefix of every live entity of `kind`.
pub(crate) fn current_entity_kind_prefix(kind: EntityKind) -> Vec<u8> {
    prefix_of(&Key::current(kind.sample()), CUR_KIND_PREFIX_LEN)
}

/// Byte prefix of every ended entity version of `kind`.
pub(crate) fn history_entity_kind_prefix(kind: EntityKind) -> Vec<u8> {
    prefix_of(&Key::history(kind.sample(), 0), HISTORY_KIND_PREFIX_LEN)
}

/// Byte prefix of every scheduled-file record in `current`.
pub(crate) fn current_gc_file_prefix() -> Vec<u8> {
    prefix_of(&Key::Current(CurrentKey::GcFile { data_file_id: 0 }), 2)
}

#[cfg(test)]
mod tests {

    use proptest::prelude::*;

    use super::*;
    use crate::store::index_encoding::{IndexKeyValue, IntWidth, encode_key};

    fn be(v: u64) -> Vec<u8> {
        v.to_be_bytes().to_vec()
    }

    // Golden vectors: the exact on-disk bytes, pinned per kind. The derive
    // assigns each enum level's discriminant as declaration index + 2;
    // these tests make that assignment part of the on-disk format.

    #[test]
    fn golden_sys_keys() {
        assert_eq!(Key::Sys(SysKey::Format).encode(), vec![0x02, 0x02]);
        assert_eq!(Key::Sys(SysKey::Head).encode(), vec![0x02, 0x03]);
        assert_eq!(Key::Sys(SysKey::Migration).encode(), vec![0x02, 0x04]);
        assert_eq!(
            Key::Sys(SysKey::MaintenanceStatus).encode(),
            vec![0x02, 0x05]
        );
    }

    #[test]
    fn golden_snapshot_key() {
        let mut expect = vec![0x03];
        expect.extend(be(42));
        assert_eq!(Key::Snapshot { snapshot_id: 42 }.encode(), expect);
    }

    /// Appended after every existing subspace, so no key written before it
    /// existed changes a byte.
    #[test]
    fn golden_changelog_key() {
        let mut expect = vec![0x09];
        expect.extend(be(42));
        assert_eq!(Key::Changelog { snapshot_id: 42 }.encode(), expect);
    }

    #[test]
    fn golden_current_file_key() {
        let mut expect = vec![0x04, 0x02, 0x07];
        expect.extend(be(1));
        expect.extend(be(2));
        let key = Key::current(EntityKey::File {
            table_id: 1,
            data_file_id: 2,
        });
        assert_eq!(key.encode(), expect);
    }

    #[test]
    fn golden_current_sort_key() {
        let mut expect = vec![0x04, 0x02, 0x0e];
        expect.extend(be(7));
        expect.extend(be(9));
        let key = Key::current(EntityKey::Sort {
            table_id: 7,
            sort_id: 9,
        });
        assert_eq!(key.encode(), expect);
    }

    #[test]
    fn golden_current_macro_key() {
        let mut expect = vec![0x04, 0x02, 0x0f];
        expect.extend(be(3));
        let key = Key::current(EntityKey::Macro { macro_id: 3 });
        assert_eq!(key.encode(), expect);
    }

    #[test]
    fn golden_history_file_key_appends_end_snapshot() {
        let mut expect = vec![0x05, 0x07];
        expect.extend(be(1));
        expect.extend(be(2));
        expect.extend(be(9));
        let key = Key::history(
            EntityKey::File {
                table_id: 1,
                data_file_id: 2,
            },
            9,
        );
        assert_eq!(key.encode(), expect);
    }

    #[test]
    fn golden_history_macro_key_appends_end_snapshot() {
        let mut expect = vec![0x05, 0x0f];
        expect.extend(be(3));
        expect.extend(be(9));
        let key = Key::history(EntityKey::Macro { macro_id: 3 }, 9);
        assert_eq!(key.encode(), expect);
    }

    #[test]
    fn golden_current_index_key() {
        let mut expect = vec![0x04, 0x02, 0x11];
        expect.extend(be(4));
        expect.extend(be(7));
        let key = Key::current(EntityKey::Index {
            table_id: 4,
            index_id: 7,
        });
        assert_eq!(key.encode(), expect);
    }

    #[test]
    fn golden_current_mapping_key() {
        let mut expect = vec![0x04, 0x02, 0x10];
        expect.extend(be(4));
        expect.extend(be(21));
        let key = Key::current(EntityKey::Mapping {
            table_id: 4,
            mapping_id: 21,
        });
        assert_eq!(key.encode(), expect);
    }

    #[test]
    fn golden_gcfile_key() {
        let mut expect = vec![0x04, 0x03];
        expect.extend(be(5));
        let key = Key::Current(CurrentKey::GcFile { data_file_id: 5 });
        assert_eq!(key.encode(), expect);
    }

    #[test]
    fn golden_inline_ins_key() {
        let mut expect = vec![0x06, 0x03, 0x02];
        for v in [7, 3, 11, 0] {
            expect.extend(be(v));
        }
        let key = Key::Inline(InlineKey::Live(InlineOperation::Insert {
            table_id: 7,
            schema_version: 3,
            begin_snapshot: 11,
            chunk_seq: 0,
        }));
        assert_eq!(key.encode(), expect);
    }

    #[test]
    fn golden_inline_file_delete_table_key() {
        let mut expect = vec![0x06, 0x05];
        expect.extend(be(7));
        let key = Key::Inline(InlineKey::FileDeleteTable { table_id: 7 });
        assert_eq!(key.encode(), expect);
    }

    #[test]
    fn golden_inline_schema_dropped_key() {
        let mut expect = vec![0x06, 0x07];
        expect.extend(be(7));
        expect.extend(be(3));
        let key = Key::Inline(InlineKey::SchemaDropped {
            table_id: 7,
            schema_version: 3,
        });
        assert_eq!(key.encode(), expect);
    }

    #[test]
    fn golden_inline_chunk_range_key() {
        let mut expect = vec![0x06, 0x06];
        expect.extend(be(7));
        expect.extend(be(29));
        let key = Key::Inline(InlineKey::ChunkRange {
            table_id: 7,
            row_id_end: 29,
        });
        assert_eq!(key.encode(), expect);
    }

    #[test]
    fn golden_inline_chunk_locator_key() {
        let mut expect = vec![0x06, 0x08];
        expect.extend(be(7));
        expect.extend(be(29));
        expect.extend(be(3));
        expect.extend(be(11));
        expect.extend(be(2));
        let key = Key::Inline(InlineKey::ChunkLocator {
            table_id: 7,
            row_id_end: 29,
            schema_version: 3,
            begin_snapshot: 11,
            chunk_seq: 2,
        });
        assert_eq!(key.encode(), expect);
    }

    /// The superseded key and its replacement share a prefix only as far as
    /// the subspace: a legacy locator can never be read as a current one.
    #[test]
    fn golden_inline_locator_kinds_are_distinct() {
        let legacy = Key::Inline(InlineKey::ChunkRange {
            table_id: 7,
            row_id_end: 29,
        })
        .encode();
        let current = Key::Inline(InlineKey::ChunkLocator {
            table_id: 7,
            row_id_end: 29,
            schema_version: 0,
            begin_snapshot: 0,
            chunk_seq: 0,
        })
        .encode();
        assert_ne!(legacy[1], current[1]);
    }

    #[test]
    fn golden_arch_kinds_are_distinct_from_live() {
        let ops = [
            InlineOperation::Insert {
                table_id: 7,
                schema_version: 3,
                begin_snapshot: 11,
                chunk_seq: 0,
            },
            InlineOperation::InlineDelete {
                table_id: 7,
                row_id: 9,
            },
            InlineOperation::FileDelete {
                table_id: 7,
                data_file_id: 2,
                row_id: 9,
            },
        ];

        for op in ops {
            let live = Key::Inline(InlineKey::Live(op)).encode();
            let arch = Key::Inline(InlineKey::Arch(op)).encode();
            // Same subspace and components; only the form byte differs,
            // and every archived key sorts after every live key.
            assert_eq!(live[0], 0x06);
            assert_eq!(arch[0], 0x06);
            assert_ne!(live[1], arch[1]);
            assert_eq!(live[2..], arch[2..]);
            assert!(arch > live);
        }
    }

    fn canon(values: &[IndexKeyValue]) -> CanonicalKey {
        encode_key(values).unwrap()
    }

    #[test]
    fn golden_index_unique_key_leads_with_subspace_and_index() {
        let key = Key::Index(IndexKey::Unique {
            index_id: 1,
            key: canon(&[IndexKeyValue::Int {
                value: 7,
                width: IntWidth::I64,
            }]),
        });
        let bytes = key.encode();
        // Subspace byte, then the `Unique` discriminant, then the index id.
        let mut prefix = vec![0x07, 0x02];
        prefix.extend(be(1));
        assert!(bytes.starts_with(&prefix), "{bytes:?}");
    }

    #[test]
    fn borrowed_index_entry_encoder_matches_owned_key() {
        let canonical = canon(&[
            IndexKeyValue::Str("a\0\u{1}z".to_owned()),
            IndexKeyValue::Bytes(vec![0, 1, 0xff]),
        ]);
        for unique in [false, true] {
            for row_id in [0, 1_u64 << 56, u64::MAX] {
                let owned = if unique {
                    Key::Index(IndexKey::Unique {
                        index_id: 7,
                        key: canonical.clone(),
                    })
                } else {
                    Key::Index(IndexKey::Multi {
                        index_id: 7,
                        key: canonical.clone(),
                        row_id,
                    })
                };
                assert_eq!(
                    encode_index_entry(7, unique, &canonical, row_id),
                    owned.encode()
                );
            }
        }
    }

    /// A non-unique entry key ends in a raw row id, directly after the
    /// value's terminator. A row id whose leading byte is the escape byte
    /// (`0x01`) must not be mistaken for an escaped byte of the value —
    /// that is every row id in `[2^56, 2^57)`.
    #[test]
    fn index_multi_key_roundtrips_when_the_row_id_leads_with_the_escape_byte() {
        // The last case is the shortest possible value — no components at
        // all, which frames to the bare terminator.
        let values = [
            vec![IndexKeyValue::Str("a".to_owned())],
            vec![IndexKeyValue::Int {
                value: 0,
                width: IntWidth::I64,
            }],
            vec![IndexKeyValue::Bytes(vec![0, 1, 2])],
            vec![],
        ];
        for row_id in [1u64 << 56, (1 << 56) + 1, (1 << 57) - 1] {
            for value in &values {
                let key = Key::Index(IndexKey::Multi {
                    index_id: 1,
                    key: canon(value),
                    row_id,
                });
                assert_eq!(Key::decode(&key.encode()).unwrap(), key, "row_id {row_id}");
            }
        }
    }

    /// The value rides as a storekey byte string — `0x00` and `0x01` escape
    /// behind `0x01`, a `0x00` terminates — and the raw row id follows it.
    /// A [`CanonicalKey`](crate::store::index_encoding::CanonicalKey) already
    /// holds framed bytes, so the entry key frames them a second time.
    /// Pinned because these bytes are on disk: the decode fix must not move
    /// them.
    #[test]
    fn golden_index_multi_value_framing_and_trailing_row_id() {
        let key = Key::Index(IndexKey::Multi {
            index_id: 1,
            key: canon(&[IndexKeyValue::Bytes(vec![0x00, 0x01, 0x41])]),
            row_id: 1 << 56,
        });
        let mut expected = vec![0x07, 0x03];
        expected.extend(be(1));
        // The canonical key is a non-null flag (`00`, the ascending
        // NULLS-LAST flag) then the value framed as a storekey byte string
        // (`00 01 41` -> `01 00 01 01 41 00`); the outer key framing escapes
        // each of those bytes again and terminates.
        expected.extend([
            0x01, 0x00, // 0x00 — non-null flag
            0x01, 0x01, // 0x01 — framed value byte 0 (escape lead)
            0x01, 0x00, // 0x00 — framed value byte 1 (escaped 0x00)
            0x01, 0x01, // 0x01 — framed value byte 2 (escape lead)
            0x01, 0x01, // 0x01 — framed value byte 3 (escaped 0x01)
            0x41, // 'A' — framed value byte 4
            0x01, 0x00, // 0x00 — framed value byte 5 (terminator)
            0x00, // outer terminator
        ]);
        expected.extend(be(1 << 56));
        assert_eq!(key.encode(), expected);
    }

    #[test]
    fn golden_index_multi_key_leads_with_subspace_and_index() {
        let key = Key::Index(IndexKey::Multi {
            index_id: 1,
            key: canon(&[IndexKeyValue::Int {
                value: 7,
                width: IntWidth::I64,
            }]),
            row_id: 3,
        });
        let bytes = key.encode();
        let mut prefix = vec![0x07, 0x03];
        prefix.extend(be(1));
        assert!(bytes.starts_with(&prefix), "{bytes:?}");
        // The row id is the final component, appended after the value.
        assert!(bytes.ends_with(&be(3)), "{bytes:?}");
    }

    #[test]
    fn index_composite_framing_distinct_at_key_level() {
        let ab_c = Key::Index(IndexKey::Unique {
            index_id: 1,
            key: canon(&[
                IndexKeyValue::Str("ab".into()),
                IndexKeyValue::Str("c".into()),
            ]),
        });
        let a_bc = Key::Index(IndexKey::Unique {
            index_id: 1,
            key: canon(&[
                IndexKeyValue::Str("a".into()),
                IndexKeyValue::Str("bc".into()),
            ]),
        });
        assert_ne!(ab_c.encode(), a_bc.encode());
    }

    /// The hand-written index tags match what the derive encodes.
    #[test]
    fn index_tags_match_the_encoded_discriminants() {
        let unique = Key::Index(IndexKey::Unique {
            index_id: 9,
            key: CanonicalKey::empty(),
        })
        .encode();
        let multi = Key::Index(IndexKey::Multi {
            index_id: 9,
            key: CanonicalKey::empty(),
            row_id: 0,
        })
        .encode();

        assert_eq!(unique[0], INDEX_SUBSPACE_TAG);
        assert_eq!(multi[0], INDEX_SUBSPACE_TAG);
        assert_eq!(unique[1], INDEX_UNIQUE_TAG);
        assert_eq!(multi[1], INDEX_MULTI_TAG);
        assert_eq!(index_kind_tag(true), INDEX_UNIQUE_TAG);
        assert_eq!(index_kind_tag(false), INDEX_MULTI_TAG);
        assert_eq!(&unique[2..INDEX_ENTRY_PREFIX_LEN], &9_u64.to_be_bytes());
    }

    #[test]
    fn index_index_prefix_covers_one_index_and_kind() {
        let value = canon(&[IndexKeyValue::Str("v".into())]);

        let unique = Key::Index(IndexKey::Unique {
            index_id: 4,
            key: value.clone(),
        })
        .encode();
        assert!(unique.starts_with(&index_index_prefix(IndexKind::Unique, 4)));
        // A different index id, and the other kind, do not match.
        assert!(!unique.starts_with(&index_index_prefix(IndexKind::Unique, 5)));
        assert!(!unique.starts_with(&index_index_prefix(IndexKind::Multi, 4)));

        let multi = Key::Index(IndexKey::Multi {
            index_id: 4,
            key: value,
            row_id: 1,
        })
        .encode();
        assert!(multi.starts_with(&index_index_prefix(IndexKind::Multi, 4)));
        assert!(!multi.starts_with(&index_index_prefix(IndexKind::Unique, 4)));
    }

    /// The orphan sweep seeks past a live index by scanning from
    /// `index_index_prefix(kind, id + 1)`, which is only sound if index ids
    /// sort ascending within a kind and every index prefix carries the
    /// kind prefix.
    #[test]
    fn index_index_prefixes_sort_ascending_within_a_kind() {
        for kind in [IndexKind::Unique, IndexKind::Multi] {
            let kind_prefix = index_kind_prefix(kind);
            let mut previous = index_index_prefix(kind, 0);
            assert!(previous.starts_with(&kind_prefix));

            for index_id in [1, 2, 255, 256, u64::MAX - 1, u64::MAX] {
                let prefix = index_index_prefix(kind, index_id);
                assert!(prefix.starts_with(&kind_prefix));
                assert!(
                    prefix > previous,
                    "index {index_id} must sort after its predecessor in {kind:?}"
                );
                previous = prefix;
            }
        }

        // The two kinds occupy disjoint ranges, so a per-kind scan never
        // walks into the other.
        assert_ne!(
            index_kind_prefix(IndexKind::Unique),
            index_kind_prefix(IndexKind::Multi)
        );
    }

    #[test]
    fn index_multi_value_prefix_covers_all_row_ids_of_one_value() {
        let value = canon(&[IndexKeyValue::Str("shared".into())]);
        let prefix = index_multi_value_prefix(9, &value);
        for row_id in [0, 1, u64::MAX] {
            let key = Key::Index(IndexKey::Multi {
                index_id: 9,
                key: value.clone(),
                row_id,
            });
            assert!(key.encode().starts_with(&prefix), "row_id {row_id}");
        }
        // A different value, and a different index, do not match.
        let other_value = canon(&[IndexKeyValue::Str("different".into())]);
        assert!(
            !Key::Index(IndexKey::Multi {
                index_id: 9,
                key: other_value,
                row_id: 0,
            })
            .encode()
            .starts_with(&prefix)
        );
        assert!(
            !Key::Index(IndexKey::Multi {
                index_id: 10,
                key: value,
                row_id: 0,
            })
            .encode()
            .starts_with(&prefix)
        );
    }

    #[test]
    fn index_keys_roundtrip() {
        let keys = [
            Key::Index(IndexKey::Unique {
                index_id: 5,
                key: canon(&[IndexKeyValue::Str("hello".into())]),
            }),
            Key::Index(IndexKey::Multi {
                index_id: 9,
                key: canon(&[
                    IndexKeyValue::Int {
                        value: -3,
                        width: IntWidth::I32,
                    },
                    IndexKeyValue::Bool(true),
                ]),
                row_id: 42,
            }),
        ];
        for key in keys {
            let decoded = Key::decode(&key.encode()).unwrap();
            assert_eq!(decoded, key);
        }
    }

    #[test]
    fn decode_roundtrips_representative_keys() {
        let keys = [
            Key::Sys(SysKey::Head),
            Key::Snapshot {
                snapshot_id: u64::MAX,
            },
            Key::current(EntityKey::Column {
                table_id: 3,
                column_id: 4,
            }),
            Key::history(
                EntityKey::Option {
                    scope_kind: 2,
                    scope_id: 3,
                },
                8,
            ),
            Key::Current(CurrentKey::GcFile { data_file_id: 0 }),
            Key::Inline(InlineKey::Live(InlineOperation::FileDelete {
                table_id: 1,
                data_file_id: 2,
                row_id: 3,
            })),
            Key::Inline(InlineKey::Arch(InlineOperation::FileDelete {
                table_id: 1,
                data_file_id: 2,
                row_id: 3,
            })),
            Key::Inline(InlineKey::FileDeleteTable { table_id: 1 }),
        ];
        for key in keys {
            let decoded = Key::decode(&key.encode()).unwrap();
            assert_eq!(decoded, key);
        }
    }

    #[test]
    fn decode_rejects_unknown_discriminants_and_bad_arity() {
        // Unknown subspace discriminant (below and above the valid range).
        assert!(Key::decode(&[0x00, 0x02]).is_err());
        assert!(Key::decode(&[0x7f, 0x02]).is_err());
        // Unknown kind within a valid subspace.
        assert!(Key::decode(&[0x02, 0x7f]).is_err());
        assert!(Key::decode(&[0x06, 0x7f]).is_err());
        // Truncated components.
        let mut short = Key::Snapshot { snapshot_id: 1 }.encode();
        short.pop();
        assert!(Key::decode(&short).is_err());
        // Trailing bytes.
        let mut long = Key::Sys(SysKey::Head).encode();
        long.push(0);
        assert!(Key::decode(&long).is_err());
        // Empty.
        assert!(Key::decode(&[]).is_err());
    }

    // Prefixes: derived by encode-and-truncate, so they are byte prefixes
    // of full keys by construction — these tests pin the truncation
    // lengths against the structure.

    #[test]
    fn prefixes_are_byte_prefixes_of_full_keys() {
        let key = Key::current(EntityKey::File {
            table_id: 1,
            data_file_id: 2,
        });
        let bytes = key.encode();
        assert!(bytes.starts_with(&subspace_prefix(Subspace::Current)));
        assert!(bytes.starts_with(&current_table_prefix(TableScopedKind::File, 1)));
        // A different table's prefix must not match.
        assert!(!bytes.starts_with(&current_table_prefix(TableScopedKind::File, 2)));
        // A different kind's prefix must not match.
        assert!(!bytes.starts_with(&current_table_prefix(TableScopedKind::DeleteFile, 1)));
    }

    #[test]
    fn entity_kind_prefixes_select_only_the_requested_kind() {
        let current = Key::current(EntityKey::File {
            table_id: 1,
            data_file_id: 2,
        })
        .encode();
        let history = Key::history(
            EntityKey::File {
                table_id: 1,
                data_file_id: 2,
            },
            3,
        )
        .encode();

        assert!(current.starts_with(&current_entity_kind_prefix(EntityKind::File)));
        assert!(history.starts_with(&history_entity_kind_prefix(EntityKind::File)));
        assert!(!current.starts_with(&current_entity_kind_prefix(EntityKind::Table)));
        assert!(!history.starts_with(&history_entity_kind_prefix(EntityKind::Table)));
    }

    /// Every table-scoped kind's prefix matches a live key of that kind
    /// with the same table id and rejects a different table id.
    #[test]
    fn table_scoped_prefixes_cover_all_kinds() {
        let kinds = [
            TableScopedKind::Column,
            TableScopedKind::Partition,
            TableScopedKind::File,
            TableScopedKind::DeleteFile,
            TableScopedKind::FileColumnStats,
            TableScopedKind::TableStats,
            TableScopedKind::TableColumnStats,
            TableScopedKind::Sort,
        ];
        for kind in kinds {
            let key = Key::current(kind.sample(9));
            let bytes = key.encode();
            assert!(
                bytes.starts_with(&current_table_prefix(kind, 9)),
                "{kind:?} prefix must match its own key"
            );
            assert!(
                !bytes.starts_with(&current_table_prefix(kind, 10)),
                "{kind:?} prefix must not match another table"
            );
        }
    }

    /// Every inline-op kind's live prefix matches a live key of that kind
    /// with the same table id and rejects a different table id or kind.
    #[test]
    fn inline_live_table_prefixes_cover_all_operation_kinds() {
        let kinds = [
            InlineOperationKind::Insert,
            InlineOperationKind::InlineDelete,
            InlineOperationKind::FileDelete,
        ];
        for kind in kinds {
            let key = Key::Inline(InlineKey::Live(kind.sample(9)));
            let bytes = key.encode();
            assert!(
                bytes.starts_with(&inline_live_table_prefix(kind, 9)),
                "{kind:?} prefix must match its own key"
            );
            assert!(
                !bytes.starts_with(&inline_live_table_prefix(kind, 10)),
                "{kind:?} prefix must not match another table"
            );
        }
        // Different op kinds must not share a prefix even for the same table.
        assert!(
            !inline_live_table_prefix(InlineOperationKind::InlineDelete, 9)
                .starts_with(&inline_live_table_prefix(InlineOperationKind::Insert, 9))
        );
        // An archived key never matches a live prefix.
        let arch = Key::Inline(InlineKey::Arch(InlineOperationKind::Insert.sample(9))).encode();
        assert!(!arch.starts_with(&inline_live_table_prefix(InlineOperationKind::Insert, 9)));
    }

    /// The `inline/schema` prefix matches every schema version of the
    /// same table and rejects a different table.
    #[test]
    fn inline_schema_table_prefix_covers_all_versions() {
        for schema_version in [0, 1, 42] {
            let key = Key::Inline(InlineKey::Schema {
                table_id: 9,
                schema_version,
            });
            assert!(key.encode().starts_with(&inline_schema_table_prefix(9)));
        }
        let other_table = Key::Inline(InlineKey::Schema {
            table_id: 10,
            schema_version: 0,
        });
        assert!(
            !other_table
                .encode()
                .starts_with(&inline_schema_table_prefix(9))
        );
    }

    /// The dropped-schema prefixes match every version of the same table
    /// and never a live `inline/schema` key, so subtracting one scan from
    /// the other cannot confuse the two.
    #[test]
    fn inline_schema_dropped_prefixes_are_disjoint_from_live_schemas() {
        for schema_version in [0, 1, 42] {
            let key = Key::Inline(InlineKey::SchemaDropped {
                table_id: 9,
                schema_version,
            });
            assert!(
                key.encode()
                    .starts_with(&inline_schema_dropped_table_prefix(9))
            );
            assert!(key.encode().starts_with(&inline_schema_dropped_prefix()));
            assert!(!key.encode().starts_with(&inline_schema_prefix()));
        }

        let live = Key::Inline(InlineKey::Schema {
            table_id: 9,
            schema_version: 0,
        });
        assert!(!live.encode().starts_with(&inline_schema_dropped_prefix()));

        let other_table = Key::Inline(InlineKey::SchemaDropped {
            table_id: 10,
            schema_version: 0,
        });
        assert!(
            !other_table
                .encode()
                .starts_with(&inline_schema_dropped_table_prefix(9))
        );
    }

    // Property tests: roundtrip, order preservation, uniqueness.

    fn arb_entity() -> impl Strategy<Value = EntityKey> {
        prop_oneof![
            any::<u64>().prop_map(|schema_id| EntityKey::Schema { schema_id }),
            any::<u64>().prop_map(|table_id| EntityKey::Table { table_id }),
            any::<u64>().prop_map(|view_id| EntityKey::View { view_id }),
            any::<(u64, u64)>().prop_map(|(table_id, column_id)| EntityKey::Column {
                table_id,
                column_id
            }),
            any::<(u64, u64)>().prop_map(|(table_id, partition_id)| EntityKey::Partition {
                table_id,
                partition_id
            }),
            any::<(u64, u64)>().prop_map(|(table_id, data_file_id)| EntityKey::File {
                table_id,
                data_file_id
            }),
            any::<(u64, u64)>().prop_map(|(table_id, delete_file_id)| EntityKey::DeleteFile {
                table_id,
                delete_file_id
            }),
            any::<(u64, u64, u64)>().prop_map(|(table_id, data_file_id, column_id)| {
                EntityKey::FileColumnStats {
                    table_id,
                    data_file_id,
                    column_id,
                }
            }),
            any::<u64>().prop_map(|table_id| EntityKey::TableStats { table_id }),
            any::<(u64, u64)>().prop_map(|(table_id, column_id)| EntityKey::TableColumnStats {
                table_id,
                column_id
            }),
            any::<(u64, u64)>()
                .prop_map(|(table_id, sort_id)| EntityKey::Sort { table_id, sort_id }),
            any::<u64>().prop_map(|object_id| EntityKey::Tag { object_id }),
            any::<(u64, u64)>().prop_map(|(scope_kind, scope_id)| EntityKey::Option {
                scope_kind,
                scope_id
            }),
            any::<u64>().prop_map(|macro_id| EntityKey::Macro { macro_id }),
            any::<(u64, u64)>().prop_map(|(table_id, mapping_id)| EntityKey::Mapping {
                table_id,
                mapping_id
            }),
            any::<(u64, u64)>()
                .prop_map(|(table_id, index_id)| EntityKey::Index { table_id, index_id }),
        ]
    }

    fn arb_inline_op() -> impl Strategy<Value = InlineOperation> {
        prop_oneof![
            any::<(u64, u64, u64, u64)>().prop_map(
                |(table_id, schema_version, begin_snapshot, chunk_seq)| InlineOperation::Insert {
                    table_id,
                    schema_version,
                    begin_snapshot,
                    chunk_seq
                }
            ),
            any::<(u64, u64)>()
                .prop_map(|(table_id, row_id)| InlineOperation::InlineDelete { table_id, row_id }),
            any::<(u64, u64, u64)>().prop_map(|(table_id, data_file_id, row_id)| {
                InlineOperation::FileDelete {
                    table_id,
                    data_file_id,
                    row_id,
                }
            }),
        ]
    }

    fn arb_canonical_key() -> impl Strategy<Value = CanonicalKey> {
        // Arbitrary component byte strings, kept well under the size cap so
        // `encode_key` never rejects.
        proptest::collection::vec(proptest::collection::vec(any::<u8>(), 0..8), 0..4).prop_map(
            |components| {
                let values = components
                    .into_iter()
                    .map(IndexKeyValue::Bytes)
                    .collect::<Vec<_>>();
                encode_key(&values).unwrap()
            },
        )
    }

    fn arb_index() -> impl Strategy<Value = IndexKey> {
        prop_oneof![
            (any::<u64>(), arb_canonical_key())
                .prop_map(|(index_id, key)| IndexKey::Unique { index_id, key }),
            (any::<u64>(), arb_canonical_key(), any::<u64>()).prop_map(
                |(index_id, key, row_id)| {
                    IndexKey::Multi {
                        index_id,
                        key,
                        row_id,
                    }
                }
            ),
        ]
    }

    fn arb_key() -> impl Strategy<Value = Key> {
        prop_oneof![
            Just(Key::Sys(SysKey::Format)),
            Just(Key::Sys(SysKey::Head)),
            Just(Key::Sys(SysKey::Migration)),
            Just(Key::Sys(SysKey::MaintenanceStatus)),
            any::<u64>().prop_map(|snapshot_id| Key::Snapshot { snapshot_id }),
            arb_entity().prop_map(Key::current),
            (arb_entity(), any::<u64>()).prop_map(|(entity, end)| Key::history(entity, end)),
            any::<u64>().prop_map(|data_file_id| Key::Current(CurrentKey::GcFile { data_file_id })),
            any::<(u64, u64)>().prop_map(|(table_id, schema_version)| {
                Key::Inline(InlineKey::Schema {
                    table_id,
                    schema_version,
                })
            }),
            arb_inline_op().prop_map(|op| Key::Inline(InlineKey::Live(op))),
            arb_inline_op().prop_map(|op| Key::Inline(InlineKey::Arch(op))),
            any::<u64>().prop_map(|table_id| Key::Inline(InlineKey::FileDeleteTable { table_id })),
            any::<(u64, u64)>().prop_map(|(table_id, schema_version)| {
                Key::Inline(InlineKey::SchemaDropped {
                    table_id,
                    schema_version,
                })
            }),
            any::<(u64, u64)>().prop_map(|(table_id, row_id_end)| {
                Key::Inline(InlineKey::ChunkRange {
                    table_id,
                    row_id_end,
                })
            }),
            any::<(u64, u64, u64, u64, u64)>().prop_map(
                |(table_id, row_id_end, schema_version, begin_snapshot, chunk_seq)| {
                    Key::Inline(InlineKey::ChunkLocator {
                        table_id,
                        row_id_end,
                        schema_version,
                        begin_snapshot,
                        chunk_seq,
                    })
                },
            ),
            arb_index().prop_map(Key::Index),
            any::<(u64, u64, u64)>().prop_map(|(table_id, row_id, end_snapshot)| {
                Key::Inline(InlineKey::RowTombstone {
                    table_id,
                    row_id,
                    end_snapshot,
                })
            }),
            any::<u64>().prop_map(|snapshot_id| Key::Changelog { snapshot_id }),
        ]
    }

    proptest! {
        #[test]
        fn inline_tombstone_roundtrip_and_table_prefix(
            table_id in any::<u64>(), row_id in any::<u64>(), end_snapshot in any::<u64>(),
        ) {
            let key = Key::Inline(InlineKey::RowTombstone { table_id, row_id, end_snapshot });
            let encoded = key.encode();
            prop_assert_eq!(Key::decode(&encoded).unwrap(), key);
            prop_assert!(encoded.starts_with(&inline_row_tombstone_table_prefix(table_id)));
            prop_assert!(!encoded.starts_with(&inline_live_table_prefix(InlineOperationKind::InlineDelete, table_id)));
        }

        #[test]
        fn borrowed_index_entry_encoding_matches_owned(
            index_id in any::<u64>(),
            key in arb_canonical_key(),
            row_id in any::<u64>(),
            unique in any::<bool>(),
        ) {
            let owned = if unique {
                Key::Index(IndexKey::Unique {
                    index_id,
                    key: key.clone(),
                })
            } else {
                Key::Index(IndexKey::Multi {
                    index_id,
                    key: key.clone(),
                    row_id,
                })
            };
            prop_assert_eq!(
                encode_index_entry(index_id, unique, &key, row_id),
                owned.encode()
            );
        }

        #[test]
        fn roundtrip(key in arb_key()) {
            let decoded = Key::decode(&key.encode()).unwrap();
            prop_assert_eq!(decoded, key);
        }

        #[test]
        fn encoded_keys_are_unique(a in arb_key(), b in arb_key()) {
            if a != b {
                prop_assert_ne!(a.encode(), b.encode());
            }
        }

        /// The encoding preserves the typed order: every prefix scan and
        /// range bound relies on byte order agreeing with key order.
        #[test]
        fn encoding_preserves_order(a in arb_key(), b in arb_key()) {
            prop_assert_eq!(a.cmp(&b), a.encode().cmp(&b.encode()));
        }

        // Decode is total: arbitrary bytes decode or fail as
        // `Corruption`, never panic.
        #[test]
        fn decode_arbitrary_bytes_never_panics(
            bytes in proptest::collection::vec(any::<u8>(), 0..64),
        ) {
            let _ = Key::decode(&bytes);
        }

        // Steer past the subspace discriminant so component parsing
        // sees the garbage too.
        #[test]
        fn decode_garbage_in_valid_subspace_never_panics(
            subspace in 0x02u8..=0x07,
            bytes in proptest::collection::vec(any::<u8>(), 0..64),
        ) {
            let mut encoded = vec![subspace];
            encoded.extend(&bytes);
            let _ = Key::decode(&encoded);
        }

        // One corrupted byte in a valid key decodes to some key or
        // fails loudly, never panics.
        #[test]
        fn decode_corrupted_valid_key_never_panics(
            key in arb_key(),
            index in any::<proptest::sample::Index>(),
            byte in any::<u8>(),
        ) {
            let mut bytes = key.encode();
            let position = index.index(bytes.len());
            bytes[position] = byte;
            let _ = Key::decode(&bytes);
        }
    }

    mod index_value_bounds {
        use proptest::prelude::*;

        use super::*;
        use crate::store::index_encoding::{
            Direction, IndexKeyValue, IntWidth, NullOrder, encode_ordered_values,
        };

        /// Values across the categories whose framing the bound arithmetic has
        /// to survive: fixed-width integers, and the variable-length
        /// strings and blobs whose escaped bodies can end in `0xff`.
        fn arbitrary_value() -> impl Strategy<Value = IndexKeyValue> {
            prop_oneof![
                any::<i64>().prop_map(|value| IndexKeyValue::Int {
                    value: i128::from(value),
                    width: IntWidth::I64,
                }),
                any::<u32>().prop_map(|value| IndexKeyValue::UInt {
                    value: u128::from(value),
                    width: IntWidth::I32,
                }),
                any::<bool>().prop_map(IndexKeyValue::Bool),
                ".{0,8}".prop_map(IndexKeyValue::Str),
                proptest::collection::vec(any::<u8>(), 0..8).prop_map(IndexKeyValue::Bytes),
            ]
        }

        /// A two-column key, or its one-column leading prefix, under one shared
        /// direction. NULLS placement is irrelevant here: every value is
        /// present.
        fn encode(values: &[IndexKeyValue], descending: bool) -> CanonicalKey {
            let direction = if descending {
                Direction::Descending
            } else {
                Direction::Ascending
            };
            encode_ordered_values(
                &values.iter().cloned().map(Some).collect::<Vec<_>>(),
                &[direction, direction],
                &[NullOrder::Last, NullOrder::Last],
            )
            .expect("values within the key size cap encode")
        }

        /// The bytes a stored non-unique entry carries after the index prefix:
        /// the framed value, then the raw row id.
        fn multi_entry(index_id: u64, canon: &CanonicalKey, row_id: u64) -> Vec<u8> {
            let full = Key::Index(IndexKey::Multi {
                index_id,
                key: canon.clone(),
                row_id,
            })
            .encode();
            full[INDEX_ENTRY_PREFIX_LEN..].to_vec()
        }

        proptest! {
            /// The bound above a one-column prefix outranks every key that
            /// extends it: the value's own entry, an entry carrying a further
            /// column, and — on a non-unique index — the trailing row id.
            #[test]
            fn above_a_prefix_outranks_every_extension(
                leading in arbitrary_value(),
                extension in arbitrary_value(),
                row_id: u64,
                descending: bool,
            ) {
                let index_id = 7;
                let prefix = encode(std::slice::from_ref(&leading), descending);
                let extended = encode(&[leading, extension], descending);

                for kind in [IndexKind::Unique, IndexKind::Multi] {
                    let bound = index_value_above(kind, index_id, &prefix)
                        .expect("a framed body is never all 0xff");
                    prop_assert!(bound > index_value_suffix(kind, index_id, &prefix));
                    prop_assert!(bound > index_value_suffix(kind, index_id, &extended));
                }
                prop_assert!(
                    index_value_above(IndexKind::Multi, index_id, &prefix)
                        .expect("a framed body is never all 0xff")
                        > multi_entry(index_id, &extended, row_id)
                );
            }

            /// The bound stops below every value that sorts above the prefix, so
            /// a range ending there admits no row it should exclude.
            #[test]
            fn above_a_prefix_stops_below_a_larger_value(
                one in arbitrary_value(),
                two in arbitrary_value(),
                descending: bool,
            ) {
                let index_id = 7;
                let kind = IndexKind::Unique;
                let first = encode(&[one], descending);
                let second = encode(&[two], descending);

                let (lower, higher) = if index_value_suffix(kind, index_id, &first)
                    < index_value_suffix(kind, index_id, &second)
                {
                    (&first, &second)
                } else {
                    (&second, &first)
                };
                prop_assume!(lower != higher);

                let bound = index_value_above(kind, index_id, lower)
                    .expect("a framed body is never all 0xff");
                prop_assert!(bound <= index_value_suffix(kind, index_id, higher));
            }
        }
    }

    fn arbitrary_bound() -> impl Strategy<Value = Bound<Vec<u8>>> {
        prop_oneof![
            Just(Bound::Unbounded),
            prop::collection::vec(any::<u8>(), 0..20).prop_map(Bound::Included),
            prop::collection::vec(any::<u8>(), 0..20).prop_map(Bound::Excluded),
        ]
    }

    proptest! {
        /// A suffix inside the outer range falls in exactly one split range
        /// and one outside it in none: the ranges are disjoint and their
        /// union is exactly the outer range.
        #[test]
        fn split_ranges_partition_the_outer_range(
            lower in arbitrary_bound(),
            upper in arbitrary_bound(),
            points in prop::collection::vec(prop::collection::vec(any::<u8>(), 0..20), 0..12),
            suffix in prop::collection::vec(any::<u8>(), 0..24),
        ) {
            let ranges = split_between(lower.clone(), upper.clone(), points.clone());
            let holders = ranges
                .iter()
                .filter(|range| within(&range.0, &range.1, &suffix))
                .count();
            let expected = usize::from(within(&lower, &upper, &suffix));
            prop_assert_eq!(holders, expected, "{:?} in {:?}", suffix, ranges);

            // Every point strictly inside the outer range starts a range.
            for point in &points {
                if within(&lower, &upper, point) && lower != Bound::Included(point.clone()) {
                    prop_assert!(ranges.iter().any(|range| range.0 == Bound::Included(point.clone())));
                }
            }
        }

        /// The interpolated points sit strictly between the extremes and in
        /// key order, so a split cut at them keeps every key between the
        /// extremes and never yields an out-of-order range.
        #[test]
        fn even_points_lie_between_the_extremes(
            low in prop::collection::vec(any::<u8>(), 16..24),
            high in prop::collection::vec(any::<u8>(), 16..24),
            parts in 1_usize..12,
        ) {
            let (low, high) = if low <= high { (low, high) } else { (high, low) };
            let points = even_points_between(&low, &high, parts);
            prop_assert!(points.len() < parts);
            prop_assert!(points.windows(2).all(|pair| pair[0] < pair[1]));
            for point in &points {
                prop_assert!(point.as_slice() > low.as_slice(), "{:?} <= {:?}", point, low);
                prop_assert!(point.as_slice() < high.as_slice(), "{:?} >= {:?}", point, high);
            }
        }
    }

    /// Ids clustered near zero still split: the cut interpolates between the
    /// extremes present, not across the whole id space.
    #[test]
    fn even_points_split_between_the_extremes_present() {
        let low = 7_u64
            .to_be_bytes()
            .iter()
            .chain(&0_u64.to_be_bytes())
            .copied()
            .collect::<Vec<_>>();
        let high = 7_u64
            .to_be_bytes()
            .iter()
            .chain(&4_000_u64.to_be_bytes())
            .copied()
            .collect::<Vec<_>>();
        let points = even_points_between(&low, &high, 4);
        assert_eq!(points.len(), 3);
        assert_eq!(&points[0][..8], &7_u64.to_be_bytes());
        assert_eq!(&points[0][8..], &1_000_u64.to_be_bytes());
        assert_eq!(&points[2][8..], &3_000_u64.to_be_bytes());

        assert!(even_points_between(&low, &low, 4).is_empty());
        assert_eq!(
            split_between(Bound::Unbounded, Bound::Unbounded, Vec::new()),
            vec![(Bound::Unbounded, Bound::Unbounded)]
        );
    }
}