moraine 0.2.0

//! The staged-row commit path: DuckLake authors rows over the ABI instead
//! of driving [`crate::transaction::Transaction`]'s verbs, and this module
//! turns the accumulated rows into one atomic store write.
//!
//! Three rules bound this path: every value DuckLake supplies is stored
//! **verbatim** except one interpreted convention (an `UPDATE` setting
//! `end_snapshot` becomes current-delete + history-write); one commit is one
//! atomic batch, reusing [`super::commit::diff_writes`]; and a lost race at
//! commit is **never retried** — DuckLake authored the ids and counters in
//! the batch. The loser's error always carries the substring `conflict`
//! (via [`Error::CommitConflict`]'s `Display`), the wire contract
//! DuckLake's retry loop scans for.
//!
//! Translation applies every staged row to a cloned working
//! [`CatalogSnapshot`], then diffs it against the unmodified base exactly
//! as a verb-path commit diffs its closure's output. An `UPDATE ... SET
//! end_snapshot` row's value is validated, not trusted: DuckLake always
//! sets it to this commit's own new snapshot id — the value `diff_writes`
//! stamps on its own — so a mismatch is drift caught loudly.

use std::{collections::HashMap, sync::Arc};

use object_store::ObjectStore;
use slatedb::DbTransaction;

use crate::{
    catalog::{
        CatalogSnapshot, ColumnInfo, IndexInfo, SnapshotId, TableId,
        inline::{InlineScanKind, materialize_inline_rows},
        projection::{ProjectionCache, fold_committed_batch},
        scoped_read::{self, ScopedReadEntry},
    },
    error::{Error, Result},
    store::{
        handle::ReadHandle,
        index_encoding::encode_key,
        inline as store_inline,
        key::{EntityKey, InlineKey, InlineOperation, Key, SysKey},
        proto, read, value,
    },
    transaction::{
        commit,
        index_maintenance::{StagedIndexEntry, stage_index_entries},
    },
};

/// Which `ducklake_*` table a staged row targets. `Snapshot`,
/// `SnapshotChanges`, and `SchemaVersions` all fold into one moraine
/// `snapshot` record; every other kind maps to one `EntityKey` variant or one
/// of the three unversioned statistics kinds.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum TableKind {
    /// `ducklake_snapshot`.
    Snapshot,
    /// `ducklake_snapshot_changes`.
    SnapshotChanges,
    /// `ducklake_schema`.
    Schema,
    /// `ducklake_table`.
    Table,
    /// `ducklake_view`.
    View,
    /// `ducklake_column`.
    Column,
    /// `ducklake_data_file`.
    DataFile,
    /// `ducklake_delete_file`.
    DeleteFile,
    /// `ducklake_table_stats`.
    TableStats,
    /// `ducklake_table_column_stats`.
    TableColumnStats,
    /// `ducklake_file_column_stats`.
    FileColumnStats,
    /// `ducklake_schema_versions`: per-table schema-change history, one
    /// `(begin_snapshot, schema_version, table_id)` row per
    /// created-or-schema-altered table per commit. The first two values
    /// are always the committing snapshot's own id and `schema_version`,
    /// so the table-id set is the only new information — folded into the
    /// snapshot record's `schema_changed_table_ids`, with both redundant values
    /// validated against the `ducklake_snapshot` row in the same batch.
    SchemaVersions,
    /// `ducklake_partition_info`.
    PartitionInfo,
    /// `ducklake_partition_column` — folded into its spec's record.
    PartitionColumn,
    /// `ducklake_file_partition_value` — folded into its file's record.
    FilePartitionValue,
    /// `ducklake_sort_info`.
    SortInfo,
    /// `ducklake_sort_expression` — folded into its spec's record.
    SortExpression,
    /// `ducklake_tag` — an entry in its object's container record.
    Tag,
    /// `ducklake_column_tag` — an entry embedded in its column's record.
    ColumnTag,
    /// `ducklake_files_scheduled_for_deletion` — the physical-deletion
    /// schedule, keyed by the scheduled file's id.
    FilesScheduledForDeletion,
    /// `ducklake_macro`.
    Macro,
    /// `ducklake_macro_impl` — folded into its macro's record.
    MacroImpl,
    /// `ducklake_macro_parameters` — folded into its macro's record.
    MacroParameters,
    /// `ducklake_column_mapping`.
    ColumnMapping,
    /// `ducklake_name_mapping` — folded into its mapping's record.
    NameMapping,
}

/// One column value in a staged row, typed to the small set of primitive
/// kinds every `ducklake_*` column uses. Decoding into the right proto
/// shape happens here, table-kind by table-kind.
#[derive(Debug, Clone, PartialEq)]
pub enum Cell {
    /// SQL `NULL`.
    Null,
    /// An unsigned integer column (every id, counter, and count).
    U64(u64),
    /// A signed integer column (currently only `snapshot_time`, a
    /// `TIMESTAMPTZ` carried as microseconds since the epoch).
    I64(i64),
    /// A boolean column.
    Bool(bool),
    /// A text column (also used for `UUID`, carried as its text form).
    Str(String),
}

/// One staged row mutation, DuckLake-authored. `cells` are positional, in
/// the exact `ducklake_*` column order pinned in
/// `crates/moraine-duckdb/cpp/metadata_tables.cpp`.
#[derive(Debug, Clone)]
pub enum RowOperation {
    /// A new row: becomes a live `current` record (versioned kinds) or
    /// overwrites the current record in place (unversioned statistics
    /// kinds).
    Insert {
        /// The row's table.
        table: TableKind,
        /// The row's column values, in table order.
        cells: Vec<Cell>,
    },
    /// A row removed with no history mirror: the unversioned statistics
    /// kinds and the deletion schedule, plus the hard prunes maintenance
    /// issues — snapshot records, dead entity versions (`current` or
    /// `history`, named by their `end_snapshot`), and dead tag entries.
    Delete {
        /// The row's table.
        table: TableKind,
        /// The removed row's key columns, in table order (id columns
        /// only — see [`Self::Insert`]'s `cells` for the full row shape).
        cells: Vec<Cell>,
    },
    /// An `UPDATE ... SET end_snapshot = <v>` row: the one lifecycle
    /// convention this path interprets — ends the live version (moves it
    /// to `history`). Defined only for the six versioned kinds.
    UpdateSetEnd {
        /// The row's table.
        table: TableKind,
        /// The ended row's key columns, in table order, followed by the
        /// new `end_snapshot` value.
        cells: Vec<Cell>,
    },
    /// An `UPDATE ... SET begin_snapshot = <v>` row: rebases a data
    /// file's visibility window in place. DuckLake issues it only during
    /// a delete-rewrite, against the replacement file the same
    /// transaction just inserted — any other target is a shape error.
    UpdateSetBegin {
        /// The row's table (only `ducklake_data_file`).
        table: TableKind,
        /// The rebased row's key columns, followed by the new
        /// `begin_snapshot` value.
        cells: Vec<Cell>,
    },
    /// `inline/schema`: the Arrow IPC schema for one `(table_id,
    /// schema_version)`, written once at inline-table creation, stored
    /// verbatim.
    InlineSchema {
        /// Owning table.
        table_id: u64,
        /// Schema version the layout is pinned to.
        schema_version: u64,
        /// The Arrow IPC schema message, verbatim.
        arrow_schema: Vec<u8>,
    },
    /// `inline/insert`: one Arrow record-batch chunk of inlined rows.
    /// `chunk_seq` is not carried here — translation allocates it, so
    /// several `InlineInsert`s staged in one commit against the same
    /// `(table_id, schema_version, begin_snapshot)` land as sequential
    /// chunks in stage order.
    InlineInsert {
        /// Owning table.
        table_id: u64,
        /// Schema version the chunk was written under.
        schema_version: u64,
        /// Commit snapshot of the insert, DuckLake-authored verbatim.
        begin_snapshot: u64,
        /// The chunk's first row id; later rows are dense from here.
        row_id_start: u64,
        /// Row count carried by `arrow_body`.
        row_count: u64,
        /// The user-column cells, encoded as one Arrow IPC record-batch
        /// body (opaque bytes to this layer).
        arrow_body: Vec<u8>,
    },
    /// `inline/inline_delete`: tombstones one inlined-insert row.
    InlineInlineDelete {
        /// Owning table.
        table_id: u64,
        /// The tombstoned row.
        row_id: u64,
        /// The commit snapshot the row ends at.
        end_snapshot: u64,
    },
    /// `inline/file_delete`: an inlined delete against a Parquet-file row.
    InlineFileDelete {
        /// Owning table.
        table_id: u64,
        /// Targeted data file.
        data_file_id: u64,
        /// Deleted row.
        row_id: u64,
        /// The commit snapshot the delete takes effect at.
        begin_snapshot: u64,
    },
    /// Removes every `inline/insert` chunk begun at or before
    /// `flush_snapshot` for `(table_id, schema_version)`, plus the
    /// `inline/inline_delete` tombstones those chunks' rows consumed — the
    /// flushed data survives only as the backdated `ducklake_data_file`
    /// DuckLake registers through the ordinary file path.
    InlineFlushDelete {
        /// Owning table.
        table_id: u64,
        /// Schema version being flushed.
        schema_version: u64,
        /// Chunks begun at or before this snapshot are flushed.
        flush_snapshot: u64,
    },
    /// Removes every `inline/*` record for `table_id`: schema, chunks,
    /// and tombstones.
    InlineDrop {
        /// The dropped table.
        table_id: u64,
    },
    /// Removes only the `inline/schema` record for one `(table_id,
    /// schema_version)` — the superseded-schema-version cleanup a flush
    /// issues once its chunks are gone, leaving any other schema
    /// version's `inline/*` records (a newer version accumulating
    /// concurrently) untouched. Distinct from [`Self::InlineDrop`], which
    /// is table-wide (the whole-table `DROP TABLE` cascade).
    InlineSchemaDrop {
        /// Owning table.
        table_id: u64,
        /// The schema version deregistered.
        schema_version: u64,
    },
}

/// A malformed staged row: wrong cell count or a cell of the wrong kind
/// for its column. Never produced by a correct shim translation; this
/// path fails loudly rather than guessing.
fn corrupt_row(table: TableKind, detail: impl std::fmt::Display) -> Error {
    Error::Corruption(format!("staged row for {table:?}: {detail}"))
}

/// A positional reader over one staged row's cells, typed cell-by-cell
/// against the column each `ducklake_*` table declares.
struct Cursor<'a> {
    table: TableKind,
    items: std::slice::Iter<'a, Cell>,
}

impl<'a> Cursor<'a> {
    fn new(table: TableKind, cells: &'a [Cell]) -> Self {
        Self {
            table,
            items: cells.iter(),
        }
    }

    fn next(&mut self) -> Result<&'a Cell> {
        self.items
            .next()
            .ok_or_else(|| corrupt_row(self.table, "too few cells"))
    }

    fn u64(&mut self) -> Result<u64> {
        match self.next()? {
            Cell::U64(v) => Ok(*v),
            other => Err(corrupt_row(
                self.table,
                format!("expected u64, got {other:?}"),
            )),
        }
    }

    fn opt_u64(&mut self) -> Result<Option<u64>> {
        match self.next()? {
            Cell::Null => Ok(None),
            Cell::U64(v) => Ok(Some(*v)),
            other => Err(corrupt_row(
                self.table,
                format!("expected optional u64, got {other:?}"),
            )),
        }
    }

    fn i64(&mut self) -> Result<i64> {
        match self.next()? {
            Cell::I64(v) => Ok(*v),
            other => Err(corrupt_row(
                self.table,
                format!("expected i64, got {other:?}"),
            )),
        }
    }

    fn bool(&mut self) -> Result<bool> {
        match self.next()? {
            Cell::Bool(v) => Ok(*v),
            other => Err(corrupt_row(
                self.table,
                format!("expected bool, got {other:?}"),
            )),
        }
    }

    fn opt_bool(&mut self) -> Result<Option<bool>> {
        match self.next()? {
            Cell::Null => Ok(None),
            Cell::Bool(v) => Ok(Some(*v)),
            other => Err(corrupt_row(
                self.table,
                format!("expected optional bool, got {other:?}"),
            )),
        }
    }

    fn string(&mut self) -> Result<String> {
        match self.next()? {
            Cell::Str(v) => Ok(v.clone()),
            other => Err(corrupt_row(
                self.table,
                format!("expected string, got {other:?}"),
            )),
        }
    }

    fn opt_string(&mut self) -> Result<Option<String>> {
        match self.next()? {
            Cell::Null => Ok(None),
            Cell::Str(v) => Ok(Some(v.clone())),
            other => Err(corrupt_row(
                self.table,
                format!("expected optional string, got {other:?}"),
            )),
        }
    }

    /// Confirms every cell was consumed — a row longer than its table's
    /// column list is as much a shape mismatch as one too short.
    fn finish(mut self) -> Result<()> {
        if self.items.next().is_some() {
            Err(corrupt_row(self.table, "too many cells"))
        } else {
            Ok(())
        }
    }
}

fn decode_schema(cells: &[Cell]) -> Result<proto::SchemaValue> {
    let mut c = Cursor::new(TableKind::Schema, cells);
    let value = proto::SchemaValue {
        schema_id: c.u64()?,
        schema_uuid: c.string()?,
        begin_snapshot: c.u64()?,
        end_snapshot: c.opt_u64()?,
        schema_name: c.string()?,
        path: c.string()?,
        path_is_relative: c.bool()?,
    };
    c.finish()?;
    Ok(value)
}

/// `ducklake_table`'s row shape, minus `next_column_id` — moraine-internal
/// bookkeeping DuckLake never authors (see [`table_value`]).
struct TableCells {
    table_id: u64,
    table_uuid: String,
    begin_snapshot: u64,
    end_snapshot: Option<u64>,
    schema_id: u64,
    table_name: String,
    path: String,
    path_is_relative: bool,
}

fn decode_table(cells: &[Cell]) -> Result<TableCells> {
    let mut c = Cursor::new(TableKind::Table, cells);
    let value = TableCells {
        table_id: c.u64()?,
        table_uuid: c.string()?,
        begin_snapshot: c.u64()?,
        end_snapshot: c.opt_u64()?,
        schema_id: c.u64()?,
        table_name: c.string()?,
        path: c.string()?,
        path_is_relative: c.bool()?,
    };
    c.finish()?;
    Ok(value)
}

/// Builds the full `TableValue`: `next_column_id` is moraine-internal
/// per-table field-id bookkeeping, not a DuckLake column. Carried forward
/// from the table's prior version in `base` when one exists (floor of 1
/// for a brand-new id).
fn table_value(base: &CatalogSnapshot, cells: TableCells) -> proto::TableValue {
    let next_column_id = base
        .tables
        .get(&cells.table_id)
        .map_or(1, |t| t.next_column_id);
    proto::TableValue {
        table_id: cells.table_id,
        table_uuid: cells.table_uuid,
        begin_snapshot: cells.begin_snapshot,
        end_snapshot: cells.end_snapshot,
        schema_id: cells.schema_id,
        table_name: cells.table_name,
        path: cells.path,
        path_is_relative: cells.path_is_relative,
        next_column_id,
    }
}

fn decode_view(cells: &[Cell]) -> Result<proto::ViewValue> {
    let mut c = Cursor::new(TableKind::View, cells);
    let value = proto::ViewValue {
        view_id: c.u64()?,
        view_uuid: c.string()?,
        begin_snapshot: c.u64()?,
        end_snapshot: c.opt_u64()?,
        schema_id: c.u64()?,
        view_name: c.string()?,
        dialect: c.string()?,
        sql: c.string()?,
        column_aliases: c.opt_string()?,
    };
    c.finish()?;
    Ok(value)
}

fn decode_column(cells: &[Cell]) -> Result<proto::ColumnValue> {
    let mut c = Cursor::new(TableKind::Column, cells);
    let value = proto::ColumnValue {
        column_id: c.u64()?,
        begin_snapshot: c.u64()?,
        end_snapshot: c.opt_u64()?,
        table_id: c.u64()?,
        column_order: c.u64()?,
        column_name: c.string()?,
        column_type: c.string()?,
        initial_default: c.opt_string()?,
        default_value: c.opt_string()?,
        nulls_allowed: c.bool()?,
        parent_column: c.opt_u64()?,
        default_value_type: c.opt_string()?,
        default_value_dialect: c.opt_string()?,
        tags: Vec::new(),
    };
    c.finish()?;
    Ok(value)
}

fn decode_data_file(cells: &[Cell]) -> Result<proto::DataFileValue> {
    let mut c = Cursor::new(TableKind::DataFile, cells);
    let value = proto::DataFileValue {
        data_file_id: c.u64()?,
        table_id: c.u64()?,
        begin_snapshot: c.u64()?,
        end_snapshot: c.opt_u64()?,
        file_order: c.opt_u64()?,
        path: c.string()?,
        path_is_relative: c.bool()?,
        file_format: c.string()?,
        record_count: c.u64()?,
        file_size_bytes: c.u64()?,
        footer_size: c.u64()?,
        row_id_start: c.opt_u64()?,
        partition_id: c.opt_u64()?,
        encryption_key: c.opt_string()?,
        mapping_id: c.opt_u64()?,
        partial_max: c.opt_u64()?,
        partition_values: Vec::new(),
    };
    c.finish()?;
    Ok(value)
}

fn decode_delete_file(cells: &[Cell]) -> Result<proto::DeleteFileValue> {
    let mut c = Cursor::new(TableKind::DeleteFile, cells);
    let value = proto::DeleteFileValue {
        delete_file_id: c.u64()?,
        table_id: c.u64()?,
        begin_snapshot: c.u64()?,
        end_snapshot: c.opt_u64()?,
        data_file_id: c.u64()?,
        path: c.string()?,
        path_is_relative: c.bool()?,
        format: c.string()?,
        delete_count: c.u64()?,
        file_size_bytes: c.u64()?,
        footer_size: c.u64()?,
        encryption_key: c.opt_string()?,
        partial_max: c.opt_u64()?,
    };
    c.finish()?;
    Ok(value)
}

fn decode_table_stats(cells: &[Cell]) -> Result<proto::TableStatsValue> {
    let mut c = Cursor::new(TableKind::TableStats, cells);
    let value = proto::TableStatsValue {
        table_id: c.u64()?,
        record_count: c.u64()?,
        next_row_id: c.u64()?,
        file_size_bytes: c.u64()?,
    };
    c.finish()?;
    Ok(value)
}

fn decode_table_column_stats(cells: &[Cell]) -> Result<proto::TableColumnStatsValue> {
    let mut c = Cursor::new(TableKind::TableColumnStats, cells);
    let value = proto::TableColumnStatsValue {
        table_id: c.u64()?,
        column_id: c.u64()?,
        contains_null: c.opt_bool()?,
        contains_nan: c.opt_bool()?,
        min_value: c.opt_string()?,
        max_value: c.opt_string()?,
        extra_stats: c.opt_string()?,
    };
    c.finish()?;
    Ok(value)
}

fn decode_file_column_stats(cells: &[Cell]) -> Result<proto::FileColumnStatsValue> {
    let mut c = Cursor::new(TableKind::FileColumnStats, cells);
    let value = proto::FileColumnStatsValue {
        data_file_id: c.u64()?,
        table_id: c.u64()?,
        column_id: c.u64()?,
        column_size_bytes: c.u64()?,
        value_count: c.u64()?,
        null_count: c.u64()?,
        min_value: c.opt_string()?,
        max_value: c.opt_string()?,
        contains_nan: c.opt_bool()?,
        extra_stats: c.opt_string()?,
        variant_stats: Vec::new(),
    };
    c.finish()?;
    Ok(value)
}

fn decode_partition_info(cells: &[Cell]) -> Result<proto::PartitionValue> {
    let mut c = Cursor::new(TableKind::PartitionInfo, cells);
    let value = proto::PartitionValue {
        partition_id: c.u64()?,
        table_id: c.u64()?,
        begin_snapshot: c.u64()?,
        end_snapshot: c.opt_u64()?,
        columns: Vec::new(),
    };
    c.finish()?;
    Ok(value)
}

fn decode_partition_column(cells: &[Cell]) -> Result<(u64, proto::PartitionColumn)> {
    let mut c = Cursor::new(TableKind::PartitionColumn, cells);
    let partition_id = c.u64()?;
    let _table_id = c.u64()?;
    let column = proto::PartitionColumn {
        partition_key_index: c.u64()?,
        column_id: c.u64()?,
        transform: c.string()?,
    };
    c.finish()?;
    Ok((partition_id, column))
}

fn decode_file_partition_value(cells: &[Cell]) -> Result<((u64, u64), proto::FilePartitionValue)> {
    let mut c = Cursor::new(TableKind::FilePartitionValue, cells);
    let data_file_id = c.u64()?;
    let table_id = c.u64()?;
    let value = proto::FilePartitionValue {
        partition_key_index: c.u64()?,
        partition_value: c.string()?,
    };
    c.finish()?;
    Ok(((table_id, data_file_id), value))
}

fn decode_sort_info(cells: &[Cell]) -> Result<proto::SortValue> {
    let mut c = Cursor::new(TableKind::SortInfo, cells);
    let value = proto::SortValue {
        sort_id: c.u64()?,
        table_id: c.u64()?,
        begin_snapshot: c.u64()?,
        end_snapshot: c.opt_u64()?,
        expressions: Vec::new(),
    };
    c.finish()?;
    Ok(value)
}

fn decode_tag_row(cells: &[Cell]) -> Result<(u64, proto::TagEntry)> {
    let mut c = Cursor::new(TableKind::Tag, cells);
    let object_id = c.u64()?;
    let entry = proto::TagEntry {
        begin_snapshot: c.u64()?,
        end_snapshot: c.opt_u64()?,
        key: c.string()?,
        value: c.string()?,
    };
    c.finish()?;

    Ok((object_id, entry))
}

fn decode_column_tag_row(cells: &[Cell]) -> Result<((u64, u64), proto::ColumnTag)> {
    let mut c = Cursor::new(TableKind::ColumnTag, cells);
    let table_id = c.u64()?;
    let column_id = c.u64()?;
    let tag = proto::ColumnTag {
        begin_snapshot: c.u64()?,
        end_snapshot: c.opt_u64()?,
        key: c.string()?,
        value: c.string()?,
    };
    c.finish()?;

    Ok(((table_id, column_id), tag))
}

fn decode_gc_file_row(cells: &[Cell]) -> Result<proto::GcFileValue> {
    let mut c = Cursor::new(TableKind::FilesScheduledForDeletion, cells);
    let value = proto::GcFileValue {
        data_file_id: c.u64()?,
        path: c.string()?,
        path_is_relative: c.bool()?,
        schedule_start_micros: c.i64()?,
    };
    c.finish()?;

    Ok(value)
}

fn decode_sort_expression(cells: &[Cell]) -> Result<(u64, proto::SortExpression)> {
    let mut c = Cursor::new(TableKind::SortExpression, cells);
    let sort_id = c.u64()?;
    let _table_id = c.u64()?;
    let expression = proto::SortExpression {
        sort_key_index: c.u64()?,
        expression: c.string()?,
        dialect: c.string()?,
        sort_direction: c.string()?,
        null_order: c.string()?,
    };
    c.finish()?;
    Ok((sort_id, expression))
}

fn decode_macro(cells: &[Cell]) -> Result<proto::MacroValue> {
    let mut c = Cursor::new(TableKind::Macro, cells);
    let schema_id = c.u64()?;
    let macro_id = c.u64()?;
    let macro_name = c.string()?;
    let begin_snapshot = c.u64()?;
    let end_snapshot = c.opt_u64()?;
    c.finish()?;

    Ok(proto::MacroValue {
        macro_id,
        begin_snapshot,
        end_snapshot,
        schema_id,
        macro_name,
        implementations: Vec::new(),
    })
}

fn decode_macro_impl(cells: &[Cell]) -> Result<(u64, proto::MacroImplementation)> {
    let mut c = Cursor::new(TableKind::MacroImpl, cells);
    let macro_id = c.u64()?;
    let implementation = proto::MacroImplementation {
        impl_id: c.u64()?,
        dialect: c.string()?,
        sql: c.string()?,
        macro_type: c.string()?,
        parameters: Vec::new(),
    };
    c.finish()?;

    Ok((macro_id, implementation))
}

fn decode_macro_parameter(cells: &[Cell]) -> Result<((u64, u64), proto::MacroParameter)> {
    let mut c = Cursor::new(TableKind::MacroParameters, cells);
    let macro_id = c.u64()?;
    let impl_id = c.u64()?;
    let parameter = proto::MacroParameter {
        column_id: c.u64()?,
        parameter_name: c.string()?,
        parameter_type: c.string()?,
        default_value: c.opt_string()?,
        default_value_type: c.string()?,
    };
    c.finish()?;

    Ok(((macro_id, impl_id), parameter))
}

fn decode_column_mapping(cells: &[Cell]) -> Result<proto::MappingValue> {
    let mut c = Cursor::new(TableKind::ColumnMapping, cells);
    let mapping_id = c.u64()?;
    let table_id = c.u64()?;
    let map_type = c.string()?;
    c.finish()?;

    Ok(proto::MappingValue {
        mapping_id,
        table_id,
        map_type,
        name_mappings: Vec::new(),
    })
}

fn decode_name_mapping(cells: &[Cell]) -> Result<(u64, proto::NameMapping)> {
    let mut c = Cursor::new(TableKind::NameMapping, cells);
    let mapping_id = c.u64()?;
    let row = proto::NameMapping {
        column_id: c.u64()?,
        source_name: c.string()?,
        target_field_id: c.u64()?,
        parent_column: c.opt_u64()?,
        is_partition: c.bool()?,
    };
    c.finish()?;

    Ok((mapping_id, row))
}

/// Child-table rows collected before the insert pass: each is folded
/// into its parent record when the parent's insert applies, and a
/// leftover after the pass means a child row named a parent this commit
/// never inserted — a shape error.
#[derive(Default)]
struct ChildRows {
    partition_columns: HashMap<u64, Vec<proto::PartitionColumn>>,
    sort_expressions: HashMap<u64, Vec<proto::SortExpression>>,
    file_partition_values: HashMap<(u64, u64), Vec<proto::FilePartitionValue>>,
    macro_implementations: HashMap<u64, Vec<proto::MacroImplementation>>,
    macro_parameters: HashMap<(u64, u64), Vec<proto::MacroParameter>>,
    name_mappings: HashMap<u64, Vec<proto::NameMapping>>,
}

fn collect_child_rows(ops: &[RowOperation]) -> Result<ChildRows> {
    let mut children = ChildRows::default();
    for op in ops {
        if let RowOperation::Insert { table, cells } = op {
            match table {
                TableKind::PartitionColumn => {
                    let (partition_id, column) = decode_partition_column(cells)?;
                    children
                        .partition_columns
                        .entry(partition_id)
                        .or_default()
                        .push(column);
                }
                TableKind::SortExpression => {
                    let (sort_id, expression) = decode_sort_expression(cells)?;
                    children
                        .sort_expressions
                        .entry(sort_id)
                        .or_default()
                        .push(expression);
                }
                TableKind::FilePartitionValue => {
                    let (file, value) = decode_file_partition_value(cells)?;
                    children
                        .file_partition_values
                        .entry(file)
                        .or_default()
                        .push(value);
                }
                TableKind::MacroImpl => {
                    let (macro_id, implementation) = decode_macro_impl(cells)?;
                    children
                        .macro_implementations
                        .entry(macro_id)
                        .or_default()
                        .push(implementation);
                }
                TableKind::MacroParameters => {
                    let (key, parameter) = decode_macro_parameter(cells)?;
                    children
                        .macro_parameters
                        .entry(key)
                        .or_default()
                        .push(parameter);
                }
                TableKind::NameMapping => {
                    let (mapping_id, row) = decode_name_mapping(cells)?;
                    children
                        .name_mappings
                        .entry(mapping_id)
                        .or_default()
                        .push(row);
                }
                _ => {}
            }
        }
    }
    for columns in children.partition_columns.values_mut() {
        columns.sort_by_key(|c| c.partition_key_index);
    }
    for expressions in children.sort_expressions.values_mut() {
        expressions.sort_by_key(|e| e.sort_key_index);
    }
    for values in children.file_partition_values.values_mut() {
        values.sort_by_key(|v| v.partition_key_index);
    }
    for implementations in children.macro_implementations.values_mut() {
        implementations.sort_by_key(|i| i.impl_id);
    }
    for parameters in children.macro_parameters.values_mut() {
        parameters.sort_by_key(|p| p.column_id);
    }
    for rows in children.name_mappings.values_mut() {
        rows.sort_by_key(|r| r.column_id);
    }

    Ok(children)
}

/// Decodes an `UPDATE ... SET end_snapshot` row into the ended entity's
/// key and the new `end_snapshot` value. Defined only for the versioned
/// kinds.
fn decode_end(table: TableKind, cells: &[Cell]) -> Result<(EntityKey, u64)> {
    let mut c = Cursor::new(table, cells);
    let key = match table {
        TableKind::Schema => EntityKey::Schema {
            schema_id: c.u64()?,
        },
        TableKind::Table => EntityKey::Table { table_id: c.u64()? },
        TableKind::View => EntityKey::View { view_id: c.u64()? },
        TableKind::Column => EntityKey::Column {
            table_id: c.u64()?,
            column_id: c.u64()?,
        },
        TableKind::DataFile => EntityKey::File {
            table_id: c.u64()?,
            data_file_id: c.u64()?,
        },
        TableKind::DeleteFile => EntityKey::DeleteFile {
            table_id: c.u64()?,
            delete_file_id: c.u64()?,
        },
        TableKind::PartitionInfo => EntityKey::Partition {
            table_id: c.u64()?,
            partition_id: c.u64()?,
        },
        TableKind::SortInfo => EntityKey::Sort {
            table_id: c.u64()?,
            sort_id: c.u64()?,
        },
        TableKind::Macro => EntityKey::Macro { macro_id: c.u64()? },
        // Tag entries end via their own path (`apply_update_set_end`
        // handles them before this decoder) — reaching here is a bug.
        TableKind::Snapshot
        | TableKind::SnapshotChanges
        | TableKind::SchemaVersions
        | TableKind::TableStats
        | TableKind::TableColumnStats
        | TableKind::FileColumnStats
        | TableKind::PartitionColumn
        | TableKind::FilePartitionValue
        | TableKind::SortExpression
        | TableKind::Tag
        | TableKind::ColumnTag
        | TableKind::FilesScheduledForDeletion
        | TableKind::MacroImpl
        | TableKind::MacroParameters
        | TableKind::ColumnMapping
        | TableKind::NameMapping => {
            return Err(Error::Constraint(format!(
                "update_set_end is not defined for {table:?}"
            )));
        }
    };
    let end_snapshot = c.u64()?;
    c.finish()?;
    Ok((key, end_snapshot))
}

/// Decodes a raw-delete row's key. Defined only for the three unversioned
/// statistics kinds.
enum StatsKey {
    Table(u64),
    Column(u64, u64),
    FileColumn(u64, u64, u64),
}

fn decode_delete_key(table: TableKind, cells: &[Cell]) -> Result<StatsKey> {
    let mut c = Cursor::new(table, cells);
    let key = match table {
        TableKind::TableStats => StatsKey::Table(c.u64()?),
        TableKind::TableColumnStats => StatsKey::Column(c.u64()?, c.u64()?),
        TableKind::FileColumnStats => {
            let data_file_id = c.u64()?;
            let table_id = c.u64()?;
            let column_id = c.u64()?;
            StatsKey::FileColumn(table_id, data_file_id, column_id)
        }
        _ => {
            return Err(Error::Constraint(format!(
                "delete is not defined for {table:?}"
            )));
        }
    };
    c.finish()?;
    Ok(key)
}

/// Applies one staged row to the working snapshot. `new_id` is this
/// commit's own new snapshot id — the only value an `UpdateSetEnd` row's
/// `end_snapshot` cell is ever allowed to carry (see the module doc).
fn apply_op(
    base: &CatalogSnapshot,
    state: &mut CatalogSnapshot,
    op: &RowOperation,
    new_id: u64,
    children: &mut ChildRows,
    direct: &mut Vec<commit::StagedWrite>,
) -> Result<()> {
    match op {
        RowOperation::Insert { table, cells } => apply_insert(base, state, *table, cells, children),
        RowOperation::UpdateSetEnd { table, cells } => {
            apply_update_set_end(state, *table, cells, new_id)
        }
        RowOperation::UpdateSetBegin { table, cells } => {
            apply_update_set_begin(base, state, *table, cells, new_id)
        }
        RowOperation::Delete { table, cells } => apply_delete(state, *table, cells, direct),
        // Inline ops never reach here — `translate` routes them to
        // `translate_inline`. Kept only for match exhaustiveness.
        RowOperation::InlineSchema { .. }
        | RowOperation::InlineInsert { .. }
        | RowOperation::InlineInlineDelete { .. }
        | RowOperation::InlineFileDelete { .. }
        | RowOperation::InlineFlushDelete { .. }
        | RowOperation::InlineDrop { .. }
        | RowOperation::InlineSchemaDrop { .. } => Ok(()),
    }
}

/// Whether `op` is one of the seven inline variants — routed to
/// `translate_inline`, never to [`apply_op`]'s `CatalogSnapshot` diff.
fn is_inline_op(op: &RowOperation) -> bool {
    matches!(
        op,
        RowOperation::InlineSchema { .. }
            | RowOperation::InlineInsert { .. }
            | RowOperation::InlineInlineDelete { .. }
            | RowOperation::InlineFileDelete { .. }
            | RowOperation::InlineFlushDelete { .. }
            | RowOperation::InlineDrop { .. }
            | RowOperation::InlineSchemaDrop { .. }
    )
}

/// Folds a `ducklake_macro` insert with its collected impl and parameter
/// rows into one record: at least one impl, ordinals contiguous from
/// zero, one `macro_type` across the macro.
fn apply_macro_insert(
    state: &mut CatalogSnapshot,
    cells: &[Cell],
    children: &mut ChildRows,
) -> Result<()> {
    let mut value = decode_macro(cells)?;
    let mut implementations = children
        .macro_implementations
        .remove(&value.macro_id)
        .unwrap_or_default();
    if implementations.is_empty() {
        return Err(corrupt_row(
            TableKind::Macro,
            "a macro insert requires at least one macro_impl row in the same commit",
        ));
    }
    for (index, implementation) in implementations.iter().enumerate() {
        if implementation.impl_id != index as u64 {
            return Err(corrupt_row(
                TableKind::MacroImpl,
                "impl_id values must be contiguous from zero",
            ));
        }
        if implementation.macro_type != implementations[0].macro_type {
            return Err(corrupt_row(
                TableKind::MacroImpl,
                "all implementations of one macro must share a type",
            ));
        }
    }
    for implementation in &mut implementations {
        let parameters = children
            .macro_parameters
            .remove(&(value.macro_id, implementation.impl_id))
            .unwrap_or_default();
        for (index, parameter) in parameters.iter().enumerate() {
            if parameter.column_id != index as u64 {
                return Err(corrupt_row(
                    TableKind::MacroParameters,
                    "column_id values must be contiguous from zero",
                ));
            }
        }
        implementation.parameters = parameters;
    }
    value.implementations = implementations;
    state.put_macro(value);

    Ok(())
}

/// Folds a `ducklake_column_mapping` insert with its collected
/// name-mapping rows into one record: at least one row, unique ordinals,
/// parents preceding children, and a mapping id never written before.
fn apply_mapping_insert(
    state: &mut CatalogSnapshot,
    cells: &[Cell],
    children: &mut ChildRows,
) -> Result<()> {
    let mut value = decode_column_mapping(cells)?;
    let rows = children
        .name_mappings
        .remove(&value.mapping_id)
        .unwrap_or_default();
    if rows.is_empty() {
        return Err(corrupt_row(
            TableKind::ColumnMapping,
            "a column_mapping insert requires name_mapping rows in the same commit",
        ));
    }
    for (index, row) in rows.iter().enumerate() {
        if index > 0 && rows[index - 1].column_id == row.column_id {
            return Err(corrupt_row(
                TableKind::NameMapping,
                "column_id values must be unique within a mapping",
            ));
        }
        if row
            .parent_column
            .is_some_and(|parent| parent >= row.column_id)
        {
            return Err(corrupt_row(
                TableKind::NameMapping,
                "parent_column must reference an earlier column_id",
            ));
        }
    }
    if state
        .mappings
        .get(&value.table_id)
        .is_some_and(|per_table| per_table.contains_key(&value.mapping_id))
    {
        return Err(corrupt_row(
            TableKind::ColumnMapping,
            "mapping_id already exists for this table",
        ));
    }
    value.name_mappings = rows;
    state.put_mapping(value);

    Ok(())
}

fn apply_insert(
    base: &CatalogSnapshot,
    state: &mut CatalogSnapshot,
    table: TableKind,
    cells: &[Cell],
    children: &mut ChildRows,
) -> Result<()> {
    match table {
        // Snapshot rows fold into the snapshot record separately; child
        // rows fold into their parent records via `collect_child_rows`.
        // Neither is an entity mutation of its own.
        TableKind::Snapshot
        | TableKind::SnapshotChanges
        | TableKind::SchemaVersions
        | TableKind::PartitionColumn
        | TableKind::SortExpression
        | TableKind::FilePartitionValue
        | TableKind::MacroImpl
        | TableKind::MacroParameters
        | TableKind::NameMapping => {}
        TableKind::Schema => state.put_schema(decode_schema(cells)?),
        TableKind::Table => state.put_table(table_value(base, decode_table(cells)?)),
        TableKind::View => state.put_view(decode_view(cells)?),
        TableKind::Column => {
            let mut value = decode_column(cells)?;
            // Column tags outlive column versions (DuckLake keys them by
            // (table_id, column_id) with their own lifecycle), so a new
            // version carries the prior version's entries forward —
            // DuckLake never re-authors tag rows on a column alter.
            if let Some(prior) = base
                .columns
                .get(&value.table_id)
                .and_then(|cols| cols.get(&value.column_id))
            {
                value.tags.clone_from(&prior.tags);
            }
            state.put_column(value);
        }
        TableKind::DataFile => {
            let mut value = decode_data_file(cells)?;
            value.partition_values = children
                .file_partition_values
                .remove(&(value.table_id, value.data_file_id))
                .unwrap_or_default();
            state.put_data_file(value);
        }
        TableKind::DeleteFile => state.put_delete_file(decode_delete_file(cells)?),
        TableKind::PartitionInfo => {
            let mut value = decode_partition_info(cells)?;
            value.columns = children
                .partition_columns
                .remove(&value.partition_id)
                .unwrap_or_default();
            state.put_partition(value);
        }
        TableKind::SortInfo => {
            let mut value = decode_sort_info(cells)?;
            value.expressions = children
                .sort_expressions
                .remove(&value.sort_id)
                .unwrap_or_default();
            state.put_sort(value);
        }
        TableKind::Macro => apply_macro_insert(state, cells, children)?,
        TableKind::ColumnMapping => apply_mapping_insert(state, cells, children)?,
        TableKind::TableStats => state.put_table_stats(decode_table_stats(cells)?),
        TableKind::TableColumnStats => {
            state.put_table_column_stats(decode_table_column_stats(cells)?);
        }
        TableKind::FileColumnStats => state.put_file_column_stats(decode_file_column_stats(cells)?),
        TableKind::FilesScheduledForDeletion => {
            state.put_gc_file(decode_gc_file_row(cells)?);
        }
        TableKind::Tag => {
            let (object_id, entry) = decode_tag_row(cells)?;
            state
                .tags
                .entry(object_id)
                .or_insert_with(|| proto::TagValue {
                    object_id,
                    entries: Vec::new(),
                })
                .entries
                .push(entry);
        }
        TableKind::ColumnTag => {
            let ((table_id, column_id), tag) = decode_column_tag_row(cells)?;
            let Some(column) = state
                .columns
                .get_mut(&table_id)
                .and_then(|cols| cols.get_mut(&column_id))
            else {
                return Err(corrupt_row(
                    table,
                    format!("column tag names an absent column ({table_id}, {column_id})"),
                ));
            };
            column.tags.push(tag);
        }
    }
    Ok(())
}

/// Ends the one live entry (`end_snapshot IS NULL`) for `key` in
/// `entries`, in place. `false` means no live entry matched — the caller
/// reports the shape error (DuckLake's UPDATE only names rows it read).
fn end_live_entry<E>(
    entries: &mut [E],
    is_match: impl Fn(&E) -> bool,
    set_end: impl Fn(&mut E),
) -> bool {
    for entry in entries.iter_mut() {
        if is_match(entry) {
            set_end(entry);
            return true;
        }
    }
    false
}

/// Rejects a set-end cell whose `end_snapshot` is not this commit's own
/// snapshot id.
fn check_end_snapshot(table: TableKind, end_snapshot: u64, new_id: u64) -> Result<()> {
    if end_snapshot == new_id {
        Ok(())
    } else {
        Err(corrupt_row(
            table,
            format!(
                "end_snapshot {end_snapshot} does not match this commit's snapshot id {new_id}"
            ),
        ))
    }
}

/// Ends a `ducklake_tag` row: the live entry named by `(object_id, key)`
/// gets its `end_snapshot` set, in place — containers never move to
/// history.
fn apply_tag_set_end(state: &mut CatalogSnapshot, cells: &[Cell], new_id: u64) -> Result<()> {
    let mut c = Cursor::new(TableKind::Tag, cells);
    let object_id = c.u64()?;
    let key = c.string()?;
    let end_snapshot = c.u64()?;
    c.finish()?;
    check_end_snapshot(TableKind::Tag, end_snapshot, new_id)?;

    let ended = state.tags.get_mut(&object_id).is_some_and(|container| {
        end_live_entry(
            &mut container.entries,
            |e| e.key == key && e.end_snapshot.is_none(),
            |e| e.end_snapshot = Some(new_id),
        )
    });
    if ended {
        Ok(())
    } else {
        Err(corrupt_row(
            TableKind::Tag,
            format!("no live tag entry ({object_id}, {key:?}) to end"),
        ))
    }
}

/// Ends a `ducklake_column_tag` row: the live entry named by
/// `(table_id, column_id, key)` on the current column record.
fn apply_column_tag_set_end(
    state: &mut CatalogSnapshot,
    cells: &[Cell],
    new_id: u64,
) -> Result<()> {
    let mut c = Cursor::new(TableKind::ColumnTag, cells);
    let table_id = c.u64()?;
    let column_id = c.u64()?;
    let key = c.string()?;
    let end_snapshot = c.u64()?;
    c.finish()?;
    check_end_snapshot(TableKind::ColumnTag, end_snapshot, new_id)?;

    let ended = state
        .columns
        .get_mut(&table_id)
        .and_then(|cols| cols.get_mut(&column_id))
        .is_some_and(|column| {
            end_live_entry(
                &mut column.tags,
                |t| t.key == key && t.end_snapshot.is_none(),
                |t| t.end_snapshot = Some(new_id),
            )
        });
    if ended {
        Ok(())
    } else {
        Err(corrupt_row(
            TableKind::ColumnTag,
            format!("no live column tag ({table_id}, {column_id}, {key:?}) to end"),
        ))
    }
}

fn apply_update_set_end(
    state: &mut CatalogSnapshot,
    table: TableKind,
    cells: &[Cell],
    new_id: u64,
) -> Result<()> {
    // Tag entries are embedded, not entity records of their own: ending
    // one rewrites its container in place rather than moving a key to
    // history.
    match table {
        TableKind::Tag => return apply_tag_set_end(state, cells, new_id),
        TableKind::ColumnTag => return apply_column_tag_set_end(state, cells, new_id),
        _ => {}
    }

    let (key, end_snapshot) = decode_end(table, cells)?;
    check_end_snapshot(table, end_snapshot, new_id)?;
    // End only the one row DuckLake named — never a cascade. DuckLake
    // authors every row change explicitly (a rename ends the table row but
    // keeps its columns live); the verb-path `delete_*` helpers would
    // cascade and end those siblings.
    let ended = match key {
        EntityKey::Schema { schema_id } => state.schemas.remove(&schema_id).is_some(),
        EntityKey::Table { table_id } => state.tables.remove(&table_id).is_some(),
        EntityKey::View { view_id } => state.views.remove(&view_id).is_some(),
        EntityKey::Column {
            table_id,
            column_id,
        } => state
            .columns
            .get_mut(&table_id)
            .is_some_and(|columns| columns.remove(&column_id).is_some()),
        EntityKey::File {
            table_id,
            data_file_id,
        } => state
            .data_files
            .get_mut(&table_id)
            .is_some_and(|files| files.remove(&data_file_id).is_some()),
        EntityKey::DeleteFile {
            table_id,
            delete_file_id,
        } => state
            .delete_files
            .get_mut(&table_id)
            .is_some_and(|files| files.remove(&delete_file_id).is_some()),
        EntityKey::Partition {
            table_id,
            partition_id,
        } => {
            let live = state
                .partitions
                .get(&table_id)
                .is_some_and(|specs| specs.contains_key(&partition_id));
            state.delete_partition(table_id, partition_id);
            live
        }
        EntityKey::Sort { table_id, sort_id } => {
            let live = state
                .sorts
                .get(&table_id)
                .is_some_and(|specs| specs.contains_key(&sort_id));
            state.delete_sort(table_id, sort_id);
            live
        }
        EntityKey::Macro { macro_id } => state.macros.remove(&macro_id).is_some(),
        // decode_end only ever returns the keys matched above.
        _ => return Err(corrupt_row(table, "unreachable entity key")),
    };
    // DuckLake's UPDATE only names rows it read; ending an absent row is
    // drift and must fail loudly, never pass as a no-op.
    if !ended {
        return Err(corrupt_row(
            table,
            format!("no live row to end for {key:?}"),
        ));
    }

    Ok(())
}

/// Decodes a versioned kind's hard-delete row: the entity's key columns
/// (decoder order) followed by the row's `end_snapshot` — `NULL` names
/// the live `current` record, a value names that `history` record.
fn decode_hard_delete(table: TableKind, cells: &[Cell]) -> Result<(EntityKey, Option<u64>)> {
    let mut c = Cursor::new(table, cells);
    let key = match table {
        TableKind::Schema => EntityKey::Schema {
            schema_id: c.u64()?,
        },
        TableKind::Table => EntityKey::Table { table_id: c.u64()? },
        TableKind::View => EntityKey::View { view_id: c.u64()? },
        TableKind::Column => EntityKey::Column {
            table_id: c.u64()?,
            column_id: c.u64()?,
        },
        TableKind::DataFile => EntityKey::File {
            table_id: c.u64()?,
            data_file_id: c.u64()?,
        },
        TableKind::DeleteFile => EntityKey::DeleteFile {
            table_id: c.u64()?,
            delete_file_id: c.u64()?,
        },
        TableKind::PartitionInfo => EntityKey::Partition {
            table_id: c.u64()?,
            partition_id: c.u64()?,
        },
        TableKind::SortInfo => EntityKey::Sort {
            table_id: c.u64()?,
            sort_id: c.u64()?,
        },
        TableKind::Macro => EntityKey::Macro { macro_id: c.u64()? },
        // Callers dispatch every other kind before reaching here.
        _ => return Err(corrupt_row(table, "not a versioned kind")),
    };
    let end_snapshot = c.opt_u64()?;
    c.finish()?;
    Ok((key, end_snapshot))
}

/// Rebases a data file's `begin_snapshot` in place. The target must have
/// been inserted by this same transaction (absent from `base`): DuckLake
/// only rebases the replacement file a delete-rewrite just created, and
/// rebasing a pre-existing row would rewrite committed visibility.
fn apply_update_set_begin(
    base: &CatalogSnapshot,
    state: &mut CatalogSnapshot,
    table: TableKind,
    cells: &[Cell],
    new_id: u64,
) -> Result<()> {
    if table != TableKind::DataFile {
        return Err(Error::Constraint(format!(
            "update_set_begin is not defined for {table:?}"
        )));
    }
    let mut c = Cursor::new(table, cells);
    let table_id = c.u64()?;
    let data_file_id = c.u64()?;
    let begin_snapshot = c.u64()?;
    c.finish()?;
    if begin_snapshot != new_id {
        return Err(corrupt_row(
            table,
            format!(
                "begin_snapshot {begin_snapshot} does not match this commit's snapshot id {new_id}"
            ),
        ));
    }
    if base
        .data_files
        .get(&table_id)
        .is_some_and(|files| files.contains_key(&data_file_id))
    {
        return Err(corrupt_row(
            table,
            format!("file ({table_id}, {data_file_id}) predates this commit and cannot be rebased"),
        ));
    }

    let Some(file) = state
        .data_files
        .get_mut(&table_id)
        .and_then(|files| files.get_mut(&data_file_id))
    else {
        return Err(corrupt_row(
            table,
            format!("no data file ({table_id}, {data_file_id}) to rebase"),
        ));
    };
    file.begin_snapshot = begin_snapshot;
    Ok(())
}

/// A raw `DELETE` row. Three shapes share the op: unversioned records
/// (statistics, the deletion schedule) leave the working state and their
/// removal reaches the store through the diff; versioned rows and
/// snapshot records are pruned with direct key deletes (`history` keys
/// exist only in the store, never in the working state); embedded rows
/// (tag entries, spec columns) rewrite or ride their parent.
fn apply_delete(
    state: &mut CatalogSnapshot,
    table: TableKind,
    cells: &[Cell],
    direct: &mut Vec<commit::StagedWrite>,
) -> Result<()> {
    match table {
        TableKind::TableStats | TableKind::TableColumnStats | TableKind::FileColumnStats => {
            apply_stats_delete(state, table, cells)
        }
        TableKind::FilesScheduledForDeletion => {
            let mut c = Cursor::new(table, cells);
            let data_file_id = c.u64()?;
            c.finish()?;
            if state.gc_files.remove(&data_file_id).is_none() {
                return Err(corrupt_row(
                    table,
                    format!("no scheduled deletion for file {data_file_id}"),
                ));
            }
            Ok(())
        }
        // The merged snapshot record dies with the `ducklake_snapshot`
        // delete; the paired `ducklake_snapshot_changes` delete names the
        // same id and stages nothing.
        TableKind::Snapshot => {
            let mut c = Cursor::new(table, cells);
            let snapshot_id = c.u64()?;
            c.finish()?;
            if snapshot_id == state.snapshot.snapshot_id {
                return Err(Error::Constraint(format!(
                    "snapshot {snapshot_id} is the head and cannot be expired"
                )));
            }
            direct.push((Key::Snapshot { snapshot_id }.encode(), None));
            Ok(())
        }
        TableKind::SnapshotChanges => {
            let mut c = Cursor::new(table, cells);
            let _snapshot_id = c.u64()?;
            c.finish()?;
            Ok(())
        }
        // Mappings are unversioned create-only records with no history
        // mirror: cleanup is a direct `current` key delete. The working
        // state keeps its (now equal-to-base) entry, which the create-only
        // diff no-ops on.
        TableKind::ColumnMapping => {
            let mut c = Cursor::new(table, cells);
            let mapping_id = c.u64()?;
            let table_id = c.u64()?;
            c.finish()?;
            direct.push((
                Key::current(EntityKey::Mapping {
                    table_id,
                    mapping_id,
                })
                .encode(),
                None,
            ));
            Ok(())
        }
        TableKind::Schema
        | TableKind::Table
        | TableKind::View
        | TableKind::Column
        | TableKind::DataFile
        | TableKind::DeleteFile
        | TableKind::PartitionInfo
        | TableKind::SortInfo
        | TableKind::Macro => {
            let (entity, end_snapshot) = decode_hard_delete(table, cells)?;
            let key = match end_snapshot {
                Some(end) => Key::history(entity, end),
                None => Key::current(entity),
            };
            // The direct delete is the whole store mutation. The working
            // state is deliberately left alone: removing a live row from
            // it would make the diff stage an end-transition — minting a
            // history mirror DuckLake never authored — on top of this
            // delete. The cascade never re-touches a row it deleted, so
            // the stale working-state entry is unread.
            direct.push((key.encode(), None));
            Ok(())
        }
        TableKind::Tag => apply_tag_delete(state, cells),
        // Embedded rows ride their parent: the cascade deletes them only
        // alongside the parent record (a dead table's columns, a dead
        // file's partition values), so with the parent already pruned
        // there is nothing left to rewrite. A column-tag entry on a
        // still-current column is the one live case (its column survives
        // the entry's death) and rewrites the column in place.
        TableKind::ColumnTag => {
            let mut c = Cursor::new(table, cells);
            let table_id = c.u64()?;
            let column_id = c.u64()?;
            let key = c.string()?;
            let begin_snapshot = c.u64()?;
            c.finish()?;
            if let Some(column) = state
                .columns
                .get_mut(&table_id)
                .and_then(|cols| cols.get_mut(&column_id))
            {
                column
                    .tags
                    .retain(|t| !(t.key == key && t.begin_snapshot == begin_snapshot));
            }
            Ok(())
        }
        TableKind::PartitionColumn
        | TableKind::SortExpression
        | TableKind::FilePartitionValue
        | TableKind::MacroImpl
        | TableKind::MacroParameters
        | TableKind::NameMapping => apply_embedded_delete(state, table, cells),
        TableKind::SchemaVersions => {
            // Schema-version rows fold into snapshot records; the rows a
            // dead-table cleanup deletes are visible only through
            // snapshots the same transaction deletes, so there is nothing
            // separate to remove.
            let mut c = Cursor::new(table, cells);
            let _begin_snapshot = c.u64()?;
            let _schema_version = c.u64()?;
            let _table_id = c.u64()?;
            c.finish()?;
            Ok(())
        }
    }
}

/// Removes a dead `ducklake_tag` entry from its container; a container
/// left empty is removed outright (the key must not linger).
fn apply_tag_delete(state: &mut CatalogSnapshot, cells: &[Cell]) -> Result<()> {
    let mut c = Cursor::new(TableKind::Tag, cells);
    let object_id = c.u64()?;
    let key = c.string()?;
    let begin_snapshot = c.u64()?;
    c.finish()?;

    let removed = state.tags.get_mut(&object_id).is_some_and(|container| {
        let before = container.entries.len();
        container
            .entries
            .retain(|e| !(e.key == key && e.begin_snapshot == begin_snapshot));
        container.entries.len() < before
    });
    if !removed {
        return Err(corrupt_row(
            TableKind::Tag,
            format!("no tag entry ({object_id}, {key:?}, {begin_snapshot}) to delete"),
        ));
    }
    if state
        .tags
        .get(&object_id)
        .is_some_and(|container| container.entries.is_empty())
    {
        state.tags.remove(&object_id);
    }
    Ok(())
}

/// A spec-column / partition-value delete: only ever issued alongside its
/// parent record's own deletion, so a still-current parent means the
/// cascade named a row this model says cannot die alone.
fn apply_embedded_delete(
    state: &mut CatalogSnapshot,
    table: TableKind,
    cells: &[Cell],
) -> Result<()> {
    let mut c = Cursor::new(table, cells);
    let parent_is_current = match table {
        TableKind::PartitionColumn => {
            let partition_id = c.u64()?;
            let table_id = c.u64()?;
            state
                .partitions
                .get(&table_id)
                .is_some_and(|specs| specs.contains_key(&partition_id))
        }
        TableKind::SortExpression => {
            let sort_id = c.u64()?;
            let table_id = c.u64()?;
            state
                .sorts
                .get(&table_id)
                .is_some_and(|specs| specs.contains_key(&sort_id))
        }
        TableKind::FilePartitionValue => {
            let data_file_id = c.u64()?;
            let table_id = c.u64()?;
            state
                .data_files
                .get(&table_id)
                .is_some_and(|files| files.contains_key(&data_file_id))
        }
        TableKind::MacroImpl | TableKind::MacroParameters => {
            let macro_id = c.u64()?;
            state.macros.contains_key(&macro_id)
        }
        TableKind::NameMapping => {
            let mapping_id = c.u64()?;
            state
                .mappings
                .values()
                .any(|per_table| per_table.contains_key(&mapping_id))
        }
        _ => return Err(corrupt_row(table, "not an embedded kind")),
    };
    // Remaining identity cells vary by kind and are not needed: the row
    // dies with its parent.
    if parent_is_current {
        return Err(corrupt_row(
            table,
            "embedded row deleted while its parent record is still live",
        ));
    }
    Ok(())
}

fn apply_stats_delete(state: &mut CatalogSnapshot, table: TableKind, cells: &[Cell]) -> Result<()> {
    match decode_delete_key(table, cells)? {
        StatsKey::Table(table_id) => {
            state.table_stats.remove(&table_id);
        }
        StatsKey::Column(table_id, column_id) => {
            if let Some(cols) = state.table_column_stats.get_mut(&table_id) {
                cols.remove(&column_id);
            }
        }
        StatsKey::FileColumn(table_id, data_file_id, column_id) => {
            if let Some(cols) = state.file_column_stats.get_mut(&table_id) {
                cols.remove(&(data_file_id, column_id));
            }
        }
    }
    Ok(())
}

/// The `ducklake_snapshot` and `ducklake_snapshot_changes` rows DuckLake
/// always inserts as one pair; both are required for a staged commit.
fn find_snapshot_rows(ops: &[RowOperation]) -> Result<(&[Cell], &[Cell])> {
    let mut snapshot = None;
    let mut changes = None;
    for op in ops {
        if let RowOperation::Insert { table, cells } = op {
            match table {
                TableKind::Snapshot => snapshot = Some(cells.as_slice()),
                TableKind::SnapshotChanges => changes = Some(cells.as_slice()),
                _ => {}
            }
        }
    }
    let snapshot = snapshot.ok_or_else(|| {
        Error::Constraint("staged commit requires a ducklake_snapshot insert".to_string())
    })?;
    let changes = changes.ok_or_else(|| {
        Error::Constraint("staged commit requires a ducklake_snapshot_changes insert".to_string())
    })?;
    Ok((snapshot, changes))
}

fn build_snapshot_value(ops: &[RowOperation]) -> Result<proto::SnapshotValue> {
    let (snapshot_cells, changes_cells) = find_snapshot_rows(ops)?;

    let mut s = Cursor::new(TableKind::Snapshot, snapshot_cells);
    let snapshot_id = s.u64()?;
    let snapshot_time_micros = s.i64()?;
    let schema_version = s.u64()?;
    let next_catalog_id = s.u64()?;
    let next_file_id = s.u64()?;
    s.finish()?;

    let mut c = Cursor::new(TableKind::SnapshotChanges, changes_cells);
    let changes_snapshot_id = c.u64()?;
    let changes_made = c.string()?;
    let author = c.opt_string()?;
    let commit_message = c.opt_string()?;
    let commit_extra_info = c.opt_string()?;
    c.finish()?;

    if changes_snapshot_id != snapshot_id {
        return Err(corrupt_row(
            TableKind::SnapshotChanges,
            format!(
                "snapshot_id {changes_snapshot_id} does not match ducklake_snapshot's {snapshot_id}"
            ),
        ));
    }

    // `ducklake_schema_versions` rows fold in as the table-id set; the two
    // redundant columns are validated against this commit's own snapshot
    // values rather than trusted, so a mismatch is drift caught loudly.
    let schema_changed_table_ids = ops
        .iter()
        .filter_map(|op| match op {
            RowOperation::Insert {
                table: TableKind::SchemaVersions,
                cells,
            } => Some(cells),
            _ => None,
        })
        .map(|cells| {
            let mut cursor = Cursor::new(TableKind::SchemaVersions, cells);
            let begin_snapshot = cursor.u64()?;
            let row_schema_version = cursor.u64()?;
            let table_id = cursor.u64()?;
            cursor.finish()?;
            if begin_snapshot != snapshot_id || row_schema_version != schema_version {
                return Err(corrupt_row(
                    TableKind::SchemaVersions,
                    format!(
                        "(begin_snapshot {begin_snapshot}, schema_version {row_schema_version}) \
                         does not match ducklake_snapshot's ({snapshot_id}, {schema_version})"
                    ),
                ));
            }
            Ok(table_id)
        })
        .collect::<Result<Vec<_>>>()?;

    Ok(proto::SnapshotValue {
        snapshot_id,
        snapshot_time_micros,
        schema_version,
        next_catalog_id,
        next_file_id,
        changes_made,
        author,
        commit_message,
        commit_extra_info,
        schema_changed_table_ids,
    })
}

/// A staged-row transaction: one SlateDB transaction opened by
/// `begin` (crate-internal; [`crate::ffi_support::staged::staged_begin`]
/// is the entry point outside this crate), accumulating [`RowOperation`]s via
/// [`stage`](Self::stage) until [`commit`](Self::commit) translates and
/// lands them all in one atomic batch, or [`rollback`](Self::rollback)
/// discards them.
pub struct StagedTransaction {
    db_tx: DbTransaction,
    ops: Vec<RowOperation>,
    projections: Arc<std::sync::RwLock<ProjectionCache>>,
    /// The `DATA_PATH` object store and its bucket-relative prefix, present
    /// when the attach supplied `META_DATA_PATH`. Index maintenance
    /// scoped-reads registered data files through it; absent it is skipped.
    data_store: Option<Arc<dyn ObjectStore>>,
    data_prefix: String,
}

impl StagedTransaction {
    /// Opens a fresh transaction at the current head. Nothing is staged
    /// yet; [`stage`](Self::stage) accumulates rows in memory only. A
    /// successful commit folds its batch into `projections` (a catalog's
    /// shared maintained-projection state).
    pub(crate) fn begin(
        db_tx: DbTransaction,
        projections: Arc<std::sync::RwLock<ProjectionCache>>,
        data_store: Option<Arc<dyn ObjectStore>>,
        data_prefix: String,
    ) -> Self {
        Self {
            db_tx,
            ops: Vec::new(),
            projections,
            data_store,
            data_prefix,
        }
    }

    /// As [`begin`](Self::begin), but with a throwaway, never-served
    /// projection state and no `DATA_PATH` store — for tests that drive a
    /// `StagedTransaction` directly without a `Catalog`.
    #[cfg(test)]
    pub(crate) fn begin_detached(db_tx: DbTransaction) -> Self {
        Self::begin(
            db_tx,
            Arc::new(std::sync::RwLock::new(ProjectionCache::empty())),
            None,
            String::new(),
        )
    }

    /// Accumulates one row mutation. Nothing touches the store until
    /// [`commit`](Self::commit).
    pub fn stage(&mut self, op: RowOperation) {
        self.ops.push(op);
    }

    /// Discards every staged row without writing anything.
    pub fn rollback(self) {
        self.db_tx.rollback();
    }

    /// Snapshot records as this transaction sees them: the committed
    /// rows at its read point, minus the snapshot deletes staged so far.
    /// The expiry cascade re-reads `ducklake_snapshot` after staging its
    /// deletes (its dead-row rule is `NOT EXISTS` over the survivors), so
    /// the projection must observe the transaction's own writes.
    ///
    /// # Errors
    ///
    /// Returns an error if the scan fails or a staged snapshot-delete row
    /// is malformed.
    pub async fn visible_snapshots(&self) -> Result<Vec<proto::SnapshotValue>> {
        let committed = read::scan_snapshots(ReadHandle::Tx(&self.db_tx)).await?;

        let mut deleted = std::collections::BTreeSet::new();
        for op in &self.ops {
            if let RowOperation::Delete {
                table: TableKind::Snapshot,
                cells,
            } = op
            {
                let mut c = Cursor::new(TableKind::Snapshot, cells);
                deleted.insert(c.u64()?);
                c.finish()?;
            }
        }

        Ok(committed
            .into_iter()
            .filter(|s| !deleted.contains(&s.snapshot_id))
            .collect())
    }

    /// Translates every staged row and lands them in one atomic batch.
    ///
    /// A commit with a `ducklake_snapshot` insert mints that snapshot and
    /// advances head. A commit **without** one is a maintenance commit —
    /// snapshot expiry / file cleanup, which DuckLake runs without
    /// minting a snapshot — and lands head-preserving: reclamation
    /// deletes only, no new snapshot record, `sys/head` untouched.
    ///
    /// # Errors
    ///
    /// Returns [`Error::Constraint`] or [`Error::Corruption`] if the
    /// staged rows are malformed, mutate entities without the required
    /// `ducklake_snapshot` / `ducklake_snapshot_changes` pair, or expire
    /// the head snapshot. Returns [`Error::CommitConflict`] — **never
    /// retried internally** — if a concurrent commit advanced the head
    /// first; the store is left unchanged by the loser.
    pub async fn commit(self) -> Result<SnapshotId> {
        let Self {
            db_tx,
            ops,
            projections,
            data_store,
            data_prefix,
        } = self;

        let base = match commit::materialize(ReadHandle::Tx(&db_tx), None).await {
            Ok(base) => base,
            Err(err) => {
                db_tx.rollback();
                return Err(err);
            }
        };

        // Read before any write in this commit is staged: `InlineFlushDelete`
        // /`InlineDrop` name a table, not keys, and resolve against
        // `db_tx`'s current state exactly like `base` above.
        let inline_writes = match translate_inline(&db_tx, &ops).await {
            Ok(writes) => writes,
            Err(err) => {
                db_tx.rollback();
                return Err(err);
            }
        };

        let mints_snapshot = ops.iter().any(|op| {
            matches!(
                op,
                RowOperation::Insert {
                    table: TableKind::Snapshot,
                    ..
                }
            )
        });

        let translated = if mints_snapshot {
            translate(&base, &ops).map(|(new_id, mut writes, snap)| {
                writes.push((
                    Key::Snapshot {
                        snapshot_id: new_id,
                    }
                    .encode(),
                    Some(value::encode_value(&snap)),
                ));
                writes.push((
                    Key::Sys(SysKey::Head).encode(),
                    Some(value::encode_value(&proto::HeadValue {
                        snapshot_id: new_id,
                    })),
                ));
                (new_id, writes)
            })
        } else {
            translate_maintenance(&base, &ops).map(|writes| (base.snapshot.snapshot_id, writes))
        };

        match translated {
            Ok((result_id, mut writes)) => {
                writes.extend(inline_writes);
                // Maintain equality-index entries for any data file this
                // commit registered on an indexed table, by scoped-reading it
                // from `DATA_PATH`. Gated: a no-op unless a live index covers
                // the file's table, so non-indexed writes are untouched. A
                // Parquet file on an indexed table with no store to read it
                // aborts the commit rather than under-covering the index.
                if let Err(err) = stage_index_maintenance(
                    &db_tx,
                    &base,
                    &ops,
                    data_store.as_deref(),
                    &data_prefix,
                    &mut writes,
                )
                .await
                {
                    db_tx.rollback();
                    return Err(err);
                }
                if let Err(err) = commit::stage_writes(&db_tx, &writes) {
                    db_tx.rollback();
                    return Err(err);
                }
                match db_tx.commit_with_options(&commit::durable()).await {
                    Ok(_) => {
                        fold_committed_batch(&projections, &writes, result_id);
                        Ok(SnapshotId::new(result_id))
                    }
                    Err(err) if err.kind() == slatedb::ErrorKind::Transaction => {
                        Err(Error::CommitConflict(format!(
                            "a concurrent commit changed state this one read or wrote \
                             (attempted snapshot {result_id}); staged-row commits are never \
                             retried internally"
                        )))
                    }
                    Err(err) => Err(err.into()),
                }
            }
            Err(err) => {
                db_tx.rollback();
                Err(err)
            }
        }
    }
}

/// Derives and appends the equality-index entries for one registered data
/// file (a `RowOperation::Insert` into `TableKind::DataFile`), by scoped-
/// reading the file. A no-op if the file's table has no live index. Refuses
/// (rather than silently under-covering) when the file must be read but no
/// store is available.
async fn stage_data_file_index_entries(
    base: &CatalogSnapshot,
    cells: &[Cell],
    data_store: Option<&dyn ObjectStore>,
    data_prefix: &str,
    entries: &mut Vec<StagedIndexEntry>,
) -> Result<()> {
    let file = decode_data_file(cells)?;
    let table = TableId::new(file.table_id);
    let indexes = base.indexes_of(table);
    if indexes.is_empty() {
        return Ok(());
    }
    // The file must be read to maintain the index; no store to read it means
    // the index would silently miss these rows.
    let data_store = data_store.ok_or_else(|| {
        Error::Constraint(format!(
            "data file {} on indexed table {} cannot be read to maintain its equality index: no \
             data-path store is available",
            file.data_file_id, file.table_id
        ))
    })?;
    let row_id_start = file.row_id_start.ok_or_else(|| {
        Error::Constraint(format!(
            "data file {} on indexed table {} carries per-row ids; index maintenance of rewrite \
             files is a follow-up",
            file.data_file_id, file.table_id
        ))
    })?;
    let path = data_file_object_path(base, &file, data_prefix)?;
    let live_columns = base.columns_of(table);
    for index in &indexes {
        let positions = index_positions(&live_columns, index, table)?;
        let scoped =
            scoped_read::scoped_read_entries(data_store, &path, &positions, None, row_id_start)
                .await?;
        push_index_entries(entries, index, scoped)?;
    }
    Ok(())
}

async fn stage_index_maintenance(
    db_tx: &DbTransaction,
    base: &CatalogSnapshot,
    ops: &[RowOperation],
    data_store: Option<&dyn ObjectStore>,
    data_prefix: &str,
    writes: &mut Vec<commit::StagedWrite>,
) -> Result<()> {
    // Schemas registered in this same commit, for an inline chunk whose
    // `inline/schema` record is not yet committed (the first insert).
    let mut pending_schemas: HashMap<(u64, u64), &[u8]> = HashMap::new();
    for op in ops {
        if let RowOperation::InlineSchema {
            table_id,
            schema_version,
            arrow_schema,
        } = op
        {
            pending_schemas.insert((*table_id, *schema_version), arrow_schema.as_slice());
        }
    }

    let mut entries: Vec<StagedIndexEntry> = Vec::new();
    for op in ops {
        match op {
            RowOperation::Insert {
                table: TableKind::DataFile,
                cells,
            } => {
                stage_data_file_index_entries(base, cells, data_store, data_prefix, &mut entries)
                    .await?;
            }
            RowOperation::InlineInsert {
                table_id,
                schema_version,
                row_id_start,
                arrow_body,
                ..
            } => {
                let table = TableId::new(*table_id);
                let indexes = base.indexes_of(table);
                if indexes.is_empty() {
                    continue;
                }
                // Use this commit's `InlineSchema` op if the insert registered
                // it, else the committed inline schema.
                let stored;
                let schema_ipc: &[u8] =
                    if let Some(bytes) = pending_schemas.get(&(*table_id, *schema_version)) {
                        bytes
                    } else {
                        stored = store_inline::read_inline_schema(
                            ReadHandle::Tx(db_tx),
                            *table_id,
                            *schema_version,
                        )
                        .await?
                        .ok_or_else(|| {
                            Error::Corruption(format!(
                                "no inline schema for table {table_id} version {schema_version}"
                            ))
                        })?
                        .arrow_schema;
                        &stored
                    };
                let live_columns = base.columns_of(table);
                for index in &indexes {
                    let positions = index_positions(&live_columns, index, table)?;
                    let scoped = scoped_read::inline_batch_entries(
                        schema_ipc,
                        arrow_body,
                        &positions,
                        *row_id_start,
                    )?;
                    push_index_entries(&mut entries, index, scoped)?;
                }
            }
            _ => {}
        }
    }
    if entries.is_empty() {
        return Ok(());
    }
    stage_index_entries(ReadHandle::Tx(db_tx), &entries, writes).await
}

/// The object path of a registered data file: its stored path relative to the
/// table's data directory (`<schema path><table path>`) under `DATA_PATH`,
/// whose bucket-relative prefix leads for an `s3://` store.
fn data_file_object_path(
    base: &CatalogSnapshot,
    file: &proto::DataFileValue,
    data_prefix: &str,
) -> Result<object_store::path::Path> {
    let table_value = base.tables.get(&file.table_id).ok_or_else(|| {
        Error::Corruption(format!(
            "registered file names unknown table {}",
            file.table_id
        ))
    })?;
    let schema_value = base.schemas.get(&table_value.schema_id).ok_or_else(|| {
        Error::Corruption(format!(
            "table {} references a missing schema",
            file.table_id
        ))
    })?;
    let table_prefix = format!("{}{}", schema_value.path, table_value.path);
    let relative = match (file.path_is_relative, data_prefix.is_empty()) {
        (false, _) => file.path.clone(),
        (true, true) => format!("{table_prefix}{}", file.path),
        (true, false) => format!("{data_prefix}/{table_prefix}{}", file.path),
    };
    Ok(object_store::path::Path::from(relative.as_str()))
}

/// The physical positions of an index's columns in a file or chunk written
/// under the current schema: each column's 0-based rank among the table's
/// columns (the order `columns_of` returns).
fn index_positions(
    live_columns: &[ColumnInfo],
    index: &IndexInfo,
    table: TableId,
) -> Result<Vec<usize>> {
    index
        .columns
        .iter()
        .map(|column| {
            live_columns
                .iter()
                .position(|c| c.id == *column)
                .ok_or_else(|| Error::NotFound(format!("indexed column {column} of table {table}")))
        })
        .collect()
}

/// Turns scoped-read entries into staged index entries, skipping any row with
/// a NULL indexed value (which gets no entry).
fn push_index_entries(
    entries: &mut Vec<StagedIndexEntry>,
    index: &IndexInfo,
    scoped: Vec<ScopedReadEntry>,
) -> Result<()> {
    for entry in scoped {
        if entry.values.iter().any(Option::is_none) {
            continue;
        }
        let values: Vec<_> = entry.values.into_iter().flatten().collect();
        entries.push(StagedIndexEntry {
            index_id: index.id.get(),
            unique: index.unique,
            key: encode_key(&values)?,
            row_id: entry.row_id,
            delete: false,
        });
    }
    Ok(())
}

/// Applies every op onto a clone of `base`, then diffs the two exactly as
/// a verb-path commit diffs its closure's output.
fn translate(
    base: &CatalogSnapshot,
    ops: &[RowOperation],
) -> Result<(u64, Vec<commit::StagedWrite>, proto::SnapshotValue)> {
    let snapshot = build_snapshot_value(ops)?;
    let new_id = snapshot.snapshot_id;

    // Ends and deletes apply before inserts, independent of DuckLake's
    // emit order: a rename ends the old version and inserts a new one
    // under the same id, and the insert must win their shared `current` key —
    // an end applied afterward would delete the id and erase the new row.
    // Begin-rebases apply last: their target is the row an insert in this
    // same commit created. Inline ops are skipped here entirely —
    // `commit` translates them separately via `translate_inline`, since
    // `CatalogSnapshot` has no notion of inlined rows to diff.
    let mut state = base.clone();
    let mut children = collect_child_rows(ops)?;
    let mut direct = Vec::new();
    for op in ops {
        if !is_inline_op(op)
            && !matches!(
                op,
                RowOperation::Insert { .. } | RowOperation::UpdateSetBegin { .. }
            )
        {
            apply_op(base, &mut state, op, new_id, &mut children, &mut direct)?;
        }
    }
    for op in ops {
        if matches!(op, RowOperation::Insert { .. }) {
            apply_op(base, &mut state, op, new_id, &mut children, &mut direct)?;
        }
    }
    for op in ops {
        if matches!(op, RowOperation::UpdateSetBegin { .. }) {
            apply_op(base, &mut state, op, new_id, &mut children, &mut direct)?;
        }
    }

    if !children.partition_columns.is_empty() {
        return Err(corrupt_row(
            TableKind::PartitionColumn,
            "partition_column rows without a matching partition_info insert in this commit",
        ));
    }
    if !children.sort_expressions.is_empty() {
        return Err(corrupt_row(
            TableKind::SortExpression,
            "sort_expression rows without a matching sort_info insert in this commit",
        ));
    }
    if !children.file_partition_values.is_empty() {
        return Err(corrupt_row(
            TableKind::FilePartitionValue,
            "file_partition_value rows without a matching data_file insert in this commit",
        ));
    }
    if !children.macro_implementations.is_empty() {
        return Err(corrupt_row(
            TableKind::MacroImpl,
            "macro_impl rows without a matching macro insert in this commit",
        ));
    }
    if !children.macro_parameters.is_empty() {
        return Err(corrupt_row(
            TableKind::MacroParameters,
            "macro_parameters rows without a matching macro_impl in this commit",
        ));
    }
    if !children.name_mappings.is_empty() {
        return Err(corrupt_row(
            TableKind::NameMapping,
            "name_mapping rows without a matching column_mapping insert in this commit",
        ));
    }

    // DuckLake authors column ids itself, so its inserts advance no
    // counter; float each table's field-id counter above every live id so
    // a later verb-path `add_column` can never re-allocate one.
    for (table_id, columns) in &state.columns {
        let Some(max_id) = columns.keys().max() else {
            continue;
        };
        if let Some(table) = state.tables.get_mut(table_id)
            && table.next_column_id <= *max_id
        {
            table.next_column_id = max_id + 1;
        }
    }

    let mut writes = commit::diff_writes(base, &state, new_id);
    writes.extend(direct);
    Ok((new_id, writes, snapshot))
}

/// Translates a head-preserving maintenance commit: snapshot expiry and
/// file cleanup arrive with no `ducklake_snapshot` insert (DuckLake mints
/// no snapshot for them), so nothing advances head and no snapshot record
/// is written. Only reclamation-shaped operations are legal — raw
/// deletes, schedule inserts, and the inline drops a dead table's cleanup
/// issues; any entity insert or lifecycle update without a snapshot row
/// is a constraint violation (DuckLake always mints a snapshot for real
/// catalog changes).
fn translate_maintenance(
    base: &CatalogSnapshot,
    ops: &[RowOperation],
) -> Result<Vec<commit::StagedWrite>> {
    let head = base.snapshot.snapshot_id;

    let mut state = base.clone();
    let mut children = ChildRows::default();
    let mut direct = Vec::new();
    for op in ops {
        let allowed = matches!(
            op,
            RowOperation::Delete { .. }
                | RowOperation::Insert {
                    table: TableKind::FilesScheduledForDeletion,
                    ..
                }
        ) || is_inline_op(op);
        if !allowed {
            return Err(Error::Constraint(
                "a staged commit without a ducklake_snapshot insert may only reclaim state                  (maintenance deletes and deletion-schedule inserts)"
                    .to_string(),
            ));
        }
        apply_op(base, &mut state, op, head, &mut children, &mut direct)?;
    }

    let mut writes = commit::diff_writes(base, &state, head);
    writes.extend(direct);
    Ok(writes)
}

/// Allocates `inline/insert` chunk sequence numbers within one commit: the
/// first [`RowOp::InlineInsert`] staged for a given `(table_id,
/// schema_version, begin_snapshot)` gets `chunk_seq` `0`, the next `1`,
/// and so on — disambiguating multiple chunks the same commit stages
/// against the same key prefix.
#[derive(Default)]
struct ChunkSeqAllocator(HashMap<(u64, u64, u64), u64>);

impl ChunkSeqAllocator {
    fn next(&mut self, table_id: u64, schema_version: u64, begin_snapshot: u64) -> u64 {
        let seq = self
            .0
            .entry((table_id, schema_version, begin_snapshot))
            .or_insert(0);
        let allocated = *seq;
        *seq += 1;
        allocated
    }
}

/// Removes every `inline/insert` chunk begun at or before `flush_snapshot`
/// for `(table_id, schema_version)`, plus the `inline/inline_delete` tombstones
/// those chunks' rows consumed. Reads `db_tx`'s current (pre-commit)
/// inline records — the flush op only ever names the table and snapshot,
/// never the keys to remove.
async fn translate_inline_flush_delete(
    db_tx: &DbTransaction,
    table_id: u64,
    schema_version: u64,
    flush_snapshot: u64,
    writes: &mut Vec<commit::StagedWrite>,
) -> Result<()> {
    let chunks = store_inline::scan_inline_chunks(ReadHandle::Tx(db_tx), table_id).await?;
    let inline_deletes =
        store_inline::scan_inline_inline_deletes(ReadHandle::Tx(db_tx), table_id).await?;

    let scoped: Vec<(InlineOperation, proto::InlineChunkValue)> = chunks
        .into_iter()
        .filter(
            |(op, _)| matches!(op, InlineOperation::Insert { schema_version: v, .. } if *v == schema_version),
        )
        .collect();

    for (op, _) in &scoped {
        if let InlineOperation::Insert { begin_snapshot, .. } = op {
            if *begin_snapshot <= flush_snapshot {
                writes.push((Key::Inline(InlineKey::Live(*op)).encode(), None));
            }
        }
    }

    // The rows the flushed chunks carried, including already-tombstoned
    // ones (`ForFlush`) — their `inline/inline_delete` records become orphaned
    // once the owning chunk is gone above, and must go with it.
    let rows = materialize_inline_rows(&scoped, &inline_deletes);
    for row in InlineScanKind::ForFlush.select(&rows, flush_snapshot, 0) {
        if row.end_snapshot.is_some() {
            writes.push((
                Key::Inline(InlineKey::Live(InlineOperation::InlineDelete {
                    table_id,
                    row_id: row.row_id,
                }))
                .encode(),
                None,
            ));
        }
    }

    Ok(())
}

/// Removes every `inline/*` record for `table_id`: schema, chunks, and
/// tombstones, read from `db_tx`'s current (pre-commit) state.
async fn translate_inline_drop(
    db_tx: &DbTransaction,
    table_id: u64,
    writes: &mut Vec<commit::StagedWrite>,
) -> Result<()> {
    for (op, _) in store_inline::scan_inline_chunks(ReadHandle::Tx(db_tx), table_id).await? {
        writes.push((Key::Inline(InlineKey::Live(op)).encode(), None));
    }
    for (row_id, _) in
        store_inline::scan_inline_inline_deletes(ReadHandle::Tx(db_tx), table_id).await?
    {
        writes.push((
            Key::Inline(InlineKey::Live(InlineOperation::InlineDelete {
                table_id,
                row_id,
            }))
            .encode(),
            None,
        ));
    }
    for (data_file_id, row_id, _) in
        store_inline::scan_inline_file_deletes(ReadHandle::Tx(db_tx), table_id).await?
    {
        writes.push((
            Key::Inline(InlineKey::Live(InlineOperation::FileDelete {
                table_id,
                data_file_id,
                row_id,
            }))
            .encode(),
            None,
        ));
    }
    for (schema_version, _) in
        store_inline::scan_inline_schemas(ReadHandle::Tx(db_tx), table_id).await?
    {
        writes.push((
            Key::Inline(InlineKey::Schema {
                table_id,
                schema_version,
            })
            .encode(),
            None,
        ));
    }
    Ok(())
}

/// Translates every staged inline op into direct `inline/*` key writes —
/// a separate pass from [`translate`], since inline records live outside
/// `CatalogSnapshot`'s entity model and are never diffed. `db_tx` is read
/// (for `InlineFlushDelete`/`InlineDrop`, which name a table rather than
/// the keys to remove) at its pre-commit state, before any of this
/// commit's own writes are staged onto it.
fn inline_schema_write(
    table_id: u64,
    schema_version: u64,
    arrow_schema: &[u8],
) -> commit::StagedWrite {
    (
        Key::Inline(InlineKey::Schema {
            table_id,
            schema_version,
        })
        .encode(),
        Some(value::encode_value(&proto::InlineSchemaValue {
            arrow_schema: arrow_schema.to_vec(),
        })),
    )
}

#[allow(clippy::too_many_arguments)]
fn inline_insert_write(
    table_id: u64,
    schema_version: u64,
    begin_snapshot: u64,
    chunk_seq: u64,
    row_id_start: u64,
    row_count: u64,
    arrow_body: &[u8],
) -> commit::StagedWrite {
    (
        Key::Inline(InlineKey::Live(InlineOperation::Insert {
            table_id,
            schema_version,
            begin_snapshot,
            chunk_seq,
        }))
        .encode(),
        Some(value::encode_value(&proto::InlineChunkValue {
            body: arrow_body.to_vec(),
            row_id_start,
            row_count,
            data_file_id: None,
        })),
    )
}

fn inline_inline_delete_write(
    table_id: u64,
    row_id: u64,
    end_snapshot: u64,
) -> commit::StagedWrite {
    (
        Key::Inline(InlineKey::Live(InlineOperation::InlineDelete {
            table_id,
            row_id,
        }))
        .encode(),
        Some(value::encode_value(&proto::InlineInlineDeleteValue {
            end_snapshot,
        })),
    )
}

fn inline_file_delete_write(
    table_id: u64,
    data_file_id: u64,
    row_id: u64,
    begin_snapshot: u64,
) -> commit::StagedWrite {
    (
        Key::Inline(InlineKey::Live(InlineOperation::FileDelete {
            table_id,
            data_file_id,
            row_id,
        }))
        .encode(),
        Some(value::encode_value(&proto::InlineFileDeleteValue {
            begin_snapshot,
        })),
    )
}

async fn translate_inline(
    db_tx: &DbTransaction,
    ops: &[RowOperation],
) -> Result<Vec<commit::StagedWrite>> {
    let mut writes = Vec::new();
    let mut chunk_seqs = ChunkSeqAllocator::default();

    for op in ops {
        match op {
            RowOperation::InlineSchema {
                table_id,
                schema_version,
                arrow_schema,
            } => writes.push(inline_schema_write(
                *table_id,
                *schema_version,
                arrow_schema,
            )),
            RowOperation::InlineInsert {
                table_id,
                schema_version,
                begin_snapshot,
                row_id_start,
                row_count,
                arrow_body,
            } => {
                let chunk_seq = chunk_seqs.next(*table_id, *schema_version, *begin_snapshot);
                writes.push(inline_insert_write(
                    *table_id,
                    *schema_version,
                    *begin_snapshot,
                    chunk_seq,
                    *row_id_start,
                    *row_count,
                    arrow_body,
                ));
            }
            RowOperation::InlineInlineDelete {
                table_id,
                row_id,
                end_snapshot,
            } => writes.push(inline_inline_delete_write(
                *table_id,
                *row_id,
                *end_snapshot,
            )),
            RowOperation::InlineFileDelete {
                table_id,
                data_file_id,
                row_id,
                begin_snapshot,
            } => writes.push(inline_file_delete_write(
                *table_id,
                *data_file_id,
                *row_id,
                *begin_snapshot,
            )),
            RowOperation::InlineFlushDelete {
                table_id,
                schema_version,
                flush_snapshot,
            } => {
                translate_inline_flush_delete(
                    db_tx,
                    *table_id,
                    *schema_version,
                    *flush_snapshot,
                    &mut writes,
                )
                .await?;
            }
            RowOperation::InlineDrop { table_id } => {
                translate_inline_drop(db_tx, *table_id, &mut writes).await?;
            }
            RowOperation::InlineSchemaDrop {
                table_id,
                schema_version,
            } => {
                writes.push((
                    Key::Inline(InlineKey::Schema {
                        table_id: *table_id,
                        schema_version: *schema_version,
                    })
                    .encode(),
                    None,
                ));
            }
            RowOperation::Insert { .. }
            | RowOperation::Delete { .. }
            | RowOperation::UpdateSetEnd { .. }
            | RowOperation::UpdateSetBegin { .. } => {}
        }
    }

    Ok(writes)
}

#[cfg(test)]
mod tests {
    use std::sync::Arc;

    use object_store::memory::InMemory;

    use super::*;
    use crate::catalog::{Catalog, CatalogOptions};

    fn schema_row(id: u64, name: &str, begin: u64) -> Vec<Cell> {
        vec![
            Cell::U64(id),
            Cell::Str(format!("uuid-{id}")),
            Cell::U64(begin),
            Cell::Null,
            Cell::Str(name.to_string()),
            Cell::Str(format!("{name}/")),
            Cell::Bool(true),
        ]
    }

    fn table_row(id: u64, schema_id: u64, name: &str, begin: u64, end: Option<u64>) -> Vec<Cell> {
        vec![
            Cell::U64(id),
            Cell::Str(format!("uuid-t{id}")),
            Cell::U64(begin),
            end.map_or(Cell::Null, Cell::U64),
            Cell::U64(schema_id),
            Cell::Str(name.to_string()),
            Cell::Str(format!("{name}/")),
            Cell::Bool(true),
        ]
    }

    fn column_row(table_id: u64, column_id: u64, name: &str, order: u64) -> Vec<Cell> {
        vec![
            Cell::U64(column_id),
            Cell::U64(0),
            Cell::Null,
            Cell::U64(table_id),
            Cell::U64(order),
            Cell::Str(name.to_string()),
            Cell::Str("BIGINT".to_string()),
            Cell::Null,
            Cell::Null,
            Cell::Bool(true),
            Cell::Null,
            Cell::Null,
            Cell::Null,
        ]
    }

    fn snapshot_row(id: u64, schema_version: u64, next_catalog_id: u64) -> Vec<Cell> {
        vec![
            Cell::U64(id),
            Cell::I64(1),
            Cell::U64(schema_version),
            Cell::U64(next_catalog_id),
            Cell::U64(0),
        ]
    }

    fn snapshot_changes_row(id: u64, changes_made: &str) -> Vec<Cell> {
        vec![
            Cell::U64(id),
            Cell::Str(changes_made.to_string()),
            Cell::Null,
            Cell::Null,
            Cell::Null,
        ]
    }

    async fn open() -> Catalog {
        Catalog::open(Arc::new(InMemory::new()), CatalogOptions::default())
            .await
            .unwrap()
    }

    /// A DuckLake-shaped snapshot bump plus table create: table `t` (id
    /// 1, schema 0 = bootstrap's `main`) with one column, staged and
    /// committed as one batch, then verified through the ordinary
    /// snapshot read (the same view the dump ABI serves).
    #[tokio::test]
    async fn stages_table_create_and_snapshot_bump() {
        let catalog = open().await;
        let db_tx = catalog.begin_write_tx().await.unwrap();
        let mut tx = StagedTransaction::begin_detached(db_tx);

        tx.stage(RowOperation::Insert {
            table: TableKind::Table,
            cells: table_row(1, 0, "t", 1, None),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::Column,
            cells: column_row(1, 1, "a", 0),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::Snapshot,
            cells: snapshot_row(1, 1, 2),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::SnapshotChanges,
            cells: snapshot_changes_row(1, r#"created_table:"main"."t""#),
        });

        let id = tx.commit().await.unwrap();
        assert_eq!(id.get(), 1);

        let snapshot = catalog.snapshot().await.unwrap();
        let tables = snapshot.tables_in(crate::catalog::SchemaId::new(0));
        assert_eq!(tables.len(), 1);
        assert_eq!(tables[0].name, "t");
        let cols = snapshot.columns_of(tables[0].id);
        assert_eq!(cols.len(), 1);
        assert_eq!(cols[0].name, "a");
    }

    /// Staged column inserts advance the table's field-id counter: a later
    /// verb `add_column` — even after the highest staged column is dropped
    /// — must never re-allocate a DuckLake-authored field id.
    #[tokio::test]
    async fn staged_columns_advance_the_field_id_counter() {
        use crate::catalog::{ColumnDef, ColumnId, SchemaId, TableId};

        let catalog = open().await;
        let db_tx = catalog.begin_write_tx().await.unwrap();
        let mut tx = StagedTransaction::begin_detached(db_tx);
        tx.stage(RowOperation::Insert {
            table: TableKind::Table,
            cells: table_row(1, 0, "t", 1, None),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::Column,
            cells: column_row(1, 2, "a", 0),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::Column,
            cells: column_row(1, 5, "b", 1),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::Snapshot,
            cells: snapshot_row(1, 1, 2),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::SnapshotChanges,
            cells: snapshot_changes_row(1, r#"created_table:"main"."t""#),
        });
        tx.commit().await.unwrap();

        // Drop the max-id column, then add: the freed id must not return.
        let table = TableId::new(1);
        catalog
            .commit(|tx| tx.drop_column(table, ColumnId::new(5)))
            .await
            .unwrap();
        let added = std::cell::Cell::new(None);
        catalog
            .commit(|tx| {
                added.set(Some(tx.add_column(
                    table,
                    &ColumnDef {
                        name: "c".into(),
                        column_type: "BIGINT".into(),
                        nulls_allowed: true,
                        default_value: None,
                    },
                )?));
                Ok(())
            })
            .await
            .unwrap();
        assert_eq!(
            added.get(),
            Some(ColumnId::new(6)),
            "field id 5 must not be reused"
        );

        // The counter-only table update stayed in place: one live table
        // row for snapshot 1's create, no counter-minted history version.
        let snapshot = catalog.snapshot().await.unwrap();
        assert_eq!(snapshot.tables_in(SchemaId::new(0)).len(), 1);
    }

    /// DuckLake's UPDATE only names rows it read: an `UpdateSetEnd` for a
    /// row that is not live is drift and must fail the commit loudly, not
    /// pass as a silent no-op that drops the authored end.
    #[tokio::test]
    async fn ending_an_absent_row_is_rejected() {
        let catalog = open().await;
        let db_tx = catalog.begin_write_tx().await.unwrap();
        let mut tx = StagedTransaction::begin_detached(db_tx);

        // No table 7 exists; end it at this commit's snapshot id (1).
        tx.stage(RowOperation::UpdateSetEnd {
            table: TableKind::Table,
            cells: vec![Cell::U64(7), Cell::U64(1)],
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::Snapshot,
            cells: snapshot_row(1, 1, 1),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::SnapshotChanges,
            cells: snapshot_changes_row(1, "dropped_table:7"),
        });

        let err = tx.commit().await.unwrap_err();
        assert!(matches!(err, crate::error::Error::Corruption(_)), "{err}");
    }

    /// DuckLake-authored data-file and delete-file rows carry
    /// `encryption_key` through commit and back out of the snapshot read
    /// verbatim — the faithful-conduit guarantee for key material.
    #[tokio::test]
    async fn encryption_keys_round_trip_through_staged_rows() {
        let catalog = open().await;
        let db_tx = catalog.begin_write_tx().await.unwrap();
        let mut tx = StagedTransaction::begin_detached(db_tx);

        tx.stage(RowOperation::Insert {
            table: TableKind::Table,
            cells: table_row(1, 0, "t", 1, None),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::Column,
            cells: column_row(1, 1, "a", 0),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::DataFile,
            cells: vec![
                Cell::U64(1),                     // data_file_id
                Cell::U64(1),                     // table_id
                Cell::U64(1),                     // begin_snapshot
                Cell::Null,                       // end_snapshot
                Cell::Null,                       // file_order
                Cell::Str("data.parquet".into()), // path
                Cell::Bool(true),                 // path_is_relative
                Cell::Str("parquet".into()),      // file_format
                Cell::U64(10),                    // record_count
                Cell::U64(1024),                  // file_size_bytes
                Cell::U64(64),                    // footer_size
                Cell::U64(0),                     // row_id_start
                Cell::Null,                       // partition_id
                Cell::Str("ZGF0YS1rZXk=".into()), // encryption_key
                Cell::Null,                       // mapping_id
                Cell::Null,                       // partial_max
            ],
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::DeleteFile,
            cells: vec![
                Cell::U64(2),                         // delete_file_id
                Cell::U64(1),                         // table_id
                Cell::U64(1),                         // begin_snapshot
                Cell::Null,                           // end_snapshot
                Cell::U64(1),                         // data_file_id
                Cell::Str("delete.parquet".into()),   // path
                Cell::Bool(true),                     // path_is_relative
                Cell::Str("parquet".into()),          // format
                Cell::U64(2),                         // delete_count
                Cell::U64(128),                       // file_size_bytes
                Cell::U64(32),                        // footer_size
                Cell::Str("ZGVsZXRlLWtleQ==".into()), // encryption_key
                Cell::Null,                           // partial_max
            ],
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::Snapshot,
            cells: snapshot_row(1, 1, 2),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::SnapshotChanges,
            cells: snapshot_changes_row(1, r#"created_table:"main"."t""#),
        });
        tx.commit().await.unwrap();

        let head = catalog.snapshot().await.unwrap();
        let table = head.tables_in(crate::catalog::SchemaId::new(0))[0].id;
        assert_eq!(
            head.data_files_of(table)[0].encryption_key.as_deref(),
            Some("ZGF0YS1rZXk=")
        );
        assert_eq!(
            head.delete_files_of(table)[0].encryption_key.as_deref(),
            Some("ZGVsZXRlLWtleQ==")
        );
    }

    /// An `UPDATE ... SET end_snapshot` row ends a live table version:
    /// the old row moves to `history`, the new one lands in `current`, exactly
    /// the lifecycle convention this path interprets.
    #[tokio::test]
    async fn update_set_end_moves_the_old_version_to_history() {
        let catalog = open().await;

        // Seed schema `s` (id 1) and table `t` (id 1) via a plain insert.
        let db_tx1 = catalog.begin_write_tx().await.unwrap();
        let mut setup = StagedTransaction::begin_detached(db_tx1);
        setup.stage(RowOperation::Insert {
            table: TableKind::Schema,
            cells: schema_row(1, "s", 1),
        });
        setup.stage(RowOperation::Insert {
            table: TableKind::Table,
            cells: table_row(1, 1, "t_old", 1, None),
        });
        setup.stage(RowOperation::Insert {
            table: TableKind::Snapshot,
            cells: snapshot_row(1, 1, 2),
        });
        setup.stage(RowOperation::Insert {
            table: TableKind::SnapshotChanges,
            cells: snapshot_changes_row(1, r#"created_schema:"s""#),
        });
        setup.commit().await.unwrap();

        // Rename: end the old table version, insert the renamed one.
        let db_tx2 = catalog.begin_write_tx().await.unwrap();
        let mut rename = StagedTransaction::begin_detached(db_tx2);
        rename.stage(RowOperation::UpdateSetEnd {
            table: TableKind::Table,
            cells: vec![Cell::U64(1), Cell::U64(2)],
        });
        rename.stage(RowOperation::Insert {
            table: TableKind::Table,
            cells: table_row(1, 1, "t_new", 2, None),
        });
        rename.stage(RowOperation::Insert {
            table: TableKind::Snapshot,
            cells: snapshot_row(2, 1, 2),
        });
        rename.stage(RowOperation::Insert {
            table: TableKind::SnapshotChanges,
            cells: snapshot_changes_row(2, "altered_table:1"),
        });
        rename.commit().await.unwrap();

        let head = catalog.snapshot().await.unwrap();
        assert_eq!(
            head.tables_in(crate::catalog::SchemaId::new(1))[0].name,
            "t_new"
        );

        let past = catalog
            .snapshot_at(crate::catalog::SnapshotId::new(1))
            .await
            .unwrap();
        assert_eq!(
            past.tables_in(crate::catalog::SchemaId::new(1))[0].name,
            "t_old"
        );
    }

    /// A rename staged in DuckLake's live order — the new version's
    /// insert *before* the old version's end — keeps the new version
    /// live. Translation applies ends before inserts, so the shared `current`
    /// key resolves to the insert regardless of stage order.
    #[tokio::test]
    async fn rename_survives_insert_before_end_order() {
        let catalog = open().await;

        let db_tx1 = catalog.begin_write_tx().await.unwrap();
        let mut setup = StagedTransaction::begin_detached(db_tx1);
        setup.stage(RowOperation::Insert {
            table: TableKind::Schema,
            cells: schema_row(1, "s", 1),
        });
        setup.stage(RowOperation::Insert {
            table: TableKind::Table,
            cells: table_row(1, 1, "t_old", 1, None),
        });
        setup.stage(RowOperation::Insert {
            table: TableKind::Snapshot,
            cells: snapshot_row(1, 1, 2),
        });
        setup.stage(RowOperation::Insert {
            table: TableKind::SnapshotChanges,
            cells: snapshot_changes_row(1, r#"created_schema:"s""#),
        });
        setup.commit().await.unwrap();

        // Insert the renamed version first, then end the old one — the
        // reverse of the safe order, matching what DuckLake emits.
        let db_tx2 = catalog.begin_write_tx().await.unwrap();
        let mut rename = StagedTransaction::begin_detached(db_tx2);
        rename.stage(RowOperation::Insert {
            table: TableKind::Table,
            cells: table_row(1, 1, "t_new", 2, None),
        });
        rename.stage(RowOperation::UpdateSetEnd {
            table: TableKind::Table,
            cells: vec![Cell::U64(1), Cell::U64(2)],
        });
        rename.stage(RowOperation::Insert {
            table: TableKind::Snapshot,
            cells: snapshot_row(2, 1, 2),
        });
        rename.stage(RowOperation::Insert {
            table: TableKind::SnapshotChanges,
            cells: snapshot_changes_row(2, "altered_table:1"),
        });
        rename.commit().await.unwrap();

        let head = catalog.snapshot().await.unwrap();
        let live = head.tables_in(crate::catalog::SchemaId::new(1));
        assert_eq!(live.len(), 1, "exactly one live table after rename");
        assert_eq!(live[0].name, "t_new");
    }

    /// A lost race at commit is never retried: the loser's error carries
    /// the literal substring `conflict`, and the store is left exactly as
    /// the winner left it.
    #[tokio::test]
    async fn lost_race_is_not_retried_and_carries_conflict_text() {
        let catalog = open().await;

        let tx_a = catalog.begin_write_tx().await.unwrap();
        let tx_b = catalog.begin_write_tx().await.unwrap();
        let mut a = StagedTransaction::begin_detached(tx_a);
        let mut b = StagedTransaction::begin_detached(tx_b);

        for (tx, name) in [(&mut a, "a"), (&mut b, "b")] {
            tx.stage(RowOperation::Insert {
                table: TableKind::Schema,
                cells: schema_row(1, name, 1),
            });
            tx.stage(RowOperation::Insert {
                table: TableKind::Snapshot,
                cells: snapshot_row(1, 1, 2),
            });
            tx.stage(RowOperation::Insert {
                table: TableKind::SnapshotChanges,
                cells: snapshot_changes_row(1, format!(r#"created_schema:"{name}""#).as_str()),
            });
        }

        a.commit().await.unwrap();
        let err = b.commit().await.unwrap_err();
        assert!(
            err.to_string().contains("conflict"),
            "error must carry the literal substring `conflict`: {err}"
        );

        // The store reflects only the winner: schema `a`, not `b`.
        let head = catalog.snapshot().await.unwrap();
        assert!(head.schema_by_name("a").is_some());
        assert!(head.schema_by_name("b").is_none());
    }

    /// A malformed staged row (wrong cell count) fails loudly as
    /// `Corruption` rather than panicking or silently truncating.
    #[tokio::test]
    async fn malformed_row_is_corruption_not_a_panic() {
        let catalog = open().await;
        let db_tx = catalog.begin_write_tx().await.unwrap();
        let mut tx = StagedTransaction::begin_detached(db_tx);
        tx.stage(RowOperation::Insert {
            table: TableKind::Schema,
            cells: vec![Cell::U64(1)], // far too few cells
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::Snapshot,
            cells: snapshot_row(1, 1, 2),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::SnapshotChanges,
            cells: snapshot_changes_row(1, ""),
        });
        let err = tx.commit().await.unwrap_err();
        assert!(matches!(err, Error::Corruption(_)));
    }

    /// Stages an inline schema plus two inserts against the same
    /// `(table_id, schema_version, begin_snapshot)` in one commit: the
    /// chunks land with sequential `chunk_seq` (stage order), and the
    /// schema is readable back verbatim.
    #[tokio::test]
    async fn stages_inline_schema_and_sequential_inserts() {
        let catalog = open().await;
        let db_tx = catalog.begin_write_tx().await.unwrap();
        let mut tx = StagedTransaction::begin_detached(db_tx);

        tx.stage(RowOperation::InlineSchema {
            table_id: 1,
            schema_version: 0,
            arrow_schema: b"schema".to_vec(),
        });
        tx.stage(RowOperation::InlineInsert {
            table_id: 1,
            schema_version: 0,
            begin_snapshot: 1,
            row_id_start: 0,
            row_count: 2,
            arrow_body: b"chunk-a".to_vec(),
        });
        tx.stage(RowOperation::InlineInsert {
            table_id: 1,
            schema_version: 0,
            begin_snapshot: 1,
            row_id_start: 2,
            row_count: 1,
            arrow_body: b"chunk-b".to_vec(),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::Snapshot,
            cells: snapshot_row(1, 0, 1),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::SnapshotChanges,
            cells: snapshot_changes_row(1, "inlined_insert:1"),
        });
        tx.commit().await.unwrap();

        let tx = catalog.begin_write_tx().await.unwrap();
        let chunks = store_inline::scan_inline_chunks(ReadHandle::Tx(&tx), 1)
            .await
            .unwrap();
        assert_eq!(chunks.len(), 2);
        assert_eq!(
            chunks[0].0,
            InlineOperation::Insert {
                table_id: 1,
                schema_version: 0,
                begin_snapshot: 1,
                chunk_seq: 0,
            }
        );
        assert_eq!(chunks[0].1.body, b"chunk-a");
        assert_eq!(chunks[0].1.row_id_start, 0);
        assert_eq!(chunks[0].1.row_count, 2);
        assert_eq!(
            chunks[1].0,
            InlineOperation::Insert {
                table_id: 1,
                schema_version: 0,
                begin_snapshot: 1,
                chunk_seq: 1,
            }
        );
        assert_eq!(chunks[1].1.body, b"chunk-b");

        let schemas = store_inline::scan_inline_schemas(ReadHandle::Tx(&tx), 1)
            .await
            .unwrap();
        assert_eq!(
            schemas,
            vec![(
                0,
                proto::InlineSchemaValue {
                    arrow_schema: b"schema".to_vec(),
                }
            )]
        );
        tx.rollback();
    }

    /// An `InlineIdel` tombstones a row: the row is absent from a
    /// `Table`-kind materialization at or after its `end_snapshot`.
    #[tokio::test]
    async fn stages_inline_idel_and_row_disappears_from_table_scan_after_it() {
        let catalog = open().await;

        let db_tx1 = catalog.begin_write_tx().await.unwrap();
        let mut setup = StagedTransaction::begin_detached(db_tx1);
        setup.stage(RowOperation::InlineInsert {
            table_id: 1,
            schema_version: 0,
            begin_snapshot: 1,
            row_id_start: 0,
            row_count: 2,
            arrow_body: b"chunk".to_vec(),
        });
        setup.stage(RowOperation::Insert {
            table: TableKind::Snapshot,
            cells: snapshot_row(1, 0, 1),
        });
        setup.stage(RowOperation::Insert {
            table: TableKind::SnapshotChanges,
            cells: snapshot_changes_row(1, "inlined_insert:1"),
        });
        setup.commit().await.unwrap();

        let db_tx2 = catalog.begin_write_tx().await.unwrap();
        let mut inline_delete = StagedTransaction::begin_detached(db_tx2);
        inline_delete.stage(RowOperation::InlineInlineDelete {
            table_id: 1,
            row_id: 0,
            end_snapshot: 2,
        });
        inline_delete.stage(RowOperation::Insert {
            table: TableKind::Snapshot,
            cells: snapshot_row(2, 0, 1),
        });
        inline_delete.stage(RowOperation::Insert {
            table: TableKind::SnapshotChanges,
            cells: snapshot_changes_row(2, "inlined_delete:1"),
        });
        inline_delete.commit().await.unwrap();

        let tx = catalog.begin_write_tx().await.unwrap();
        let chunks = store_inline::scan_inline_chunks(ReadHandle::Tx(&tx), 1)
            .await
            .unwrap();
        let inline_deletes = store_inline::scan_inline_inline_deletes(ReadHandle::Tx(&tx), 1)
            .await
            .unwrap();
        tx.rollback();
        assert_eq!(
            inline_deletes,
            vec![(0, proto::InlineInlineDeleteValue { end_snapshot: 2 })]
        );

        let rows = materialize_inline_rows(&chunks, &inline_deletes);
        assert_eq!(
            InlineScanKind::Table
                .select(&rows, 1, 0)
                .iter()
                .map(|r| r.row_id)
                .collect::<Vec<_>>(),
            vec![0, 1]
        );
        assert_eq!(
            InlineScanKind::Table
                .select(&rows, 2, 0)
                .iter()
                .map(|r| r.row_id)
                .collect::<Vec<_>>(),
            vec![1]
        );
    }

    /// `InlineFlushDelete` removes chunks begun at or before the flush
    /// snapshot for the named schema version, plus the `inline/inline_delete`
    /// tombstones those chunks' rows consumed — a later schema version's
    /// chunk (begun after the flush point) survives untouched.
    #[tokio::test]
    async fn stages_inline_flush_delete_removes_flushed_chunks_and_their_idels() {
        let catalog = open().await;

        let db_tx1 = catalog.begin_write_tx().await.unwrap();
        let mut setup = StagedTransaction::begin_detached(db_tx1);
        setup.stage(RowOperation::InlineInsert {
            table_id: 1,
            schema_version: 0,
            begin_snapshot: 1,
            row_id_start: 0,
            row_count: 2,
            arrow_body: b"chunk".to_vec(),
        });
        setup.stage(RowOperation::Insert {
            table: TableKind::Snapshot,
            cells: snapshot_row(1, 0, 1),
        });
        setup.stage(RowOperation::Insert {
            table: TableKind::SnapshotChanges,
            cells: snapshot_changes_row(1, "inlined_insert:1"),
        });
        setup.commit().await.unwrap();

        // A later commit tombstones one row (a tombstone only ever ends a
        // version begun before it — DuckLake's writer never stamps a row
        // with its own insertion snapshot).
        let db_tx2 = catalog.begin_write_tx().await.unwrap();
        let mut delete = StagedTransaction::begin_detached(db_tx2);
        delete.stage(RowOperation::InlineInlineDelete {
            table_id: 1,
            row_id: 0,
            end_snapshot: 2,
        });
        delete.stage(RowOperation::Insert {
            table: TableKind::Snapshot,
            cells: snapshot_row(2, 0, 1),
        });
        delete.stage(RowOperation::Insert {
            table: TableKind::SnapshotChanges,
            cells: snapshot_changes_row(2, "inlined_delete:1"),
        });
        delete.commit().await.unwrap();

        let db_tx3 = catalog.begin_write_tx().await.unwrap();
        let mut flush = StagedTransaction::begin_detached(db_tx3);
        flush.stage(RowOperation::InlineFlushDelete {
            table_id: 1,
            schema_version: 0,
            flush_snapshot: 2,
        });
        flush.stage(RowOperation::Insert {
            table: TableKind::Snapshot,
            cells: snapshot_row(3, 0, 1),
        });
        flush.stage(RowOperation::Insert {
            table: TableKind::SnapshotChanges,
            cells: snapshot_changes_row(3, "flushed_inlined_data:1"),
        });
        flush.commit().await.unwrap();

        let tx = catalog.begin_write_tx().await.unwrap();
        let chunks = store_inline::scan_inline_chunks(ReadHandle::Tx(&tx), 1)
            .await
            .unwrap();
        let inline_deletes = store_inline::scan_inline_inline_deletes(ReadHandle::Tx(&tx), 1)
            .await
            .unwrap();
        tx.rollback();
        assert!(chunks.is_empty(), "flushed chunk must be gone: {chunks:?}");
        assert!(
            inline_deletes.is_empty(),
            "consumed inline_delete must be gone: {inline_deletes:?}"
        );
    }

    /// `InlineDrop` removes every `inline/*` record for the table:
    /// schema, chunks, and tombstones.
    #[tokio::test]
    async fn stages_inline_drop_removes_every_record_for_the_table() {
        let catalog = open().await;

        let db_tx1 = catalog.begin_write_tx().await.unwrap();
        let mut setup = StagedTransaction::begin_detached(db_tx1);
        setup.stage(RowOperation::InlineSchema {
            table_id: 1,
            schema_version: 0,
            arrow_schema: b"schema".to_vec(),
        });
        setup.stage(RowOperation::InlineInsert {
            table_id: 1,
            schema_version: 0,
            begin_snapshot: 1,
            row_id_start: 0,
            row_count: 1,
            arrow_body: b"chunk".to_vec(),
        });
        setup.stage(RowOperation::InlineFileDelete {
            table_id: 1,
            data_file_id: 9,
            row_id: 5,
            begin_snapshot: 1,
        });
        setup.stage(RowOperation::Insert {
            table: TableKind::Snapshot,
            cells: snapshot_row(1, 0, 1),
        });
        setup.stage(RowOperation::Insert {
            table: TableKind::SnapshotChanges,
            cells: snapshot_changes_row(1, "inlined_insert:1"),
        });
        setup.commit().await.unwrap();

        let db_tx2 = catalog.begin_write_tx().await.unwrap();
        let mut drop_tx = StagedTransaction::begin_detached(db_tx2);
        drop_tx.stage(RowOperation::InlineDrop { table_id: 1 });
        drop_tx.stage(RowOperation::Insert {
            table: TableKind::Snapshot,
            cells: snapshot_row(2, 0, 1),
        });
        drop_tx.stage(RowOperation::Insert {
            table: TableKind::SnapshotChanges,
            cells: snapshot_changes_row(2, r#"dropped_table:"main"."t""#),
        });
        drop_tx.commit().await.unwrap();

        let tx = catalog.begin_write_tx().await.unwrap();
        let chunks = store_inline::scan_inline_chunks(ReadHandle::Tx(&tx), 1)
            .await
            .unwrap();
        let file_deletes = store_inline::scan_inline_file_deletes(ReadHandle::Tx(&tx), 1)
            .await
            .unwrap();
        let schemas = store_inline::scan_inline_schemas(ReadHandle::Tx(&tx), 1)
            .await
            .unwrap();
        tx.rollback();
        assert!(chunks.is_empty());
        assert!(file_deletes.is_empty());
        assert!(schemas.is_empty());
    }

    /// `InlineSchemaDrop` removes only the named schema version's
    /// `inline/schema` record, leaving a different schema version's
    /// record (and its chunks) untouched — the scoped cleanup a
    /// superseded-inlined-table flush needs, as opposed to `InlineDrop`'s
    /// whole-table sweep.
    #[tokio::test]
    async fn stages_inline_schema_drop_removes_only_the_named_schema_version() {
        let catalog = open().await;

        let db_tx1 = catalog.begin_write_tx().await.unwrap();
        let mut setup = StagedTransaction::begin_detached(db_tx1);
        setup.stage(RowOperation::InlineSchema {
            table_id: 1,
            schema_version: 0,
            arrow_schema: b"schema-v0".to_vec(),
        });
        setup.stage(RowOperation::InlineSchema {
            table_id: 1,
            schema_version: 1,
            arrow_schema: b"schema-v1".to_vec(),
        });
        setup.stage(RowOperation::InlineInsert {
            table_id: 1,
            schema_version: 1,
            begin_snapshot: 1,
            row_id_start: 0,
            row_count: 1,
            arrow_body: b"chunk".to_vec(),
        });
        setup.stage(RowOperation::Insert {
            table: TableKind::Snapshot,
            cells: snapshot_row(1, 0, 1),
        });
        setup.stage(RowOperation::Insert {
            table: TableKind::SnapshotChanges,
            cells: snapshot_changes_row(1, "inlined_insert:1"),
        });
        setup.commit().await.unwrap();

        let db_tx2 = catalog.begin_write_tx().await.unwrap();
        let mut drop_tx = StagedTransaction::begin_detached(db_tx2);
        drop_tx.stage(RowOperation::InlineSchemaDrop {
            table_id: 1,
            schema_version: 0,
        });
        drop_tx.stage(RowOperation::Insert {
            table: TableKind::Snapshot,
            cells: snapshot_row(2, 0, 1),
        });
        drop_tx.stage(RowOperation::Insert {
            table: TableKind::SnapshotChanges,
            cells: snapshot_changes_row(2, "flushed_inlined_data:1"),
        });
        drop_tx.commit().await.unwrap();

        let tx = catalog.begin_write_tx().await.unwrap();
        let schemas = store_inline::scan_inline_schemas(ReadHandle::Tx(&tx), 1)
            .await
            .unwrap();
        let chunks = store_inline::scan_inline_chunks(ReadHandle::Tx(&tx), 1)
            .await
            .unwrap();
        tx.rollback();
        assert_eq!(
            schemas,
            vec![(
                1,
                proto::InlineSchemaValue {
                    arrow_schema: b"schema-v1".to_vec()
                }
            )]
        );
        assert_eq!(chunks.len(), 1, "schema_version 1's chunk must survive");
    }

    fn partition_info_row(partition_id: u64, table_id: u64, begin: u64) -> Vec<Cell> {
        vec![
            Cell::U64(partition_id),
            Cell::U64(table_id),
            Cell::U64(begin),
            Cell::Null,
        ]
    }

    fn partition_column_row(
        partition_id: u64,
        table_id: u64,
        index: u64,
        column_id: u64,
    ) -> Vec<Cell> {
        vec![
            Cell::U64(partition_id),
            Cell::U64(table_id),
            Cell::U64(index),
            Cell::U64(column_id),
            Cell::Str("identity".to_string()),
        ]
    }

    fn file_partition_value_row(
        data_file_id: u64,
        table_id: u64,
        index: u64,
        value: &str,
    ) -> Vec<Cell> {
        vec![
            Cell::U64(data_file_id),
            Cell::U64(table_id),
            Cell::U64(index),
            Cell::Str(value.to_string()),
        ]
    }

    fn data_file_row(data_file_id: u64, table_id: u64, begin: u64) -> Vec<Cell> {
        vec![
            Cell::U64(data_file_id),
            Cell::U64(table_id),
            Cell::U64(begin),
            Cell::Null,                       // end_snapshot
            Cell::Null,                       // file_order
            Cell::Str("data.parquet".into()), // path
            Cell::Bool(true),                 // path_is_relative
            Cell::Str("parquet".into()),      // file_format
            Cell::U64(10),                    // record_count
            Cell::U64(1024),                  // file_size_bytes
            Cell::U64(64),                    // footer_size
            Cell::U64(0),                     // row_id_start
            Cell::Null,                       // partition_id
            Cell::Null,                       // encryption_key
            Cell::Null,                       // mapping_id
            Cell::Null,                       // partial_max
        ]
    }

    /// A partition spec, its columns, and a file's partition values land
    /// verbatim; repartitioning ends the old spec; time travel
    /// reconstructs the spec-in-force, and every file still reports the
    /// spec it was written under.
    #[tokio::test]
    #[allow(clippy::too_many_lines)]
    async fn partition_spec_rows_land_fold_and_time_travel() {
        let catalog = open().await;

        // Commit 1: table + column + spec 10 (identity on column 1) + one
        // data file written under it, carrying one partition value.
        let db_tx = catalog.begin_write_tx().await.unwrap();
        let mut tx = StagedTransaction::begin_detached(db_tx);
        tx.stage(RowOperation::Insert {
            table: TableKind::Table,
            cells: table_row(1, 0, "t", 1, None),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::Column,
            cells: column_row(1, 1, "part_key", 0),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::PartitionInfo,
            cells: partition_info_row(10, 1, 1),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::PartitionColumn,
            cells: partition_column_row(10, 1, 0, 1),
        });
        let mut file = data_file_row(1, 1, 1);
        file[12] = Cell::U64(10); // partition_id
        tx.stage(RowOperation::Insert {
            table: TableKind::DataFile,
            cells: file,
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::FilePartitionValue,
            cells: file_partition_value_row(1, 1, 0, "7"),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::Snapshot,
            cells: snapshot_row(1, 1, 11),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::SnapshotChanges,
            cells: snapshot_changes_row(1, r#"created_table:"main"."t""#),
        });
        tx.commit().await.unwrap();

        let head = catalog.snapshot().await.unwrap();
        let spec = &head.partitions[&1][&10];
        assert_eq!(spec.begin_snapshot, 1);
        assert_eq!(spec.columns.len(), 1);
        assert_eq!(spec.columns[0].column_id, 1);
        assert_eq!(spec.columns[0].transform, "identity");
        let stored = &head.data_files[&1][&1];
        assert_eq!(stored.partition_id, Some(10));
        assert_eq!(stored.partition_values.len(), 1);
        assert_eq!(stored.partition_values[0].partition_value, "7");

        // Commit 2: repartition — end spec 10, insert spec 11.
        let db_tx = catalog.begin_write_tx().await.unwrap();
        let mut tx = StagedTransaction::begin_detached(db_tx);
        tx.stage(RowOperation::UpdateSetEnd {
            table: TableKind::PartitionInfo,
            cells: vec![Cell::U64(1), Cell::U64(10), Cell::U64(2)],
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::PartitionInfo,
            cells: partition_info_row(11, 1, 2),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::PartitionColumn,
            cells: partition_column_row(11, 1, 0, 1),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::Snapshot,
            cells: snapshot_row(2, 1, 12),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::SnapshotChanges,
            cells: snapshot_changes_row(2, "altered_table:1"),
        });
        tx.commit().await.unwrap();

        let head = catalog.snapshot().await.unwrap();
        assert!(!head.partitions[&1].contains_key(&10));
        assert!(head.partitions[&1].contains_key(&11));
        // The file still names the spec it was written under.
        assert_eq!(head.data_files[&1][&1].partition_id, Some(10));

        // Time travel reconstructs spec 10 at snapshot 1.
        let at_one = catalog
            .snapshot_at(crate::catalog::SnapshotId::new(1))
            .await
            .unwrap();
        assert!(at_one.partitions[&1].contains_key(&10));
        assert!(!at_one.partitions[&1].contains_key(&11));

        // The dump surface serves current and history rows unfiltered.
        let specs = crate::ffi_support::dump_partition_info(&catalog)
            .await
            .unwrap();
        assert!(
            specs
                .iter()
                .any(|p| p.partition_id == 10 && p.end_snapshot == Some(2))
        );
        assert!(
            specs
                .iter()
                .any(|p| p.partition_id == 11 && p.end_snapshot.is_none())
        );

        // Commit 3: clear — end spec 11, insert nothing.
        let db_tx = catalog.begin_write_tx().await.unwrap();
        let mut tx = StagedTransaction::begin_detached(db_tx);
        tx.stage(RowOperation::UpdateSetEnd {
            table: TableKind::PartitionInfo,
            cells: vec![Cell::U64(1), Cell::U64(11), Cell::U64(3)],
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::Snapshot,
            cells: snapshot_row(3, 1, 12),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::SnapshotChanges,
            cells: snapshot_changes_row(3, "altered_table:1"),
        });
        tx.commit().await.unwrap();

        let head = catalog.snapshot().await.unwrap();
        assert!(
            head.partitions
                .get(&1)
                .is_none_or(std::collections::BTreeMap::is_empty)
        );

        catalog.close().await.unwrap();
    }

    fn sort_info_row(sort_id: u64, table_id: u64, begin: u64) -> Vec<Cell> {
        vec![
            Cell::U64(sort_id),
            Cell::U64(table_id),
            Cell::U64(begin),
            Cell::Null,
        ]
    }

    fn sort_expression_row(sort_id: u64, table_id: u64, index: u64, expression: &str) -> Vec<Cell> {
        vec![
            Cell::U64(sort_id),
            Cell::U64(table_id),
            Cell::U64(index),
            Cell::Str(expression.to_string()),
            Cell::Str("duckdb".to_string()),
            Cell::Str("DESC".to_string()),
            Cell::Str("NULLS_FIRST".to_string()),
        ]
    }

    /// A sort spec and its expressions land verbatim — direction, null
    /// order, and dialect untouched; re-sorting ends the old spec; time
    /// travel reconstructs the spec-in-force.
    #[tokio::test]
    async fn sort_spec_rows_land_fold_and_time_travel() {
        let catalog = open().await;

        let db_tx = catalog.begin_write_tx().await.unwrap();
        let mut tx = StagedTransaction::begin_detached(db_tx);
        tx.stage(RowOperation::Insert {
            table: TableKind::Table,
            cells: table_row(1, 0, "t", 1, None),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::Column,
            cells: column_row(1, 1, "v", 0),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::SortInfo,
            cells: sort_info_row(20, 1, 1),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::SortExpression,
            cells: sort_expression_row(20, 1, 0, "v"),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::Snapshot,
            cells: snapshot_row(1, 1, 21),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::SnapshotChanges,
            cells: snapshot_changes_row(1, r#"created_table:"main"."t""#),
        });
        tx.commit().await.unwrap();

        let head = catalog.snapshot().await.unwrap();
        let spec = &head.sorts[&1][&20];
        assert_eq!(spec.begin_snapshot, 1);
        assert_eq!(spec.expressions.len(), 1);
        assert_eq!(spec.expressions[0].expression, "v");
        assert_eq!(spec.expressions[0].dialect, "duckdb");
        assert_eq!(spec.expressions[0].sort_direction, "DESC");
        assert_eq!(spec.expressions[0].null_order, "NULLS_FIRST");

        // End spec 20, insert spec 21 — the snapshot row keeps the same
        // schema_version: DuckLake does not bump it for sort changes.
        let db_tx = catalog.begin_write_tx().await.unwrap();
        let mut tx = StagedTransaction::begin_detached(db_tx);
        tx.stage(RowOperation::UpdateSetEnd {
            table: TableKind::SortInfo,
            cells: vec![Cell::U64(1), Cell::U64(20), Cell::U64(2)],
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::SortInfo,
            cells: sort_info_row(21, 1, 2),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::SortExpression,
            cells: sort_expression_row(21, 1, 0, "v || 'x'"),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::Snapshot,
            cells: snapshot_row(2, 1, 22),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::SnapshotChanges,
            cells: snapshot_changes_row(2, "altered_table:1"),
        });
        tx.commit().await.unwrap();

        let head = catalog.snapshot().await.unwrap();
        assert!(!head.sorts[&1].contains_key(&20));
        assert_eq!(head.sorts[&1][&21].expressions[0].expression, "v || 'x'");

        let at_one = catalog
            .snapshot_at(crate::catalog::SnapshotId::new(1))
            .await
            .unwrap();
        assert!(at_one.sorts[&1].contains_key(&20));
        assert!(!at_one.sorts[&1].contains_key(&21));

        // The dump surface serves current and history rows unfiltered.
        let specs = crate::ffi_support::dump_sort_info(&catalog).await.unwrap();
        assert!(
            specs
                .iter()
                .any(|s| s.sort_id == 20 && s.end_snapshot == Some(2))
        );
        assert!(
            specs
                .iter()
                .any(|s| s.sort_id == 21 && s.end_snapshot.is_none())
        );

        catalog.close().await.unwrap();
    }

    /// A `partition_column` row whose spec is not inserted in the same
    /// commit is a shape error, not a silent drop.
    #[tokio::test]
    async fn orphaned_partition_column_row_is_rejected() {
        let catalog = open().await;
        let db_tx = catalog.begin_write_tx().await.unwrap();
        let mut tx = StagedTransaction::begin_detached(db_tx);
        tx.stage(RowOperation::Insert {
            table: TableKind::PartitionColumn,
            cells: partition_column_row(99, 1, 0, 1),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::Snapshot,
            cells: snapshot_row(1, 1, 2),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::SnapshotChanges,
            cells: snapshot_changes_row(1, "none"),
        });
        let err = tx.commit().await.unwrap_err();
        assert!(err.to_string().contains("partition"), "{err}");
        catalog.close().await.unwrap();
    }

    fn tag_row(object_id: u64, begin: u64, key: &str, value: &str) -> Vec<Cell> {
        vec![
            Cell::U64(object_id),
            Cell::U64(begin),
            Cell::Null,
            Cell::Str(key.to_string()),
            Cell::Str(value.to_string()),
        ]
    }

    fn column_tag_row(
        table_id: u64,
        column_id: u64,
        begin: u64,
        key: &str,
        value: &str,
    ) -> Vec<Cell> {
        vec![
            Cell::U64(table_id),
            Cell::U64(column_id),
            Cell::U64(begin),
            Cell::Null,
            Cell::Str(key.to_string()),
            Cell::Str(value.to_string()),
        ]
    }

    async fn seed_table(catalog: &Catalog) {
        let db_tx = catalog.begin_write_tx().await.unwrap();
        let mut tx = StagedTransaction::begin_detached(db_tx);
        tx.stage(RowOperation::Insert {
            table: TableKind::Table,
            cells: table_row(1, 0, "t", 1, None),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::Column,
            cells: column_row(1, 1, "a", 0),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::Snapshot,
            cells: snapshot_row(1, 1, 2),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::SnapshotChanges,
            cells: snapshot_changes_row(1, r#"created_table:"main"."t""#),
        });
        tx.commit().await.unwrap();
    }

    /// A re-comment (`COMMENT ON` twice) is DuckLake's set-end + insert
    /// pair: the old entry ends, the new one lands live, both kept in the
    /// object's container for time travel.
    #[tokio::test]
    async fn tag_rows_land_and_a_recomment_ends_the_old_entry() {
        let catalog = open().await;
        seed_table(&catalog).await;

        let db_tx = catalog.begin_write_tx().await.unwrap();
        let mut tx = StagedTransaction::begin_detached(db_tx);
        tx.stage(RowOperation::Insert {
            table: TableKind::Tag,
            cells: tag_row(1, 2, "comment", "first"),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::Snapshot,
            cells: snapshot_row(2, 1, 2),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::SnapshotChanges,
            cells: snapshot_changes_row(2, "altered_table:1"),
        });
        tx.commit().await.unwrap();

        let db_tx = catalog.begin_write_tx().await.unwrap();
        let mut tx = StagedTransaction::begin_detached(db_tx);
        tx.stage(RowOperation::UpdateSetEnd {
            table: TableKind::Tag,
            cells: vec![Cell::U64(1), Cell::Str("comment".into()), Cell::U64(3)],
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::Tag,
            cells: tag_row(1, 3, "comment", "second"),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::Snapshot,
            cells: snapshot_row(3, 1, 2),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::SnapshotChanges,
            cells: snapshot_changes_row(3, "altered_table:1"),
        });
        tx.commit().await.unwrap();

        let head = catalog.snapshot().await.unwrap();
        let tags = head.tags_of(1);
        assert_eq!(tags.len(), 2);
        assert_eq!(
            (tags[0].value.as_str(), tags[0].end_snapshot),
            ("first", Some(3))
        );
        assert_eq!(
            (tags[1].value.as_str(), tags[1].end_snapshot),
            ("second", None)
        );
        catalog.close().await.unwrap();
    }

    /// A column tag rides its column's record without minting a column
    /// version: after tagging, the column still has exactly one row on
    /// the dump surface, and the tag ends in place on a re-comment.
    #[tokio::test]
    async fn column_tags_ride_the_column_record_without_a_version_transition() {
        let catalog = open().await;
        seed_table(&catalog).await;

        let db_tx = catalog.begin_write_tx().await.unwrap();
        let mut tx = StagedTransaction::begin_detached(db_tx);
        tx.stage(RowOperation::Insert {
            table: TableKind::ColumnTag,
            cells: column_tag_row(1, 1, 2, "comment", "col comment"),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::Snapshot,
            cells: snapshot_row(2, 1, 2),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::SnapshotChanges,
            cells: snapshot_changes_row(2, "altered_table:1"),
        });
        tx.commit().await.unwrap();

        let columns = crate::ffi_support::dump_columns(&catalog).await.unwrap();
        let rows: Vec<_> = columns.iter().filter(|c| c.column_id == 1).collect();
        assert_eq!(rows.len(), 1, "a tag change must not mint a column version");
        assert_eq!(rows[0].begin_snapshot, 0);
        assert!(rows[0].end_snapshot.is_none());
        assert_eq!(rows[0].tags.len(), 1);
        assert_eq!(rows[0].tags[0].value, "col comment");
        catalog.close().await.unwrap();
    }

    /// A column version transition (rename) carries the prior version's
    /// tag entries onto the new current record — DuckLake keys column
    /// tags by (table, column) with their own lifecycle, so an alter
    /// never re-authors them.
    #[tokio::test]
    async fn column_alter_carries_tags_forward() {
        let catalog = open().await;
        seed_table(&catalog).await;

        let db_tx = catalog.begin_write_tx().await.unwrap();
        let mut tx = StagedTransaction::begin_detached(db_tx);
        tx.stage(RowOperation::Insert {
            table: TableKind::ColumnTag,
            cells: column_tag_row(1, 1, 2, "comment", "kept"),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::Snapshot,
            cells: snapshot_row(2, 1, 2),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::SnapshotChanges,
            cells: snapshot_changes_row(2, "altered_table:1"),
        });
        tx.commit().await.unwrap();

        // Rename the column: end the old version, insert the new one.
        let db_tx = catalog.begin_write_tx().await.unwrap();
        let mut tx = StagedTransaction::begin_detached(db_tx);
        tx.stage(RowOperation::UpdateSetEnd {
            table: TableKind::Column,
            cells: vec![Cell::U64(1), Cell::U64(1), Cell::U64(3)],
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::Column,
            cells: {
                let mut cells = column_row(1, 1, "renamed", 0);
                cells[1] = Cell::U64(3); // begin_snapshot
                cells
            },
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::Snapshot,
            cells: snapshot_row(3, 2, 2),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::SnapshotChanges,
            cells: snapshot_changes_row(3, "altered_table:1"),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::SchemaVersions,
            cells: vec![Cell::U64(3), Cell::U64(2), Cell::U64(1)],
        });
        tx.commit().await.unwrap();

        let head = catalog.snapshot().await.unwrap();
        let column = &head.columns[&1][&1];
        assert_eq!(column.column_name, "renamed");
        assert_eq!(column.tags.len(), 1);
        assert_eq!(column.tags[0].value, "kept");
        catalog.close().await.unwrap();
    }

    /// Seeds `t` (table 1) with data file 9 (snapshot 1), then expires the
    /// file's live version at snapshot 2 — leaving one dead history row,
    /// the fixture the expiry tests prune.
    async fn seed_expired_file(catalog: &Catalog) {
        let db_tx = catalog.begin_write_tx().await.unwrap();
        let mut tx = StagedTransaction::begin_detached(db_tx);
        tx.stage(RowOperation::Insert {
            table: TableKind::Table,
            cells: table_row(1, 0, "t", 1, None),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::Column,
            cells: column_row(1, 1, "a", 0),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::DataFile,
            cells: data_file_row(9, 1, 1),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::Snapshot,
            cells: snapshot_row(1, 1, 2),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::SnapshotChanges,
            cells: snapshot_changes_row(1, r#"created_table:"main"."t""#),
        });
        tx.commit().await.unwrap();

        let db_tx = catalog.begin_write_tx().await.unwrap();
        let mut tx = StagedTransaction::begin_detached(db_tx);
        tx.stage(RowOperation::UpdateSetEnd {
            table: TableKind::DataFile,
            cells: vec![Cell::U64(1), Cell::U64(9), Cell::U64(2)],
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::Snapshot,
            cells: snapshot_row(2, 1, 2),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::SnapshotChanges,
            cells: snapshot_changes_row(2, "deleted_from_table:1"),
        });
        tx.commit().await.unwrap();
    }

    /// The expiry cascade's staged shape: delete a dead snapshot, prune
    /// the history row it alone saw, and schedule the file's bytes — all
    /// in one head-preserving maintenance commit.
    #[tokio::test]
    async fn expiry_prunes_history_and_schedules_files_without_advancing_head() {
        let catalog = open().await;
        seed_expired_file(&catalog).await;

        let db_tx = catalog.begin_write_tx().await.unwrap();
        let mut tx = StagedTransaction::begin_detached(db_tx);
        tx.stage(RowOperation::Delete {
            table: TableKind::Snapshot,
            cells: vec![Cell::U64(1)],
        });
        tx.stage(RowOperation::Delete {
            table: TableKind::SnapshotChanges,
            cells: vec![Cell::U64(1)],
        });
        tx.stage(RowOperation::Delete {
            table: TableKind::DataFile,
            cells: vec![Cell::U64(1), Cell::U64(9), Cell::U64(2)],
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::FilesScheduledForDeletion,
            cells: vec![
                Cell::U64(9),
                Cell::Str("f9.parquet".to_string()),
                Cell::Bool(true),
                Cell::I64(1_000),
            ],
        });
        let id = tx.commit().await.unwrap();
        assert_eq!(id.get(), 2, "maintenance must not advance head");

        let head = catalog.snapshot().await.unwrap();
        assert_eq!(head.current_snapshot().id.get(), 2);
        let schedule = head.scheduled_deletions();
        assert_eq!(schedule.len(), 1);
        assert_eq!(schedule[0].data_file_id, 9);
        assert_eq!(schedule[0].path, "f9.parquet");

        // The dead snapshot no longer resolves; the survivor does, and
        // the pruned history row is gone from the dump surface.
        let expired = catalog
            .snapshot_at(crate::catalog::SnapshotId::new(1))
            .await
            .unwrap_err();
        assert!(matches!(expired, Error::NotFound(_)), "{expired}");
        let surviving = catalog
            .snapshot_at(crate::catalog::SnapshotId::new(2))
            .await
            .unwrap();
        assert!(
            surviving
                .data_files_of(crate::catalog::TableId::new(1))
                .is_empty()
        );
        let files = crate::ffi_support::dump_data_files(&catalog).await.unwrap();
        assert!(files.is_empty(), "history row must be pruned: {files:?}");

        catalog.close().await.unwrap();
    }

    /// Cleanup's staged shape: after DuckDB deletes the bytes, the
    /// schedule row is forgotten in a head-preserving commit.
    #[tokio::test]
    async fn cleanup_forgets_the_schedule() {
        let catalog = open().await;
        seed_expired_file(&catalog).await;

        let db_tx = catalog.begin_write_tx().await.unwrap();
        let mut tx = StagedTransaction::begin_detached(db_tx);
        tx.stage(RowOperation::Insert {
            table: TableKind::FilesScheduledForDeletion,
            cells: vec![
                Cell::U64(9),
                Cell::Str("f9.parquet".to_string()),
                Cell::Bool(true),
                Cell::I64(1_000),
            ],
        });
        tx.commit().await.unwrap();

        let db_tx = catalog.begin_write_tx().await.unwrap();
        let mut tx = StagedTransaction::begin_detached(db_tx);
        tx.stage(RowOperation::Delete {
            table: TableKind::FilesScheduledForDeletion,
            cells: vec![Cell::U64(9)],
        });
        tx.commit().await.unwrap();

        let head = catalog.snapshot().await.unwrap();
        assert!(head.scheduled_deletions().is_empty());
        assert_eq!(head.current_snapshot().id.get(), 2);
        catalog.close().await.unwrap();
    }

    /// The head snapshot can never be expired.
    #[tokio::test]
    async fn deleting_the_head_snapshot_is_rejected() {
        let catalog = open().await;
        seed_expired_file(&catalog).await;

        let db_tx = catalog.begin_write_tx().await.unwrap();
        let mut tx = StagedTransaction::begin_detached(db_tx);
        tx.stage(RowOperation::Delete {
            table: TableKind::Snapshot,
            cells: vec![Cell::U64(2)],
        });
        let err = tx.commit().await.unwrap_err();
        assert!(matches!(err, Error::Constraint(_)), "{err}");
        catalog.close().await.unwrap();
    }

    /// A commit that mutates entities without minting a snapshot is not a
    /// maintenance commit — it is a malformed write.
    #[tokio::test]
    async fn maintenance_commit_rejects_entity_inserts() {
        let catalog = open().await;

        let db_tx = catalog.begin_write_tx().await.unwrap();
        let mut tx = StagedTransaction::begin_detached(db_tx);
        tx.stage(RowOperation::Insert {
            table: TableKind::Table,
            cells: table_row(1, 0, "t", 1, None),
        });
        let err = tx.commit().await.unwrap_err();
        assert!(matches!(err, Error::Constraint(_)), "{err}");
        catalog.close().await.unwrap();
    }

    /// A merge-shaped compaction commit: the merged file lands backdated,
    /// the source rows (current and history alike) are hard-deleted, the
    /// source bytes are scheduled, and `next_row_id` is untouched — all
    /// in one ordinary snapshot-minting commit.
    #[tokio::test]
    async fn merge_shaped_commit_replaces_files_and_schedules_sources() {
        let catalog = open().await;

        // Seed: table 1 with files 9 and 10, both live.
        let db_tx = catalog.begin_write_tx().await.unwrap();
        let mut tx = StagedTransaction::begin_detached(db_tx);
        tx.stage(RowOperation::Insert {
            table: TableKind::Table,
            cells: table_row(1, 0, "t", 1, None),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::Column,
            cells: column_row(1, 1, "a", 0),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::DataFile,
            cells: data_file_row(9, 1, 1),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::DataFile,
            cells: data_file_row(10, 1, 1),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::TableStats,
            cells: vec![Cell::U64(1), Cell::U64(20), Cell::U64(20), Cell::U64(2048)],
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::Snapshot,
            cells: snapshot_row(1, 1, 2),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::SnapshotChanges,
            cells: snapshot_changes_row(1, r#"created_table:"main"."t""#),
        });
        tx.commit().await.unwrap();

        // The merge: insert file 11 backdated to the sources' begin,
        // hard-delete both sources, schedule their bytes.
        let db_tx = catalog.begin_write_tx().await.unwrap();
        let mut tx = StagedTransaction::begin_detached(db_tx);
        tx.stage(RowOperation::Insert {
            table: TableKind::DataFile,
            cells: data_file_row(11, 1, 1),
        });
        tx.stage(RowOperation::Delete {
            table: TableKind::DataFile,
            cells: vec![Cell::U64(1), Cell::U64(9), Cell::Null],
        });
        tx.stage(RowOperation::Delete {
            table: TableKind::DataFile,
            cells: vec![Cell::U64(1), Cell::U64(10), Cell::Null],
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::FilesScheduledForDeletion,
            cells: vec![
                Cell::U64(9),
                Cell::Str("data.parquet".to_string()),
                Cell::Bool(true),
                Cell::I64(1_000),
            ],
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::FilesScheduledForDeletion,
            cells: vec![
                Cell::U64(10),
                Cell::Str("data.parquet".to_string()),
                Cell::Bool(true),
                Cell::I64(1_000),
            ],
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::Snapshot,
            cells: snapshot_row(2, 1, 2),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::SnapshotChanges,
            cells: snapshot_changes_row(2, "merge_adjacent:1"),
        });
        let id = tx.commit().await.unwrap();
        assert_eq!(id.get(), 2, "compaction mints an ordinary snapshot");

        let head = catalog.snapshot().await.unwrap();
        let files = head.data_files_of(crate::catalog::TableId::new(1));
        assert_eq!(files.len(), 1);
        assert_eq!(files[0].id.get(), 11);
        assert_eq!(
            head.scheduled_deletions()
                .iter()
                .map(|d| d.data_file_id)
                .collect::<Vec<_>>(),
            vec![9, 10]
        );
        assert_eq!(
            head.table_stats(crate::catalog::TableId::new(1))
                .unwrap()
                .next_row_id,
            20,
            "compaction never allocates row ids"
        );

        // The sources are gone outright — no history mirror.
        let rows = crate::ffi_support::dump_data_files(&catalog).await.unwrap();
        assert_eq!(rows.len(), 1);
        assert_eq!(rows[0].data_file_id, 11);
        assert_eq!(rows[0].begin_snapshot, 1, "the merged file is backdated");
        catalog.close().await.unwrap();
    }

    /// A rewrite-shaped commit ends the source file and its delete file
    /// into history and rebases the replacement's `begin_snapshot` in
    /// place; nothing is scheduled.
    #[tokio::test]
    async fn rewrite_shaped_commit_ends_rows_and_rebases_the_new_file() {
        let catalog = open().await;
        seed_expired_file(&catalog).await; // file 9 ended at snapshot 2

        // Re-seed a live file 10 with a delete file 11 over it.
        let db_tx = catalog.begin_write_tx().await.unwrap();
        let mut tx = StagedTransaction::begin_detached(db_tx);
        tx.stage(RowOperation::Insert {
            table: TableKind::DataFile,
            cells: data_file_row(10, 1, 3),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::DeleteFile,
            cells: vec![
                Cell::U64(11),
                Cell::U64(1),
                Cell::U64(3),
                Cell::Null,
                Cell::U64(10),
                Cell::Str("delete.parquet".to_string()),
                Cell::Bool(true),
                Cell::Str("parquet".to_string()),
                Cell::U64(2),
                Cell::U64(128),
                Cell::U64(32),
                Cell::Null,
                Cell::Null,
            ],
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::Snapshot,
            cells: snapshot_row(3, 1, 2),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::SnapshotChanges,
            cells: snapshot_changes_row(3, "inserted_into_table:1"),
        });
        tx.commit().await.unwrap();

        // The rewrite: new file 12, end file 10 and delete file 11,
        // rebase 12's begin to this commit's snapshot.
        let db_tx = catalog.begin_write_tx().await.unwrap();
        let mut tx = StagedTransaction::begin_detached(db_tx);
        tx.stage(RowOperation::Insert {
            table: TableKind::DataFile,
            cells: data_file_row(12, 1, 4),
        });
        tx.stage(RowOperation::UpdateSetEnd {
            table: TableKind::DataFile,
            cells: vec![Cell::U64(1), Cell::U64(10), Cell::U64(4)],
        });
        tx.stage(RowOperation::UpdateSetEnd {
            table: TableKind::DeleteFile,
            cells: vec![Cell::U64(1), Cell::U64(11), Cell::U64(4)],
        });
        tx.stage(RowOperation::UpdateSetBegin {
            table: TableKind::DataFile,
            cells: vec![Cell::U64(1), Cell::U64(12), Cell::U64(4)],
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::Snapshot,
            cells: snapshot_row(4, 1, 2),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::SnapshotChanges,
            cells: snapshot_changes_row(4, "rewrite_delete:1"),
        });
        tx.commit().await.unwrap();

        let head = catalog.snapshot().await.unwrap();
        let files = head.data_files_of(crate::catalog::TableId::new(1));
        assert_eq!(files.len(), 1);
        assert_eq!(files[0].id.get(), 12);
        assert!(
            head.delete_files_of(crate::catalog::TableId::new(1))
                .is_empty()
        );
        assert!(head.scheduled_deletions().is_empty());

        // The ended rows survive in history; the replacement is rebased.
        let rows = crate::ffi_support::dump_data_files(&catalog).await.unwrap();
        assert!(
            rows.iter()
                .any(|f| f.data_file_id == 10 && f.end_snapshot == Some(4))
        );
        assert!(
            rows.iter()
                .any(|f| f.data_file_id == 12 && f.begin_snapshot == 4 && f.end_snapshot.is_none())
        );
        catalog.close().await.unwrap();
    }

    /// Rebasing a file that predates the commit is a shape error —
    /// DuckLake only rebases the replacement it just inserted.
    #[tokio::test]
    async fn set_begin_on_a_preexisting_file_is_rejected() {
        let catalog = open().await;
        seed_expired_file(&catalog).await;

        let db_tx = catalog.begin_write_tx().await.unwrap();
        let mut tx = StagedTransaction::begin_detached(db_tx);
        tx.stage(RowOperation::Insert {
            table: TableKind::DataFile,
            cells: data_file_row(10, 1, 3),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::Snapshot,
            cells: snapshot_row(3, 1, 2),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::SnapshotChanges,
            cells: snapshot_changes_row(3, "inserted_into_table:1"),
        });
        tx.commit().await.unwrap();

        let db_tx = catalog.begin_write_tx().await.unwrap();
        let mut tx = StagedTransaction::begin_detached(db_tx);
        tx.stage(RowOperation::UpdateSetBegin {
            table: TableKind::DataFile,
            cells: vec![Cell::U64(1), Cell::U64(10), Cell::U64(4)],
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::Snapshot,
            cells: snapshot_row(4, 1, 2),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::SnapshotChanges,
            cells: snapshot_changes_row(4, "rewrite_delete:1"),
        });
        let err = tx.commit().await.unwrap_err();
        assert!(err.to_string().contains("predates this commit"), "{err}");
        catalog.close().await.unwrap();
    }

    /// Ending a tag entry that does not exist is a shape error — DuckLake
    /// only updates rows it just read, so a miss means drift.
    #[tokio::test]
    async fn ending_an_absent_tag_entry_is_rejected() {
        let catalog = open().await;
        seed_table(&catalog).await;

        let db_tx = catalog.begin_write_tx().await.unwrap();
        let mut tx = StagedTransaction::begin_detached(db_tx);
        tx.stage(RowOperation::UpdateSetEnd {
            table: TableKind::Tag,
            cells: vec![Cell::U64(1), Cell::Str("comment".into()), Cell::U64(2)],
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::Snapshot,
            cells: snapshot_row(2, 1, 2),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::SnapshotChanges,
            cells: snapshot_changes_row(2, "altered_table:1"),
        });
        let err = tx.commit().await.unwrap_err();
        assert!(err.to_string().contains("tag"), "{err}");
        catalog.close().await.unwrap();
    }

    fn macro_row(schema_id: u64, macro_id: u64, name: &str, begin: u64) -> Vec<Cell> {
        vec![
            Cell::U64(schema_id),
            Cell::U64(macro_id),
            Cell::Str(name.to_string()),
            Cell::U64(begin),
            Cell::Null, // end_snapshot
        ]
    }

    fn macro_impl_row(macro_id: u64, impl_id: u64, sql: &str, macro_type: &str) -> Vec<Cell> {
        vec![
            Cell::U64(macro_id),
            Cell::U64(impl_id),
            Cell::Str("duckdb".into()),
            Cell::Str(sql.to_string()),
            Cell::Str(macro_type.to_string()),
        ]
    }

    fn macro_parameter_row(
        macro_id: u64,
        impl_id: u64,
        column_id: u64,
        name: &str,
        default: Option<&str>,
    ) -> Vec<Cell> {
        vec![
            Cell::U64(macro_id),
            Cell::U64(impl_id),
            Cell::U64(column_id),
            Cell::Str(name.to_string()),
            Cell::Str("unknown".into()),
            default.map_or(Cell::Null, |d| Cell::Str(d.to_string())),
            Cell::Str(default.map_or("unknown", |_| "int32").to_string()),
        ]
    }

    /// Stages one commit's rows plus the snapshot pair and returns the
    /// commit error, if any.
    async fn stage_macro_batch(
        catalog: &Catalog,
        snapshot_id: u64,
        next_catalog_id: u64,
        rows: Vec<(TableKind, Vec<Cell>)>,
    ) -> Result<()> {
        let db_tx = catalog.begin_write_tx().await.unwrap();
        let mut tx = StagedTransaction::begin_detached(db_tx);
        for (table, cells) in rows {
            tx.stage(RowOperation::Insert { table, cells });
        }
        tx.stage(RowOperation::Insert {
            table: TableKind::Snapshot,
            cells: snapshot_row(snapshot_id, 1, next_catalog_id),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::SnapshotChanges,
            cells: snapshot_changes_row(snapshot_id, r#"created_scalar_macro:"main"."m""#),
        });
        tx.commit().await.map(|_| ())
    }

    /// A macro insert with its impl and parameter rows folds into one
    /// record (children ordered by their ordinals regardless of emit
    /// order); a later drop ends the whole record into history, children
    /// intact, and time travel still reads it.
    #[tokio::test]
    async fn macro_rows_land_fold_and_drop() {
        let catalog = open().await;

        // Rows deliberately emitted out of ordinal order.
        stage_macro_batch(
            &catalog,
            1,
            11,
            vec![
                (
                    TableKind::MacroParameters,
                    macro_parameter_row(10, 1, 1, "b", Some("5")),
                ),
                (
                    TableKind::MacroImpl,
                    macro_impl_row(10, 1, "(a + b)", "scalar"),
                ),
                (TableKind::Macro, macro_row(0, 10, "add", 1)),
                (
                    TableKind::MacroImpl,
                    macro_impl_row(10, 0, "(a + 1)", "scalar"),
                ),
                (
                    TableKind::MacroParameters,
                    macro_parameter_row(10, 0, 0, "a", None),
                ),
                (
                    TableKind::MacroParameters,
                    macro_parameter_row(10, 1, 0, "a", None),
                ),
            ],
        )
        .await
        .unwrap();

        let head = catalog.snapshot().await.unwrap();
        let stored = &head.macros[&10];
        assert_eq!(stored.begin_snapshot, 1);
        assert_eq!(stored.implementations.len(), 2);
        assert_eq!(stored.implementations[0].impl_id, 0);
        assert_eq!(stored.implementations[0].sql, "(a + 1)");
        assert_eq!(stored.implementations[1].parameters.len(), 2);
        assert_eq!(stored.implementations[1].parameters[1].parameter_name, "b");
        assert_eq!(
            stored.implementations[1].parameters[1]
                .default_value
                .as_deref(),
            Some("5")
        );

        // Drop: the one UPDATE DuckLake issues, nothing touching children.
        let db_tx = catalog.begin_write_tx().await.unwrap();
        let mut tx = StagedTransaction::begin_detached(db_tx);
        tx.stage(RowOperation::UpdateSetEnd {
            table: TableKind::Macro,
            cells: vec![Cell::U64(10), Cell::U64(2)],
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::Snapshot,
            cells: snapshot_row(2, 1, 11),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::SnapshotChanges,
            cells: snapshot_changes_row(2, "dropped_scalar_macro:10"),
        });
        tx.commit().await.unwrap();

        let head = catalog.snapshot().await.unwrap();
        assert!(head.macros.is_empty());
        let past = catalog.snapshot_at(SnapshotId::new(1)).await.unwrap();
        let past_macro = &past.macros[&10];
        assert_eq!(past_macro.end_snapshot, Some(2));
        assert_eq!(past_macro.implementations.len(), 2);
        catalog.close().await.unwrap();
    }

    #[tokio::test]
    async fn macro_insert_without_impl_rows_is_rejected() {
        let catalog = open().await;
        let err = stage_macro_batch(
            &catalog,
            1,
            11,
            vec![(TableKind::Macro, macro_row(0, 10, "m", 1))],
        )
        .await
        .unwrap_err();
        assert!(err.to_string().contains("macro_impl"), "{err}");
        catalog.close().await.unwrap();
    }

    #[tokio::test]
    async fn orphaned_macro_impl_row_is_rejected() {
        let catalog = open().await;
        let err = stage_macro_batch(
            &catalog,
            1,
            11,
            vec![(TableKind::MacroImpl, macro_impl_row(99, 0, "1", "scalar"))],
        )
        .await
        .unwrap_err();
        assert!(err.to_string().contains("macro_impl"), "{err}");
        catalog.close().await.unwrap();
    }

    #[tokio::test]
    async fn orphaned_macro_parameter_row_is_rejected() {
        let catalog = open().await;
        let err = stage_macro_batch(
            &catalog,
            1,
            11,
            vec![
                (TableKind::Macro, macro_row(0, 10, "m", 1)),
                (TableKind::MacroImpl, macro_impl_row(10, 0, "1", "scalar")),
                (
                    TableKind::MacroParameters,
                    macro_parameter_row(10, 7, 0, "a", None),
                ),
            ],
        )
        .await
        .unwrap_err();
        assert!(err.to_string().contains("macro_parameters"), "{err}");
        catalog.close().await.unwrap();
    }

    #[tokio::test]
    async fn macro_impl_id_gap_is_rejected() {
        let catalog = open().await;
        let err = stage_macro_batch(
            &catalog,
            1,
            11,
            vec![
                (TableKind::Macro, macro_row(0, 10, "m", 1)),
                (TableKind::MacroImpl, macro_impl_row(10, 0, "1", "scalar")),
                (TableKind::MacroImpl, macro_impl_row(10, 2, "2", "scalar")),
            ],
        )
        .await
        .unwrap_err();
        assert!(err.to_string().contains("contiguous"), "{err}");
        catalog.close().await.unwrap();
    }

    #[tokio::test]
    async fn macro_parameter_column_id_gap_is_rejected() {
        let catalog = open().await;
        let err = stage_macro_batch(
            &catalog,
            1,
            11,
            vec![
                (TableKind::Macro, macro_row(0, 10, "m", 1)),
                (TableKind::MacroImpl, macro_impl_row(10, 0, "1", "scalar")),
                (
                    TableKind::MacroParameters,
                    macro_parameter_row(10, 0, 1, "a", None),
                ),
            ],
        )
        .await
        .unwrap_err();
        assert!(err.to_string().contains("contiguous"), "{err}");
        catalog.close().await.unwrap();
    }

    fn column_mapping_row(mapping_id: u64, table_id: u64) -> Vec<Cell> {
        vec![
            Cell::U64(mapping_id),
            Cell::U64(table_id),
            Cell::Str("map_by_name".into()),
        ]
    }

    fn name_mapping_row(
        mapping_id: u64,
        column_id: u64,
        source_name: &str,
        target_field_id: u64,
        parent_column: Option<u64>,
        is_partition: bool,
    ) -> Vec<Cell> {
        vec![
            Cell::U64(mapping_id),
            Cell::U64(column_id),
            Cell::Str(source_name.to_string()),
            Cell::U64(target_field_id),
            parent_column.map_or(Cell::Null, Cell::U64),
            Cell::Bool(is_partition),
        ]
    }

    /// Stages one commit's rows plus the snapshot pair and returns the
    /// commit result.
    async fn stage_mapping_batch(
        catalog: &Catalog,
        snapshot_id: u64,
        rows: Vec<(TableKind, Vec<Cell>)>,
    ) -> Result<()> {
        let db_tx = catalog.begin_write_tx().await.unwrap();
        let mut tx = StagedTransaction::begin_detached(db_tx);
        for (table, cells) in rows {
            tx.stage(RowOperation::Insert { table, cells });
        }
        tx.stage(RowOperation::Insert {
            table: TableKind::Snapshot,
            cells: snapshot_row(snapshot_id, 1, 11),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::SnapshotChanges,
            cells: snapshot_changes_row(snapshot_id, "inserted_into_table:1"),
        });
        tx.commit().await.map(|_| ())
    }

    /// A column mapping folds its name-mapping rows (emitted out of
    /// ordinal order, one nested child, one hive-partition virtual
    /// column) and the added file carries its `mapping_id`; the record
    /// is served at any time-travel target.
    #[tokio::test]
    async fn mapping_rows_land_fold_and_serve_time_travel() {
        let catalog = open().await;

        let mut file = data_file_row(7, 1, 1);
        file[14] = Cell::U64(21); // mapping_id
        stage_mapping_batch(
            &catalog,
            1,
            vec![
                (
                    TableKind::NameMapping,
                    name_mapping_row(21, 2, "region", 3, None, true),
                ),
                (TableKind::ColumnMapping, column_mapping_row(21, 1)),
                (
                    TableKind::NameMapping,
                    name_mapping_row(21, 1, "id", 2, Some(0), false),
                ),
                (
                    TableKind::NameMapping,
                    name_mapping_row(21, 0, "payload", 1, None, false),
                ),
                (TableKind::DataFile, file),
            ],
        )
        .await
        .unwrap();

        let head = catalog.snapshot().await.unwrap();
        let stored = &head.mappings[&1][&21];
        assert_eq!(stored.map_type, "map_by_name");
        assert_eq!(stored.name_mappings.len(), 3);
        assert_eq!(stored.name_mappings[0].source_name, "payload");
        assert_eq!(stored.name_mappings[1].parent_column, Some(0));
        assert!(stored.name_mappings[2].is_partition);
        assert_eq!(head.data_files[&1][&7].mapping_id, Some(21));

        // Unversioned: a time-travel view still serves the mapping.
        let past = catalog.snapshot_at(SnapshotId::new(1)).await.unwrap();
        assert_eq!(past.mappings[&1][&21].name_mappings.len(), 3);
        catalog.close().await.unwrap();
    }

    #[tokio::test]
    async fn column_mapping_without_name_rows_is_rejected() {
        let catalog = open().await;
        let err = stage_mapping_batch(
            &catalog,
            1,
            vec![(TableKind::ColumnMapping, column_mapping_row(21, 1))],
        )
        .await
        .unwrap_err();
        assert!(err.to_string().contains("name_mapping"), "{err}");
        catalog.close().await.unwrap();
    }

    #[tokio::test]
    async fn orphaned_name_mapping_row_is_rejected() {
        let catalog = open().await;
        let err = stage_mapping_batch(
            &catalog,
            1,
            vec![(
                TableKind::NameMapping,
                name_mapping_row(99, 0, "id", 1, None, false),
            )],
        )
        .await
        .unwrap_err();
        assert!(err.to_string().contains("name_mapping"), "{err}");
        catalog.close().await.unwrap();
    }

    #[tokio::test]
    async fn duplicate_name_mapping_ordinals_are_rejected() {
        let catalog = open().await;
        let err = stage_mapping_batch(
            &catalog,
            1,
            vec![
                (TableKind::ColumnMapping, column_mapping_row(21, 1)),
                (
                    TableKind::NameMapping,
                    name_mapping_row(21, 0, "a", 1, None, false),
                ),
                (
                    TableKind::NameMapping,
                    name_mapping_row(21, 0, "b", 2, None, false),
                ),
            ],
        )
        .await
        .unwrap_err();
        assert!(err.to_string().contains("unique"), "{err}");
        catalog.close().await.unwrap();
    }

    #[tokio::test]
    async fn name_mapping_parent_after_child_is_rejected() {
        let catalog = open().await;
        let err = stage_mapping_batch(
            &catalog,
            1,
            vec![
                (TableKind::ColumnMapping, column_mapping_row(21, 1)),
                (
                    TableKind::NameMapping,
                    name_mapping_row(21, 0, "a", 1, Some(1), false),
                ),
                (
                    TableKind::NameMapping,
                    name_mapping_row(21, 1, "b", 2, None, false),
                ),
            ],
        )
        .await
        .unwrap_err();
        assert!(err.to_string().contains("earlier"), "{err}");
        catalog.close().await.unwrap();
    }

    #[tokio::test]
    async fn duplicate_mapping_id_against_base_is_rejected() {
        let catalog = open().await;
        let mapping = || {
            vec![
                (TableKind::ColumnMapping, column_mapping_row(21, 1)),
                (
                    TableKind::NameMapping,
                    name_mapping_row(21, 0, "id", 1, None, false),
                ),
            ]
        };
        stage_mapping_batch(&catalog, 1, mapping()).await.unwrap();
        let err = stage_mapping_batch(&catalog, 2, mapping())
            .await
            .unwrap_err();
        assert!(err.to_string().contains("already exists"), "{err}");
        catalog.close().await.unwrap();
    }

    #[tokio::test]
    async fn update_set_end_on_column_mapping_is_rejected() {
        let catalog = open().await;
        let db_tx = catalog.begin_write_tx().await.unwrap();
        let mut tx = StagedTransaction::begin_detached(db_tx);
        tx.stage(RowOperation::UpdateSetEnd {
            table: TableKind::ColumnMapping,
            cells: vec![Cell::U64(21), Cell::U64(1)],
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::Snapshot,
            cells: snapshot_row(1, 1, 11),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::SnapshotChanges,
            cells: snapshot_changes_row(1, "none"),
        });
        let err = tx.commit().await.unwrap_err();
        assert!(err.to_string().contains("not defined"), "{err}");
        catalog.close().await.unwrap();
    }

    #[tokio::test]
    async fn mixed_macro_impl_types_are_rejected() {
        let catalog = open().await;
        let err = stage_macro_batch(
            &catalog,
            1,
            11,
            vec![
                (TableKind::Macro, macro_row(0, 10, "m", 1)),
                (TableKind::MacroImpl, macro_impl_row(10, 0, "1", "scalar")),
                (
                    TableKind::MacroImpl,
                    macro_impl_row(10, 1, "SELECT 1", "table"),
                ),
            ],
        )
        .await
        .unwrap_err();
        assert!(err.to_string().contains("share a type"), "{err}");
        catalog.close().await.unwrap();
    }
}