moraine 0.1.0

//! 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 storekey::{Decode, Encode};

use crate::error::{Error, Result};

/// 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. The five variants are the five
/// subspaces; each is also a SlateDB segment (the store is created with a
/// one-byte segment extractor over the leading discriminant).
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Encode, Decode)]
pub(crate) enum Key {
    /// Store-level singletons: format version, head pointer, migration
    /// marker. 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),
}

/// 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,
}

/// 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 — the row's identity in DuckLake's own schema (inserts
    /// carry it, cleanup deletes by it), unique because a file's catalog
    /// rows are removed in the same transaction that schedules it.
    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,
    },
}

/// An inlined-data key: the per-schema-version Arrow schema, a live
/// record, or the archived (post-flush) form of a live record. `Live` and
/// `Arch` share [`InlineOp`], 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),
}

/// 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,
    },
    /// Tombstone for an inlined insert row.
    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}")))
    }
}

/// 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 keys are read/written by exact key today; no caller prefix-scans
    // the whole subspace yet.
    #[allow(dead_code)]
    System,
    /// Snapshot records.
    Snapshot,
    /// Live catalog state.
    Current,
    /// Ended entity versions.
    History,
    /// Inlined data.
    // Data inlining is not implemented yet.
    #[allow(dead_code)]
    Inline,
}

impl Subspace {
    /// 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,
            }),
        }
    }
}

/// The kinds whose live keys are scoped `table_id`-first — the only kinds
/// [`current_table_prefix`] accepts.
// No caller needs "everything about table T" yet (cascading table drop and
// per-table GC land in later slices); the prefix math is pinned by tests.
#[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,
}

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,
            },
        }
    }
}

/// Discriminant bytes preceding an entity's components in a live key:
/// subspace, `CurrentKey::Entity`, entity kind.
// Only `current_table_prefix` consumes this, and it has no caller yet.
#[allow(dead_code)]
const CUR_KIND_PREFIX_LEN: usize = 3;

/// 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;

/// 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> {
    let mut bytes = Key::Inline(InlineKey::Live(kind.sample(table_id))).encode();
    bytes.truncate(INLINE_LIVE_KIND_PREFIX_LEN + size_of::<u64>());
    bytes
}

/// 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> {
    let mut bytes = Key::Inline(InlineKey::Schema {
        table_id,
        schema_version: 0,
    })
    .encode();
    bytes.truncate(INLINE_SCHEMA_PREFIX_LEN + size_of::<u64>());
    bytes
}

/// Byte prefix of every `inline/schema` key, across every table.
pub(crate) fn inline_schema_prefix() -> Vec<u8> {
    let mut bytes = Key::Inline(InlineKey::Schema {
        table_id: 0,
        schema_version: 0,
    })
    .encode();
    bytes.truncate(INLINE_SCHEMA_PREFIX_LEN);
    bytes
}

/// 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> {
    let mut bytes = subspace.sample().encode();
    bytes.truncate(TAG_PREFIX_LEN);
    bytes
}

/// Byte prefix of every live key of `kind` scoped to `table_id`.
// No caller yet; pinned by the prefix tests below.
#[allow(dead_code)]
pub(crate) fn current_table_prefix(kind: TableScopedKind, table_id: u64) -> Vec<u8> {
    let mut bytes = Key::current(kind.sample(table_id)).encode();
    bytes.truncate(CUR_KIND_PREFIX_LEN + size_of::<u64>());
    bytes
}

#[cfg(test)]
mod tests {
    use proptest::prelude::*;

    use super::*;

    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]);
    }

    #[test]
    fn golden_snapshot_key() {
        let mut expect = vec![0x03];
        expect.extend(be(42));
        assert_eq!(Key::Snapshot { 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_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_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);
        }
    }

    // Decode: exact inverse, loud failure.

    #[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,
            })),
        ];
        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)));
    }

    /// 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))
        );
    }

    // 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
            }),
        ]
    }

    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_key() -> impl Strategy<Value = Key> {
        prop_oneof![
            Just(Key::Sys(SysKey::Format)),
            Just(Key::Sys(SysKey::Head)),
            Just(Key::Sys(SysKey::Migration)),
            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))),
        ]
    }

    proptest! {
        #[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..=0x06,
            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);
        }
    }
}