datafusion-ducklake 0.8.0

//! PostgreSQL implementation of [`MetadataWriter`], multicatalog-aware from day one.
//!
//! Each `PostgresMetadataWriter` instance is bound to a single `catalog_id`. All
//! snapshot and schema inserts are paired with mapping-table inserts in the same
//! transaction, so cross-catalog isolation is enforced at write time.
//!
//! Schema-version allocation is per-catalog dense: a writer computes the next
//! `schema_version` under `FOR UPDATE` on the catalog's mapping rows, bumps on DDL
//! (table create or column-set change), and carries forward on DML (Append/Replace
//! with unchanged columns).

use crate::Result;
use crate::error::{TypeChangeOperation, TypeChangeWriteMode};
use crate::metadata_provider::block_on;
use crate::metadata_writer::is_inlined_system_column;
use crate::metadata_writer::{
    ColumnDef, ColumnStat, CommitIds, DataFileInfo, DeleteFileEntry, DeleteFileInfo,
    ExistingCatalogColumn, INLINED_INDEX_COLUMNS_SETTING, InlinedRowRef, MetadataWriter,
    MultiTableCommit, SnapshotCommitMetadata, StagedTableData, StagedTableWrite, WriteMode,
    WriteSetupResult, assign_column_ids, catalog_column_defs, catalog_column_type_equal,
    catalog_column_type_requires_migration, catalog_columns_differ, encode_inlined_index_columns,
    inlined_delete_conflicts, inlined_delete_groups, live_inlined_index_columns,
    parse_inlined_index_columns, snapshot_has_change, staged_table_write_changes,
    table_storage_changes, table_write_changes, top_level_column_ids, validate_delete_entries,
    validate_inlined_index_columns, validate_name, validate_table_setting,
};
use crate::partition::PartitionTransform;
use arrow::array::{
    Array, BinaryArray, BinaryViewArray, FixedSizeBinaryArray, IntervalMonthDayNanoArray,
    LargeBinaryArray, LargeStringArray, StringArray,
};
use arrow::datatypes::DataType;
use arrow::record_batch::RecordBatch;
use sqlx::AssertSqlSafe;
use sqlx::QueryBuilder;
use sqlx::Row;
use sqlx::postgres::{PgPool, PgPoolOptions, Postgres};

pub(crate) const SQL_CREATE_INLINED_DATA_TABLES: &str =
    "CREATE TABLE IF NOT EXISTS ducklake_inlined_data_tables (
    table_id BIGINT,
    table_name VARCHAR,
    schema_version BIGINT
)";

const DEFAULT_MAX_CONNECTIONS: u32 = 5;

pub const DEFAULT_LOCK_TIMEOUT_MS: u32 = 30_000;

pub(crate) fn quote_ident(name: &str) -> String {
    format!("\"{}\"", name.replace('"', "\"\""))
}

fn inlined_postgres_type(data_type: &DataType, ducklake_type: &str) -> String {
    match data_type {
        DataType::Boolean => "BOOLEAN".to_string(),
        DataType::Int8 | DataType::Int16 => "SMALLINT".to_string(),
        DataType::Int32 => "INTEGER".to_string(),
        DataType::Int64 => "BIGINT".to_string(),
        DataType::UInt8 | DataType::UInt16 => "INTEGER".to_string(),
        DataType::UInt32 => "BIGINT".to_string(),
        DataType::UInt64 => "VARCHAR".to_string(),
        DataType::Float32 => "REAL".to_string(),
        DataType::Float64 => "DOUBLE PRECISION".to_string(),
        DataType::Decimal32(precision, scale)
        | DataType::Decimal64(precision, scale)
        | DataType::Decimal128(precision, scale)
        | DataType::Decimal256(precision, scale) => format!("DECIMAL({precision},{scale})"),
        DataType::Date32 => "VARCHAR".to_string(),
        DataType::Time32(_) | DataType::Time64(_) => "TIME".to_string(),
        DataType::Timestamp(_, _) => "VARCHAR".to_string(),
        DataType::Interval(_) => "INTERVAL".to_string(),
        DataType::FixedSizeBinary(16) if ducklake_type.trim().eq_ignore_ascii_case("uuid") => {
            "UUID".to_string()
        },
        DataType::Utf8
        | DataType::LargeUtf8
        | DataType::Utf8View
        | DataType::Binary
        | DataType::LargeBinary
        | DataType::BinaryView
        | DataType::FixedSizeBinary(_) => "BYTEA".to_string(),
        _ => "VARCHAR".to_string(),
    }
}

// Nested values use reference DuckDB literal text so the shared parser and external readers
// agree.
fn postgres_type_inlines(data_type: &DataType) -> bool {
    crate::nested_inline::type_supports_inlining(data_type)
}

fn push_inlined_postgres_value(
    query: &mut QueryBuilder<Postgres>,
    array: &dyn Array,
    row: usize,
    sql_type: &str,
) -> Result<()> {
    if array.data_type() == &DataType::Interval(arrow::datatypes::IntervalUnit::MonthDayNano)
        && !array.is_null(row)
    {
        let value = array
            .as_any()
            .downcast_ref::<IntervalMonthDayNanoArray>()
            .expect("Arrow data type and array implementation agree")
            .value(row);
        if value.nanoseconds % 1_000 != 0 {
            return Err(crate::DuckLakeError::Unsupported(format!(
                "PostgreSQL INTERVAL cannot inline sub-microsecond nanoseconds: {}",
                value.nanoseconds
            )));
        }
    }
    if sql_type == "BYTEA" {
        if array.is_null(row) {
            query.push_bind(Option::<Vec<u8>>::None);
            return Ok(());
        }
        let value = match array.data_type() {
            DataType::Utf8 => array
                .as_any()
                .downcast_ref::<StringArray>()
                .expect("Arrow data type and array implementation agree")
                .value(row)
                .as_bytes()
                .to_vec(),
            DataType::LargeUtf8 => array
                .as_any()
                .downcast_ref::<LargeStringArray>()
                .expect("Arrow data type and array implementation agree")
                .value(row)
                .as_bytes()
                .to_vec(),
            DataType::Binary => array
                .as_any()
                .downcast_ref::<BinaryArray>()
                .expect("Arrow data type and array implementation agree")
                .value(row)
                .to_vec(),
            DataType::LargeBinary => array
                .as_any()
                .downcast_ref::<LargeBinaryArray>()
                .expect("Arrow data type and array implementation agree")
                .value(row)
                .to_vec(),
            DataType::BinaryView => array
                .as_any()
                .downcast_ref::<BinaryViewArray>()
                .expect("Arrow data type and array implementation agree")
                .value(row)
                .to_vec(),
            DataType::FixedSizeBinary(_) => array
                .as_any()
                .downcast_ref::<FixedSizeBinaryArray>()
                .expect("Arrow data type and array implementation agree")
                .value(row)
                .to_vec(),
            _ => crate::metadata_writer::inlined_text_value(array, row)?.into_bytes(),
        };
        query.push_bind(value);
        return Ok(());
    }

    query.push("CAST(");
    if array.is_null(row) {
        query.push_bind(Option::<String>::None);
    } else {
        query.push_bind(crate::metadata_writer::inlined_text_value(array, row)?);
    }
    query.push(" AS ").push(sql_type).push(')');
    Ok(())
}

async fn postgres_inlined_index_columns(
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
    table_id: i64,
) -> Result<Vec<String>> {
    let value: Option<String> = sqlx::query_scalar(
        "SELECT value FROM ducklake_metadata
         WHERE key = $1 AND scope = 'table' AND scope_id = $2",
    )
    .bind(INLINED_INDEX_COLUMNS_SETTING)
    .bind(table_id)
    .fetch_optional(&mut **tx)
    .await?;
    let declared = parse_inlined_index_columns(value.as_deref().unwrap_or_default())?;
    let available = sqlx::query_scalar::<_, String>(
        "SELECT column_name FROM ducklake_column
         WHERE table_id = $1 AND end_snapshot IS NULL AND parent_column IS NULL",
    )
    .bind(table_id)
    .fetch_all(&mut **tx)
    .await?
    .into_iter()
    .collect::<std::collections::HashSet<_>>();
    Ok(live_inlined_index_columns(declared, &available))
}

async fn ensure_postgres_physical_indexes(
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
    table_id: i64,
    physical_name: &str,
) -> Result<()> {
    let declared = postgres_inlined_index_columns(tx, table_id).await?;
    let physical_rows = sqlx::query(
        "SELECT column_name, data_type FROM information_schema.columns
         WHERE table_schema = current_schema() AND table_name = $1",
    )
    .bind(physical_name)
    .fetch_all(&mut **tx)
    .await?;
    let present = physical_rows
        .iter()
        .filter_map(|row| row.try_get::<String, _>(0).ok())
        .collect::<std::collections::HashSet<_>>();
    for column in declared.iter().map(String::as_str) {
        if !present.contains(column) {
            continue;
        }
        let index_name = format!("{physical_name}_{column}_idx");
        sqlx::query(AssertSqlSafe(format!(
            "CREATE INDEX IF NOT EXISTS {} ON {}({})",
            quote_ident(&index_name),
            quote_ident(physical_name),
            quote_ident(column),
        )))
        .execute(&mut **tx)
        .await?;
    }
    Ok(())
}

/// Each standard DuckLake table as a separate CREATE TABLE IF NOT EXISTS.
/// sqlx executes each `query()` as a single statement, so we split.
pub(crate) const SQL_CREATE_STANDARD_TABLES: &[&str] = &[
    r#"CREATE TABLE IF NOT EXISTS ducklake_metadata (
        key VARCHAR NOT NULL,
        value VARCHAR NOT NULL,
        scope VARCHAR,
        scope_id BIGINT
    )"#,
    r#"CREATE TABLE IF NOT EXISTS ducklake_snapshot (
        snapshot_id BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
        snapshot_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
        schema_version BIGINT NOT NULL DEFAULT 0
    )"#,
    r#"CREATE TABLE IF NOT EXISTS ducklake_schema (
        schema_id BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
        schema_name VARCHAR NOT NULL,
        path VARCHAR NOT NULL DEFAULT '',
        path_is_relative BOOLEAN NOT NULL DEFAULT TRUE,
        begin_snapshot BIGINT NOT NULL,
        end_snapshot BIGINT
    )"#,
    r#"CREATE TABLE IF NOT EXISTS ducklake_table (
        table_id BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
        schema_id BIGINT NOT NULL,
        table_name VARCHAR NOT NULL,
        path VARCHAR NOT NULL DEFAULT '',
        path_is_relative BOOLEAN NOT NULL DEFAULT TRUE,
        begin_snapshot BIGINT NOT NULL,
        end_snapshot BIGINT
    )"#,
    // Multicatalog layout (NOT the upstream/DuckDB-readable single-catalog format,
    // so it is free to carry DB-level guarantees — design §4.1). `column_id` keeps
    // its IDENTITY (a global sequence the allocator reserves from), but is NO LONGER
    // a single-row PRIMARY KEY: a versioned / type-promoted column needs a second
    // row sharing the same `column_id`. Identity is the composite
    // (table_id, column_id, begin_snapshot); a partial unique index (below) enforces
    // at most one *live* version per field-id.
    r#"CREATE TABLE IF NOT EXISTS ducklake_view (
        view_id BIGINT,
        view_uuid UUID,
        begin_snapshot BIGINT,
        end_snapshot BIGINT,
        schema_id BIGINT,
        view_name VARCHAR,
        dialect VARCHAR,
        sql VARCHAR,
        column_aliases VARCHAR
    )"#,
    r#"CREATE TABLE IF NOT EXISTS ducklake_column (
        column_id BIGINT GENERATED ALWAYS AS IDENTITY,
        table_id BIGINT NOT NULL,
        column_name VARCHAR NOT NULL,
        column_type VARCHAR NOT NULL,
        column_order BIGINT NOT NULL,
        nulls_allowed BOOLEAN DEFAULT TRUE,
        parent_column BIGINT,
        initial_default VARCHAR,
        default_value VARCHAR,
        default_value_type VARCHAR,
        default_value_dialect VARCHAR,
        begin_snapshot BIGINT NOT NULL,
        end_snapshot BIGINT,
        PRIMARY KEY (table_id, column_id, begin_snapshot)
    )"#,
    // `owner_catalog_id` is this layout's file-ownership marker. NULL — the catalog
    // holding the row owns the object and reclaims it; set — the row references a
    // file the named catalog owns, so this catalog never schedules or deletes it.
    // See DataFileInfo::owner_catalog_id and migrate_file_ownership_column.
    r#"CREATE TABLE IF NOT EXISTS ducklake_data_file (
        data_file_id BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
        table_id BIGINT NOT NULL,
        path VARCHAR NOT NULL,
        path_is_relative BOOLEAN NOT NULL DEFAULT TRUE,
        file_size_bytes BIGINT NOT NULL,
        footer_size BIGINT,
        encryption_key VARCHAR,
        record_count BIGINT,
        row_id_start BIGINT,
        mapping_id BIGINT,
        begin_snapshot BIGINT NOT NULL,
        end_snapshot BIGINT,
        partial_max BIGINT,
        partition_id BIGINT,
        owner_catalog_id BIGINT
    )"#,
    // Per-table running counters maintained inside the writer's transaction
    // so concurrent writes hand out non-overlapping rowid ranges. `next_row_id`
    // increases monotonically over the table's lifetime (rowids are never
    // reused, even after end-snapshot); `record_count` and `file_size_bytes`
    // mirror the currently-visible totals so DuckDB's `ducklake_table_info`
    // aggregate sees correct numbers for tables this writer produced. Mirrors
    // the sqlite writer's `ducklake_table_stats`.
    r#"CREATE TABLE IF NOT EXISTS ducklake_table_stats (
        table_id BIGINT PRIMARY KEY,
        record_count BIGINT NOT NULL DEFAULT 0,
        next_row_id BIGINT NOT NULL DEFAULT 0,
        file_size_bytes BIGINT NOT NULL DEFAULT 0
    )"#,
    SQL_CREATE_INLINED_DATA_TABLES,
    // Per-file, per-column zone maps (DuckLake spec) — powers file pruning.
    // Column set mirrors the official extension and the SQLite writer.
    r#"CREATE TABLE IF NOT EXISTS ducklake_file_column_stats (
        data_file_id BIGINT NOT NULL,
        table_id BIGINT NOT NULL,
        column_id BIGINT NOT NULL,
        column_size_bytes BIGINT,
        value_count BIGINT,
        null_count BIGINT,
        min_value VARCHAR,
        max_value VARCHAR,
        contains_nan BOOLEAN,
        extra_stats VARCHAR
    )"#,
    // Postgres needs an index here that the DuckLake spec does not define. Every
    // planning read of this table filters `table_id` for equality and then walks
    // `data_file_id` as a range — the statistics page query and the column-size
    // roll-up in `MulticatalogProvider`, and the per-commit recomputation in this
    // writer — so without one a catalog pays a sequential scan of the whole table
    // on the planning path of every query, growing with the number of data files
    // ever written. The official extension leaves this to the operator on Postgres:
    // duckdb/ducklake#1147 taught its filter pushdown to use such an index but
    // deliberately does not create one. That is a reasonable default when a person
    // owns the catalog and a trap when a library does, so this writer creates it.
    //
    // Column order follows the predicates rather than convention: `table_id` is the
    // only equality, `data_file_id` is the only range (and the join key), and
    // `column_id` costs 8 bytes and lets the planner satisfy the `ORDER BY` without
    // a sort. `column_size_bytes` rides in INCLUDE so the roll-up query, which
    // projects nothing else from this table, can be served index-only.
    // `min_value` / `max_value` stay out on purpose: each is capped at
    // `MAX_STRING_STAT_BYTES`, and an INCLUDE payload can be compressed but never
    // moved out of line, so a wide pair would exceed the btree tuple limit and fail
    // the INSERT — turning a read optimisation into a write outage.
    //
    // The guard runs first because a `CREATE INDEX CONCURRENTLY` that is interrupted
    // leaves an invalid index behind, and `CREATE INDEX IF NOT EXISTS` then skips it
    // forever — paying the maintenance cost on every write with none of the read
    // benefit, and silently. Dropping the invalid one makes an interrupted
    // concurrent build self-healing on the next initialization.
    r#"DO $$
        BEGIN
            IF EXISTS (
                SELECT 1 FROM pg_index i
                JOIN pg_class c ON c.oid = i.indexrelid
                WHERE c.relname = 'idx_file_column_stats_table_file_column'
                  AND NOT i.indisvalid
            ) THEN
                DROP INDEX IF EXISTS idx_file_column_stats_table_file_column;
            END IF;
        END $$"#,
    r#"CREATE INDEX IF NOT EXISTS idx_file_column_stats_table_file_column
        ON ducklake_file_column_stats (table_id, data_file_id, column_id)
        INCLUDE (column_size_bytes)"#,
    // Table-wide per-column roll-up (DuckLake spec) — feeds the optimizer.
    r#"CREATE TABLE IF NOT EXISTS ducklake_table_column_stats (
        table_id BIGINT NOT NULL,
        column_id BIGINT NOT NULL,
        contains_null BOOLEAN,
        contains_nan BOOLEAN,
        min_value VARCHAR,
        max_value VARCHAR,
        extra_stats VARCHAR
    )"#,
    r#"CREATE TABLE IF NOT EXISTS ducklake_delete_file (
        delete_file_id BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
        data_file_id BIGINT NOT NULL,
        table_id BIGINT NOT NULL,
        path VARCHAR NOT NULL,
        path_is_relative BOOLEAN NOT NULL DEFAULT TRUE,
        file_size_bytes BIGINT NOT NULL,
        footer_size BIGINT,
        encryption_key VARCHAR,
        delete_count BIGINT,
        begin_snapshot BIGINT NOT NULL,
        end_snapshot BIGINT,
        -- Max embedded per-row snapshot of a cumulative delete file (current
        -- DuckLake spec). This crate's writer emits per-snapshot delete files
        -- and leaves it NULL; readers use it to window cumulative files by row.
        partial_max BIGINT,
        -- Ownership marker, as on ducklake_data_file.
        owner_catalog_id BIGINT
    )"#,
    // Idempotent guard: an existing single-catalog Postgres catalog populated by
    // another tool may not have schema_version on ducklake_snapshot.
    r#"ALTER TABLE ducklake_snapshot
        ADD COLUMN IF NOT EXISTS schema_version BIGINT NOT NULL DEFAULT 0"#,
    // Idempotent guard: an existing store may predate the v1.0 partial-file
    // marker. NULL means "not a partial file", correct for every pre-compaction
    // file.
    r#"ALTER TABLE ducklake_data_file
        ADD COLUMN IF NOT EXISTS partial_max BIGINT"#,
    // Idempotent guard: an existing store may predate partitioning support.
    // NULL means "not partitioned", correct for every pre-partitioning file.
    r#"ALTER TABLE ducklake_data_file
        ADD COLUMN IF NOT EXISTS partition_id BIGINT"#,
    // Per-snapshot change ledger (DuckLake spec).
    r#"CREATE TABLE IF NOT EXISTS ducklake_snapshot_changes (
        snapshot_id BIGINT PRIMARY KEY,
        changes_made VARCHAR,
        author VARCHAR,
        commit_message VARCHAR,
        commit_extra_info VARCHAR
    )"#,
    r#"ALTER TABLE ducklake_snapshot_changes
        ALTER COLUMN changes_made DROP NOT NULL"#,
    // Partition spec generations (DuckLake spec); end_snapshot NULL == active.
    // partition_id is IDENTITY so set_partition_spec allocates it via RETURNING
    // (like the other ids on the multicatalog path). No catalog_id needed —
    // scoped implicitly via table_id, exactly like ducklake_file_column_stats.
    r#"CREATE TABLE IF NOT EXISTS ducklake_partition_info (
        partition_id BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
        table_id BIGINT NOT NULL,
        begin_snapshot BIGINT NOT NULL,
        end_snapshot BIGINT
    )"#,
    // Partition-key columns for a spec (DuckLake spec), ordered by partition_key_index.
    r#"CREATE TABLE IF NOT EXISTS ducklake_partition_column (
        partition_id BIGINT NOT NULL,
        table_id BIGINT NOT NULL,
        partition_key_index BIGINT NOT NULL,
        column_id BIGINT NOT NULL,
        transform VARCHAR NOT NULL
    )"#,
    // Per-file partition values (DuckLake spec): the value every row in the file
    // shares for a partition key, DuckDB-canonical VARCHAR (NULL is legal).
    r#"CREATE TABLE IF NOT EXISTS ducklake_file_partition_value (
        data_file_id BIGINT NOT NULL,
        table_id BIGINT NOT NULL,
        partition_key_index BIGINT NOT NULL,
        partition_value VARCHAR
    )"#,
    // Sort spec generations (DuckLake spec); end_snapshot NULL == active. sort_id
    // is IDENTITY so set_sort_spec allocates it via RETURNING (like partition_id).
    r#"CREATE TABLE IF NOT EXISTS ducklake_sort_info (
        sort_id BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
        table_id BIGINT NOT NULL,
        begin_snapshot BIGINT NOT NULL,
        end_snapshot BIGINT
    )"#,
    // Sort-key expressions for a spec (DuckLake spec), ordered by sort_key_index.
    // expression is a sort expression in `dialect` (this crate produces bare column
    // names under `duckdb`); sort_direction ASC/DESC; null_order NULLS_FIRST/LAST.
    r#"CREATE TABLE IF NOT EXISTS ducklake_sort_expression (
        sort_id BIGINT NOT NULL,
        table_id BIGINT NOT NULL,
        sort_key_index BIGINT NOT NULL,
        expression VARCHAR NOT NULL,
        dialect VARCHAR NOT NULL,
        sort_direction VARCHAR NOT NULL,
        null_order VARCHAR NOT NULL
    )"#,
];

/// Multicatalog scaffolding tables. Always run after the standard tables.
pub(crate) const SQL_CREATE_MULTICATALOG_TABLES: &[&str] = &[
    r#"CREATE TABLE IF NOT EXISTS ducklake_catalog (
        catalog_id BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
        catalog_name VARCHAR NOT NULL UNIQUE,
        data_path VARCHAR,
        created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
    )"#,
    r#"DO $$
        BEGIN
            IF NOT EXISTS (
                SELECT 1 FROM pg_attribute
                WHERE attrelid = 'ducklake_catalog'::regclass
                  AND attname = 'data_path'
                  AND NOT attisdropped
            ) THEN
                ALTER TABLE ducklake_catalog ADD COLUMN data_path VARCHAR;
                UPDATE ducklake_catalog AS catalog
                SET data_path = (
                    SELECT value FROM ducklake_metadata
                    WHERE key = 'data_path' AND scope IS NULL
                    LIMIT 1
                )
                WHERE catalog.data_path IS NULL;
            END IF;
        END $$"#,
    r#"CREATE TABLE IF NOT EXISTS ducklake_catalog_snapshot_map (
        catalog_id BIGINT NOT NULL,
        snapshot_id BIGINT NOT NULL,
        PRIMARY KEY (catalog_id, snapshot_id)
    )"#,
    r#"CREATE TABLE IF NOT EXISTS ducklake_catalog_schema_map (
        catalog_id BIGINT NOT NULL,
        schema_id BIGINT NOT NULL,
        PRIMARY KEY (catalog_id, schema_id)
    )"#,
    r#"CREATE TABLE IF NOT EXISTS ducklake_schema_versions (
        begin_snapshot BIGINT NOT NULL,
        schema_version BIGINT NOT NULL,
        table_id BIGINT NOT NULL,
        UNIQUE (table_id, begin_snapshot)
    )"#,
    // Files queued for physical deletion by the two-phase vacuum (DuckLake
    // spec). `expire_snapshots_in_catalog` GCs unreachable catalog rows and
    // records the orphaned physical paths here; `cleanup_old_files` deletes
    // the objects and removes these rows. `path` is stored relative to the
    // catalog `data_path` root (already resolved through schema/table) so
    // cleanup needs only a single-level join with `data_path`.
    //
    // Deviation from the single-catalog upstream schema: `catalog_id` scopes
    // each scheduled file to its catalog. Without it cleanup couldn't tell
    // catalogs apart — the data-file rows it would otherwise join against are
    // already deleted by the time the file is scheduled.
    //
    // `scheduled_by_owner` records that the scheduling catalog owned the object, so
    // cleanup can tell a row written under the ownership rule from one written before
    // it. TRUE on every row inserted here — a catalog only ever schedules what it owns
    // — and NULL exactly on rows predating the rule, which may name another catalog's
    // live file and are therefore held to the stricter layout check in
    // `retain_unreferenced`. It is a boolean, not an `owner_catalog_id` like the file
    // tables carry: there a set value names ANOTHER catalog, and reusing the name for
    // the opposite polarity invites a query that gets it backwards.
    r#"CREATE TABLE IF NOT EXISTS ducklake_files_scheduled_for_deletion (
        catalog_id BIGINT NOT NULL,
        data_file_id BIGINT NOT NULL,
        path VARCHAR NOT NULL,
        path_is_relative BOOLEAN NOT NULL DEFAULT TRUE,
        schedule_start TIMESTAMPTZ DEFAULT NOW(),
        scheduled_by_owner BOOLEAN
    )"#,
    r#"CREATE INDEX IF NOT EXISTS idx_scheduled_for_deletion_catalog
        ON ducklake_files_scheduled_for_deletion(catalog_id)"#,
    r#"CREATE TABLE IF NOT EXISTS ducklake_dropped_data_path (
        data_path VARCHAR PRIMARY KEY,
        dropped_at TIMESTAMPTZ DEFAULT NOW()
    )"#,
    r#"CREATE INDEX IF NOT EXISTS idx_catalog_snapshot_map_snapshot
        ON ducklake_catalog_snapshot_map(snapshot_id)"#,
    r#"CREATE INDEX IF NOT EXISTS idx_catalog_schema_map_schema
        ON ducklake_catalog_schema_map(schema_id)"#,
    r#"CREATE INDEX IF NOT EXISTS idx_schema_versions_table
        ON ducklake_schema_versions(table_id, begin_snapshot)"#,
    // Belt-and-suspenders: app-level lock_catalog should already prevent
    // duplicates, but a partial unique index catches anyone bypassing the
    // writer (manual SQL, external migrations).
    r#"CREATE UNIQUE INDEX IF NOT EXISTS idx_active_table_per_schema
        ON ducklake_table(schema_id, table_name) WHERE end_snapshot IS NULL"#,
    // At most one *live* version per field-id (design §4.1, reviews #2/#3). The
    // promote's retire-then-insert (end the old row, then insert the new live row,
    // in one txn) keeps this satisfied at every commit boundary.
    r#"CREATE UNIQUE INDEX IF NOT EXISTS idx_ducklake_column_live
        ON ducklake_column(table_id, column_id) WHERE end_snapshot IS NULL"#,
];

/// Run a slice of DDL statements against the pool. Each statement executes independently.
pub(crate) async fn execute_ddl_statements(
    pool: &PgPool,
    statements: &[&'static str],
) -> Result<()> {
    for stmt in statements {
        sqlx::query(*stmt).execute(pool).await?;
    }
    Ok(())
}

/// Upgrade an existing multicatalog store's `ducklake_column` from the legacy
/// single-row `column_id` PRIMARY KEY to the composite
/// `(table_id, column_id, begin_snapshot)` PK, so a versioned / type-promoted
/// column can have a second row sharing its `column_id`. `CREATE TABLE IF NOT
/// EXISTS` only shapes fresh stores; Postgres can `ALTER` a PK in place (unlike
/// SQLite). Idempotent (only acts when the current PK is the single-column one;
/// a no-op once composite) and lossless. The `IDENTITY` on `column_id` (the
/// allocator's sequence) is independent of the PK and survives the swap. The
/// partial unique index is created idempotently by `SQL_CREATE_STANDARD_TABLES`.
pub(crate) async fn migrate_ducklake_column_to_composite_pk(pool: &PgPool) -> Result<()> {
    sqlx::query(
        r#"DO $$
        DECLARE pk_name text;
        BEGIN
            -- Find the PRIMARY KEY iff it is a single-column PK (the legacy shape).
            SELECT conname INTO pk_name
            FROM pg_constraint
            WHERE conrelid = 'ducklake_column'::regclass
              AND contype = 'p'
              AND array_length(conkey, 1) = 1;
            IF pk_name IS NOT NULL THEN
                EXECUTE 'ALTER TABLE ducklake_column DROP CONSTRAINT ' || quote_ident(pk_name);
                EXECUTE 'ALTER TABLE ducklake_column ADD PRIMARY KEY (table_id, column_id, begin_snapshot)';
            END IF;
        END $$;"#,
    )
    .execute(pool)
    .await?;
    Ok(())
}

/// Add default metadata to a catalog created before those DuckLake columns were
/// supported. The fields are nullable, so existing column versions retain the
/// specified absence of a default.
pub(crate) async fn migrate_column_default_metadata(pool: &PgPool) -> Result<()> {
    sqlx::query(
        "ALTER TABLE ducklake_column
         ADD COLUMN IF NOT EXISTS initial_default VARCHAR,
         ADD COLUMN IF NOT EXISTS default_value VARCHAR,
         ADD COLUMN IF NOT EXISTS default_value_type VARCHAR,
         ADD COLUMN IF NOT EXISTS default_value_dialect VARCHAR",
    )
    .execute(pool)
    .await?;
    Ok(())
}

/// Record file ownership on its own column, and re-point the reference indexes at it.
///
/// Before this, ownership was read off `path_is_relative`: an absolute-path row was
/// taken to be a reference to a file another catalog owns. But that flag already
/// meant "this path is spelled absolutely", which
/// [`crate::DuckLakeTableWriter::begin_write_to_path`] produces for a catalog's OWN
/// file. The two facts were indistinguishable, so such a file stopped being
/// scheduled by expire and compaction, and — when it lived outside `data_path`,
/// which is the reason that entry point exists — was never reclaimed at all.
///
/// `owner_catalog_id` separates them: NULL means the catalog holding the row owns
/// the object, set means the row references a file the named catalog owns. Adding
/// the column leaves every pre-existing row NULL, i.e. owned, which is what every
/// absolute row meant before the flag was overloaded — so an upgrade restores their
/// reclaim path rather than reinterpreting them.
///
/// Additive and idempotent, and runs on every boot: `ADD COLUMN IF NOT EXISTS` on
/// tables the `SQL_CREATE_*_TABLES` statements have already created, then the index
/// swap. The two indexes are created HERE rather than alongside the tables because their
/// predicate names the new column, which on an existing store does not exist until
/// the `ALTER`s above have run. They carry new names rather than redefining the old
/// ones: `CREATE INDEX IF NOT EXISTS` keeps whatever predicate an index already
/// has, so a same-named index would silently stay on `NOT path_is_relative`.
pub(crate) async fn migrate_file_ownership_column(pool: &PgPool) -> Result<()> {
    for stmt in [
        "ALTER TABLE ducklake_data_file ADD COLUMN IF NOT EXISTS owner_catalog_id BIGINT",
        "ALTER TABLE ducklake_delete_file ADD COLUMN IF NOT EXISTS owner_catalog_id BIGINT",
        "ALTER TABLE ducklake_files_scheduled_for_deletion
         ADD COLUMN IF NOT EXISTS scheduled_by_owner BOOLEAN",
        // Cleanup asks "does any catalog still reference this object", answered by a
        // scan of the reference rows. These partial indexes keep that set — bounded
        // by how many cross-catalog references exist, not by the size of the file
        // tables — servable as an index-only scan.
        "CREATE INDEX IF NOT EXISTS idx_data_file_owner_catalog
         ON ducklake_data_file(path) WHERE owner_catalog_id IS NOT NULL",
        "CREATE INDEX IF NOT EXISTS idx_delete_file_owner_catalog
         ON ducklake_delete_file(path) WHERE owner_catalog_id IS NOT NULL",
        // Their predecessors indexed `NOT path_is_relative`, which no longer answers
        // any question this crate asks.
        "DROP INDEX IF EXISTS idx_data_file_absolute_path",
        "DROP INDEX IF EXISTS idx_delete_file_absolute_path",
    ] {
        sqlx::query(AssertSqlSafe(stmt)).execute(pool).await?;
    }
    Ok(())
}

/// Batch size for one `DELETE` inside the orphan purge, and the cap on how many
/// batches a single call will run. Bounded on purpose: this runs during schema
/// initialization, which callers do on startup and often behind a lock, so no
/// single statement may run long and no single call may run away. A backlog
/// larger than the cap drains over the next few initializations.
const ORPHAN_PURGE_BATCH: i64 = 20_000;
const ORPHAN_PURGE_MAX_BATCHES: usize = 10;

/// Delete metadata rows whose owning row is already gone.
///
/// Expire used to drop `ducklake_data_file` and `ducklake_snapshot` rows without
/// the per-file and per-commit rows hanging off them, so catalogs written by an
/// earlier version carry rows that no query can reach — every read joins through
/// the owner — and that nothing will ever collect: expire only cleans what it is
/// retiring in that transaction, and these owners were retired long ago.
///
/// This repairs that history. It is a maintenance operation, not part of expire:
/// expire keeps deleting exactly what official deletes, and this runs beside it on
/// the same sweep. Deliberately not called from schema initialization -- callers
/// bootstrap on startup, often behind a lock where one long statement is a failed
/// boot, and a scan for orphans has no business on that path.
///
/// With expire and `drop_catalog` both reclaiming their children, no path left in
/// this crate produces these orphans, so on a drained catalog every pass finds
/// nothing. It stays in the sweep rather than being deleted after one run because
/// it is the only thing that can reach rows an older version stranded, and because
/// it is the check that would catch a future path that forgets its children again.
///
/// Safety rests on the anti-join, not on a guess about what looks stale. A row
/// whose owner is absent is unreachable by construction, and owner and dependants
/// are always written in one transaction, so no snapshot can observe a dependant
/// whose owner has not landed yet. Batched by `ctid` so each statement touches a
/// bounded number of rows.
pub async fn purge_orphaned_metadata(pool: &PgPool) -> Result<()> {
    // (dependant table, owning table, shared key)
    const TARGETS: &[(&str, &str, &str)] = &[
        (
            "ducklake_file_column_stats",
            "ducklake_data_file",
            "data_file_id",
        ),
        (
            "ducklake_file_partition_value",
            "ducklake_data_file",
            "data_file_id",
        ),
        (
            "ducklake_snapshot_changes",
            "ducklake_snapshot",
            "snapshot_id",
        ),
    ];

    for (child, owner, key) in TARGETS {
        let mut removed: u64 = 0;
        for _ in 0..ORPHAN_PURGE_MAX_BATCHES {
            let affected = sqlx::query(AssertSqlSafe(format!(
                "DELETE FROM {child} WHERE ctid IN (
                     SELECT c.ctid FROM {child} c
                     WHERE NOT EXISTS (
                         SELECT 1 FROM {owner} o WHERE o.{key} = c.{key}
                     )
                     LIMIT $1
                 )"
            )))
            .bind(ORPHAN_PURGE_BATCH)
            .execute(pool)
            .await?
            .rows_affected();

            removed += affected;
            if affected < ORPHAN_PURGE_BATCH as u64 {
                break;
            }
        }
        if removed > 0 {
            tracing::info!(
                table = child,
                removed,
                "purged orphaned DuckLake metadata rows whose {owner} row was already gone"
            );
        }
    }
    Ok(())
}

/// PostgreSQL-based metadata writer for DuckLake catalogs.
///
/// Bound to a single `catalog_id` at construction. To write to a different
/// catalog, construct a new writer with the desired `catalog_id`.
#[derive(Debug, Clone)]
pub struct PostgresMetadataWriter {
    pool: PgPool,
    catalog_id: i64,
    lock_timeout_ms: u32,
}

impl PostgresMetadataWriter {
    /// Bind a writer to the given pool and catalog id.
    ///
    /// Use [`crate::multicatalog::MulticatalogManager::create_catalog`] to obtain
    /// or create a catalog id by name.
    pub async fn with_pool(pool: PgPool, catalog_id: i64) -> Result<Self> {
        sqlx::query(SQL_CREATE_INLINED_DATA_TABLES)
            .execute(&pool)
            .await?;
        Ok(Self {
            pool,
            catalog_id,
            lock_timeout_ms: DEFAULT_LOCK_TIMEOUT_MS,
        })
    }

    pub async fn new(connection_string: &str, catalog_id: i64) -> Result<Self> {
        Self::with_max_connections(connection_string, catalog_id, DEFAULT_MAX_CONNECTIONS).await
    }

    pub async fn with_max_connections(
        connection_string: &str,
        catalog_id: i64,
        max_connections: u32,
    ) -> Result<Self> {
        let pool = PgPoolOptions::new()
            .max_connections(max_connections)
            .connect(connection_string)
            .await?;
        Self::with_pool(pool, catalog_id).await
    }

    /// Sets the Postgres `lock_timeout` (ms) applied before `FOR UPDATE`.
    /// `0` disables the timeout — not recommended for production.
    pub fn with_lock_timeout(mut self, ms: u32) -> Self {
        self.lock_timeout_ms = ms;
        self
    }

    pub fn catalog_id(&self) -> i64 {
        self.catalog_id
    }
}

async fn lock_catalog(
    catalog_id: i64,
    lock_timeout_ms: u32,
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
) -> Result<()> {
    if lock_timeout_ms > 0 {
        sqlx::query(AssertSqlSafe(format!(
            "SET LOCAL lock_timeout = {}",
            lock_timeout_ms
        )))
        .execute(&mut **tx)
        .await?;
    }
    let row =
        sqlx::query("SELECT catalog_id FROM ducklake_catalog WHERE catalog_id = $1 FOR UPDATE")
            .bind(catalog_id)
            .fetch_optional(&mut **tx)
            .await?;
    if row.is_none() {
        return Err(crate::DuckLakeError::CatalogNotFound(format!(
            "catalog_id {}",
            catalog_id
        )));
    }
    Ok(())
}

async fn assert_schema_in_catalog(
    catalog_id: i64,
    schema_id: i64,
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
) -> Result<()> {
    let row = sqlx::query(
        "SELECT 1 FROM ducklake_catalog_schema_map
         WHERE catalog_id = $1 AND schema_id = $2",
    )
    .bind(catalog_id)
    .bind(schema_id)
    .fetch_optional(&mut **tx)
    .await?;
    if row.is_none() {
        return Err(crate::DuckLakeError::InvalidConfig(format!(
            "schema_id {} does not belong to catalog_id {}",
            schema_id, catalog_id
        )));
    }
    Ok(())
}

async fn assert_table_in_catalog(
    catalog_id: i64,
    table_id: i64,
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
) -> Result<()> {
    let row = sqlx::query(
        "SELECT m.catalog_id FROM ducklake_table t
         LEFT JOIN ducklake_catalog_schema_map m ON m.schema_id = t.schema_id
         WHERE t.table_id = $1",
    )
    .bind(table_id)
    .fetch_optional(&mut **tx)
    .await?;
    match row {
        None => Err(crate::DuckLakeError::TableNotFound(format!(
            "table_id {}",
            table_id
        ))),
        Some(r) => {
            let owner: Option<i64> = r.try_get(0)?;
            if owner != Some(catalog_id) {
                Err(crate::DuckLakeError::InvalidConfig(format!(
                    "table_id {} does not belong to catalog_id {}",
                    table_id, catalog_id
                )))
            } else {
                Ok(())
            }
        },
    }
}

/// Create a new snapshot for a sort-spec DDL change and advance this catalog's
/// head, carrying the current `schema_version` FORWARD unchanged. Unlike a
/// partition-spec change, a sort-spec change does not bump `schema_version` or write
/// a `ducklake_schema_versions` ledger row — sort order does not alter the logical
/// schema. Returns the new snapshot id.
async fn insert_sort_snapshot(
    catalog_id: i64,
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
) -> Result<i64> {
    let snapshot_id: i64 = sqlx::query(
        "INSERT INTO ducklake_snapshot (snapshot_time, schema_version)
         VALUES (NOW(), 0) RETURNING snapshot_id",
    )
    .fetch_one(&mut **tx)
    .await?
    .try_get(0)?;
    sqlx::query(
        "INSERT INTO ducklake_catalog_snapshot_map (catalog_id, snapshot_id) VALUES ($1, $2)",
    )
    .bind(catalog_id)
    .bind(snapshot_id)
    .execute(&mut **tx)
    .await?;
    // Carry the live schema_version forward (no bump) so reads at the new head see
    // the same schema as before the sort change.
    let carried: i64 = sqlx::query(
        "SELECT COALESCE(MAX(s.schema_version), 0) FROM ducklake_snapshot s
         JOIN ducklake_catalog_snapshot_map m ON m.snapshot_id = s.snapshot_id
         WHERE m.catalog_id = $1 AND s.snapshot_id <> $2",
    )
    .bind(catalog_id)
    .bind(snapshot_id)
    .fetch_one(&mut **tx)
    .await?
    .try_get(0)?;
    sqlx::query("UPDATE ducklake_snapshot SET schema_version = $1 WHERE snapshot_id = $2")
        .bind(carried)
        .bind(snapshot_id)
        .execute(&mut **tx)
        .await?;
    Ok(snapshot_id)
}

/// Reject only a `table_id` hint that exists and belongs to ANOTHER catalog. A
/// hint that does not yet exist is fine — it is the id reserved at begin that
/// `finalize_snapshot` is about to create under this catalog (first write to a
/// new table). Used by the commit path (`register_data_file`/`publish_snapshot`),
/// which must tolerate a not-yet-created table while still catching a caller that
/// hands in a different catalog's table.
async fn assert_table_not_in_other_catalog(
    catalog_id: i64,
    table_id: i64,
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
) -> Result<()> {
    let row = sqlx::query(
        "SELECT m.catalog_id FROM ducklake_table t
         LEFT JOIN ducklake_catalog_schema_map m ON m.schema_id = t.schema_id
         WHERE t.table_id = $1",
    )
    .bind(table_id)
    .fetch_optional(&mut **tx)
    .await?;
    if let Some(r) = row {
        let owner: Option<i64> = r.try_get(0)?;
        if owner != Some(catalog_id) {
            return Err(crate::DuckLakeError::InvalidConfig(format!(
                "table_id {} does not belong to catalog_id {}",
                table_id, catalog_id
            )));
        }
    }
    Ok(())
}

/// Reserve `n` ids from the IDENTITY-backing sequence of `table.col` WITHOUT
/// inserting rows, so begin can hand out column/schema/table ids (column ids are
/// the parquet field-ids baked into the staged file) that the commit later
/// inserts explicitly via `OVERRIDING SYSTEM VALUE`. Sequences are
/// non-transactional, so gaps from an aborted write are fine and expected. The
/// ids come back in order.
async fn reserve_ids(
    table: &str,
    col: &str,
    n: i64,
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
) -> Result<Vec<i64>> {
    if n <= 0 {
        return Ok(Vec::new());
    }
    let rows =
        sqlx::query("SELECT nextval(pg_get_serial_sequence($1, $2)) FROM generate_series(1, $3)")
            .bind(table)
            .bind(col)
            .bind(n)
            .fetch_all(&mut **tx)
            .await?;
    rows.into_iter().map(|r| Ok(r.try_get(0)?)).collect()
}

/// Optimistic-concurrency check for a `Replace` commit. Run under the catalog
/// `FOR UPDATE` lock at the commit point, BEFORE this writer inserts its own
/// files/columns and before `advance_catalog_head`. Because snapshot ids are
/// assigned at commit (id order == commit order per catalog) and all metadata is
/// written at commit (no dormant rows), this scalar check is exact: if any data
/// file OR column of the table has `begin_snapshot` or `end_snapshot` > `base`
/// (the catalog head observed at begin), another writer committed a generation
/// of this table since this write began ⇒ [`DuckLakeError::Conflict`]. Catches a
/// data Replace (new file begin), a fileless `CREATE`/Replace (new column begin),
/// and a DROP (end-stamp). The writer's own rows are not written yet, so the
/// check never self-conflicts. (`Append` does not call this: concurrent appends
/// commute, matching upstream DuckLake.)
pub(crate) async fn detect_replace_conflict(
    table_id: i64,
    base_snapshot: i64,
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
) -> Result<()> {
    let conflict = sqlx::query(
        "SELECT 1 WHERE EXISTS (SELECT 1 FROM ducklake_data_file
             WHERE table_id = $1 AND (begin_snapshot > $2 OR end_snapshot > $2))
           OR EXISTS (SELECT 1 FROM ducklake_column
             WHERE table_id = $1 AND (begin_snapshot > $2 OR end_snapshot > $2))",
    )
    .bind(table_id)
    .bind(base_snapshot)
    .fetch_optional(&mut **tx)
    .await?;
    if conflict.is_some() {
        return Err(crate::DuckLakeError::Conflict(format!(
            "Replace on table {table_id} conflicts with a concurrent write committed since \
             snapshot {base_snapshot}; aborting (retry the write against the new generation)"
        )));
    }
    let inlined_tables =
        sqlx::query("SELECT table_name FROM ducklake_inlined_data_tables WHERE table_id = $1")
            .bind(table_id)
            .fetch_all(&mut **tx)
            .await?;
    for row in inlined_tables {
        let table_name: String = row.try_get(0)?;
        let sql = format!(
            "SELECT 1 FROM {} WHERE begin_snapshot > $1 OR end_snapshot > $1 LIMIT 1",
            quote_ident(&table_name)
        );
        if sqlx::query(AssertSqlSafe(sql))
            .bind(base_snapshot)
            .fetch_optional(&mut **tx)
            .await?
            .is_some()
        {
            return Err(crate::DuckLakeError::Conflict(format!(
                "Replace on table {table_id} conflicts with inlined data committed since \
                 snapshot {base_snapshot}; aborting"
            )));
        }
    }
    Ok(())
}

// The SHARE lock orders inserts from other PostgreSQL clients with the
// source-file check and surrounding metadata commit.
async fn detect_new_inlined_deletes(
    table_id: i64,
    base_snapshot: i64,
    data_file_ids: &[i64],
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
) -> Result<()> {
    if data_file_ids.is_empty() {
        return Ok(());
    }

    let table = crate::metadata_provider::inlined_delete_table_name(table_id)?;
    let exists: bool = sqlx::query_scalar("SELECT to_regclass($1) IS NOT NULL")
        .bind(&table)
        .fetch_one(&mut **tx)
        .await?;
    if !exists {
        return Ok(());
    }

    sqlx::query(AssertSqlSafe(format!(
        "LOCK TABLE \"{table}\" IN SHARE MODE"
    )))
    .execute(&mut **tx)
    .await?;
    let file_ids = data_file_ids
        .iter()
        .map(i64::to_string)
        .collect::<Vec<_>>()
        .join(", ");
    let conflict = sqlx::query(AssertSqlSafe(format!(
        "SELECT 1 FROM \"{table}\" newer
         WHERE newer.file_id IN ({file_ids})
           AND (newer.begin_snapshot IS NULL OR (
             newer.begin_snapshot > $1
             AND (newer.row_id IS NULL OR NOT EXISTS (
               SELECT 1 FROM \"{table}\" prior
               WHERE prior.file_id = newer.file_id
                 AND prior.row_id = newer.row_id
                 AND prior.begin_snapshot <= $2
             ))
           ))
         LIMIT 1"
    )))
    .bind(base_snapshot)
    .bind(base_snapshot)
    .fetch_optional(&mut **tx)
    .await?;
    if conflict.is_some() {
        return Err(crate::DuckLakeError::Conflict(format!(
            "table {table_id} gained an inlined delete on a source file since snapshot \
             {base_snapshot}; re-open the catalog and re-plan"
        )));
    }
    Ok(())
}

/// Retire the generation preceding `snapshot_id` for a Replace: end-snapshot
/// every still-live file from an earlier snapshot and zero the visible
/// record/byte totals. The `begin_snapshot < snapshot_id` guard leaves the
/// current write's own files untouched (multi-file safety); `next_row_id` stays
/// monotonic so rowids are never reused.
async fn retire_prior_generation(
    table_id: i64,
    snapshot_id: i64,
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
) -> Result<()> {
    sqlx::query(
        "UPDATE ducklake_data_file SET end_snapshot = $1
         WHERE table_id = $2 AND end_snapshot IS NULL AND begin_snapshot < $1",
    )
    .bind(snapshot_id)
    .bind(table_id)
    .execute(&mut **tx)
    .await?;

    let inlined_tables =
        sqlx::query("SELECT table_name FROM ducklake_inlined_data_tables WHERE table_id = $1")
            .bind(table_id)
            .fetch_all(&mut **tx)
            .await?;
    for row in inlined_tables {
        let table_name: String = row.try_get(0)?;
        let sql = format!(
            "UPDATE {} SET end_snapshot = $1 \
             WHERE end_snapshot IS NULL AND begin_snapshot < $1",
            quote_ident(&table_name)
        );
        sqlx::query(AssertSqlSafe(sql))
            .bind(snapshot_id)
            .execute(&mut **tx)
            .await?;
    }

    sqlx::query(
        "UPDATE ducklake_table_stats
         SET record_count = 0, file_size_bytes = 0
         WHERE table_id = $1",
    )
    .bind(table_id)
    .execute(&mut **tx)
    .await?;
    Ok(())
}

/// Publish `snapshot_id` as the catalog head by mapping it to the catalog.
/// Idempotent (the write path calls it once, but a retried/multi-file commit
/// must not fail on the PK).
async fn advance_catalog_head(
    catalog_id: i64,
    snapshot_id: i64,
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
) -> Result<()> {
    sqlx::query(
        "INSERT INTO ducklake_catalog_snapshot_map (catalog_id, snapshot_id)
         VALUES ($1, $2)
         ON CONFLICT (catalog_id, snapshot_id) DO NOTHING",
    )
    .bind(catalog_id)
    .bind(snapshot_id)
    .execute(&mut **tx)
    .await?;
    Ok(())
}

/// SQL expression resolving a `df`-aliased file row's path relative to the
/// catalog `data_path` root (file → table → schema). Mirrors the multicatalog
/// expire path's `PG_RESOLVED_PATH`; duplicated here (that one is private to
/// `multicatalog`) so the compaction commit can schedule retired files.
const COMPACTION_RESOLVED_PATH: &str = "CASE
    WHEN NOT df.path_is_relative THEN df.path
    WHEN NOT t.path_is_relative THEN t.path || '/' || df.path
    ELSE s.path || '/' || t.path || '/' || df.path
END";

/// Companion to [`COMPACTION_RESOLVED_PATH`]: true only when the whole chain is relative.
const COMPACTION_REL_FLAG: &str =
    "(df.path_is_relative AND t.path_is_relative AND s.path_is_relative)";

/// Insert `(id, resolved_path, rel, owned)` rows (as produced by
/// [`COMPACTION_RESOLVED_PATH`] / [`COMPACTION_REL_FLAG`] plus the file row's
/// `owner_catalog_id IS NULL`) into `ducklake_files_scheduled_for_deletion`, scoped
/// to `catalog_id`. Mirrors the multicatalog expire path's `schedule_pg_files`,
/// including its ownership rule: a file row naming another catalog as its owner is
/// a reference, so the row is retired by the caller but the object is never
/// scheduled here. Ownership is the test, not how the path is spelled — see
/// `schedule_pg_files`.
async fn schedule_compaction_files(
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
    catalog_id: i64,
    rows: Vec<sqlx::postgres::PgRow>,
) -> Result<()> {
    for row in rows {
        let id: i64 = row.try_get(0)?;
        let path: String = row.try_get(1)?;
        let rel: bool = row.try_get(2)?;
        let owned: bool = row.try_get(3)?;
        if !owned {
            continue;
        }
        sqlx::query(
            "INSERT INTO ducklake_files_scheduled_for_deletion
                 (catalog_id, data_file_id, path, path_is_relative, schedule_start,
                  scheduled_by_owner)
             VALUES ($1, $2, $3, $4, NOW(), TRUE)",
        )
        .bind(catalog_id)
        .bind(id)
        .bind(&path)
        .bind(rel)
        .execute(&mut **tx)
        .await?;
    }
    Ok(())
}

/// The atomic commit point for a multicatalog Postgres write, shared by
/// `register_data_file` (with a data file) and `publish_snapshot` (fileless).
/// The CALLER already holds the catalog `FOR UPDATE` lock and an open tx.
///
/// All metadata — the snapshot row, the get-or-create schema/table rows, the
/// column generation, the `schema_versions` row, and the Replace retirement — is
/// written HERE so nothing is visible until `advance_catalog_head` maps the
/// snapshot (the caller runs that LAST). The `snapshot_id` is a plain IDENTITY
/// insert, so per-catalog id order == commit order, which is what makes the
/// scalar [`detect_replace_conflict`] and the dense schema_version computation
/// exact. The reserved schema/table/column ids from begin are inserted with
/// `OVERRIDING SYSTEM VALUE`; the reused column ids keep parquet field-ids stable.
///
/// Returns `(committed_snapshot_id, table_id)`.
///
/// `table_id_hint` is the id reserved for the table at begin; it is used only
/// when the table does not yet exist (first write). The schema id is re-derived
/// here — looked up if the schema already exists, else a fresh id is reserved
/// from the sequence — because the reserved schema id from begin is not threaded
/// through the commit (it is never baked into anything; the parquet path encodes
/// the catalog id, not the schema id).
/// Persist the harvested per-column stats for a just-registered data file
/// (per-file zone maps). See the SQLite writer's equivalent for the rationale.
///
/// Sent as one `UNNEST` insert rather than a statement per column: this runs on
/// every commit, and against a networked Postgres the round trip dominates the
/// work — a wide table otherwise pays one round trip per column. Passing the
/// columns as arrays (rather than a generated `VALUES` list) keeps the bind
/// count fixed at nine, so a table cannot approach Postgres's 65535-parameter
/// ceiling however many columns it has.
async fn insert_file_column_stats(
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
    table_id: i64,
    data_file_id: i64,
    column_stats: &[ColumnStat],
) -> Result<()> {
    if column_stats.is_empty() {
        return Ok(());
    }

    let column_ids: Vec<i64> = column_stats.iter().map(|s| s.column_id).collect();
    let sizes: Vec<Option<i64>> = column_stats.iter().map(|s| s.column_size_bytes).collect();
    let value_counts: Vec<Option<i64>> = column_stats.iter().map(|s| s.value_count).collect();
    let null_counts: Vec<Option<i64>> = column_stats.iter().map(|s| s.null_count).collect();
    let mins: Vec<Option<String>> = column_stats.iter().map(|s| s.min_value.clone()).collect();
    let maxes: Vec<Option<String>> = column_stats.iter().map(|s| s.max_value.clone()).collect();
    let nans: Vec<Option<bool>> = column_stats.iter().map(|s| s.contains_nan).collect();

    sqlx::query(
        "INSERT INTO ducklake_file_column_stats
             (data_file_id, table_id, column_id, column_size_bytes,
              value_count, null_count, min_value, max_value, contains_nan, extra_stats)
         SELECT $1, $2, u.column_id, u.column_size_bytes,
                u.value_count, u.null_count, u.min_value, u.max_value, u.contains_nan, NULL
         FROM UNNEST($3::bigint[], $4::bigint[], $5::bigint[], $6::bigint[],
                     $7::text[], $8::text[], $9::boolean[])
              AS u(column_id, column_size_bytes, value_count, null_count,
                   min_value, max_value, contains_nan)",
    )
    .bind(data_file_id)
    .bind(table_id)
    .bind(&column_ids)
    .bind(&sizes)
    .bind(&value_counts)
    .bind(&null_counts)
    .bind(&mins)
    .bind(&maxes)
    .bind(&nans)
    .execute(&mut **tx)
    .await?;
    Ok(())
}

/// Persist a partitioned data file's partition metadata within the commit
/// transaction: set `ducklake_data_file.partition_id` and insert one
/// `ducklake_file_partition_value` row per partition key. A no-op for an
/// unpartitioned file.
async fn insert_partition_metadata(
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
    table_id: i64,
    data_file_id: i64,
    file: &DataFileInfo,
) -> Result<()> {
    if let Some(partition_id) = file.partition_id {
        sqlx::query("UPDATE ducklake_data_file SET partition_id = $1 WHERE data_file_id = $2")
            .bind(partition_id)
            .bind(data_file_id)
            .execute(&mut **tx)
            .await?;
    }
    for (key_index, value) in &file.partition_values {
        sqlx::query(
            "INSERT INTO ducklake_file_partition_value
                 (data_file_id, table_id, partition_key_index, partition_value)
             VALUES ($1, $2, $3, $4)",
        )
        .bind(data_file_id)
        .bind(table_id)
        .bind(i64::from(*key_index))
        .bind(value.clone())
        .execute(&mut **tx)
        .await?;
    }
    Ok(())
}

/// Read the table's live column generation as `(ColumnDef, column_id)`, ordered
/// by `column_order`. Used to drive [`recompute_table_column_stats`]'s
/// numeric-vs-not classification when the caller (e.g. `retire_appends_since`)
/// doesn't already hold the column list the way a normal write does.
async fn live_columns_for_stats(
    table_id: i64,
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
) -> Result<(Vec<ColumnDef>, Vec<i64>)> {
    let mut columns = Vec::new();
    let mut column_ids = Vec::new();
    for row in sqlx::query(
        "SELECT column_id, column_name, column_type
         FROM ducklake_column
         WHERE table_id = $1 AND end_snapshot IS NULL AND parent_column IS NULL
         ORDER BY column_order",
    )
    .bind(table_id)
    .fetch_all(&mut **tx)
    .await?
    {
        let column_id: i64 = row.try_get(0)?;
        let name: String = row.try_get(1)?;
        let ducklake_type: String = row.try_get(2)?;
        column_ids.push(column_id);
        // Same-crate construction: the type string came from the catalog and is
        // already valid, so skip ColumnDef::new's re-validation. `is_nullable` is
        // irrelevant here — recompute_table_column_stats only reads the type.
        columns.push(ColumnDef {
            name,
            data_type: crate::types::ducklake_to_arrow_type(&ducklake_type)
                .unwrap_or(arrow::datatypes::DataType::Null),
            ducklake_type,
            is_nullable: true,
            initial_default: None,
            default_value: None,
            default_value_type: None,
            default_value_dialect: None,
        });
    }
    Ok((columns, column_ids))
}

/// Recompute `ducklake_table_column_stats` from the table's live files and
/// replace the stored rows. See the SQLite writer's equivalent for the rationale
/// (widen on insert, never tighten on delete; correct for every write mode).
async fn recompute_table_column_stats(
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
    table_id: i64,
    columns: &[ColumnDef],
    column_ids: &[i64],
) -> Result<()> {
    use crate::stats_encode::{FileColumnStat, aggregate_global_column_stats};
    let catalog_columns = catalog_column_defs(columns)?;
    let column_ids = top_level_column_ids(&catalog_columns, column_ids)?;

    let live_file_count: i64 = sqlx::query(
        "SELECT COUNT(*) FROM ducklake_data_file WHERE table_id = $1 AND end_snapshot IS NULL",
    )
    .bind(table_id)
    .fetch_one(&mut **tx)
    .await?
    .try_get(0)?;

    let mut per_file: Vec<FileColumnStat> = Vec::new();
    for row in sqlx::query(
        "SELECT s.column_id, s.min_value, s.max_value, s.null_count, s.contains_nan
         FROM ducklake_file_column_stats s
         JOIN ducklake_data_file d ON d.data_file_id = s.data_file_id
         WHERE d.table_id = $1 AND d.end_snapshot IS NULL",
    )
    .bind(table_id)
    .fetch_all(&mut **tx)
    .await?
    {
        per_file.push(FileColumnStat {
            column_id: row.try_get(0)?,
            min_value: row.try_get(1)?,
            max_value: row.try_get(2)?,
            null_count: row.try_get(3)?,
            contains_nan: row.try_get(4)?,
        });
    }

    let numeric_of = |column_id: i64| -> bool {
        column_ids
            .iter()
            .position(|id| *id == column_id)
            .and_then(|i| columns.get(i))
            .map(|c| crate::stats_encode::is_numeric_ducklake_type(c.ducklake_type()))
            .unwrap_or(false)
    };
    let globals = aggregate_global_column_stats(&per_file, live_file_count, numeric_of);

    sqlx::query("DELETE FROM ducklake_table_column_stats WHERE table_id = $1")
        .bind(table_id)
        .execute(&mut **tx)
        .await?;
    if globals.is_empty() {
        return Ok(());
    }

    // One `UNNEST` insert rather than a statement per column, for the same
    // reason as the per-file stats above: this replace runs on every commit.
    let column_ids: Vec<i64> = globals.iter().map(|g| g.column_id).collect();
    let contains_null: Vec<Option<bool>> = globals.iter().map(|g| g.contains_null).collect();
    let contains_nan: Vec<Option<bool>> = globals.iter().map(|g| g.contains_nan).collect();
    let mins: Vec<Option<String>> = globals.iter().map(|g| g.min_value.clone()).collect();
    let maxes: Vec<Option<String>> = globals.iter().map(|g| g.max_value.clone()).collect();

    sqlx::query(
        "INSERT INTO ducklake_table_column_stats
             (table_id, column_id, contains_null, contains_nan, min_value, max_value, extra_stats)
         SELECT $1, u.column_id, u.contains_null, u.contains_nan, u.min_value, u.max_value, NULL
         FROM UNNEST($2::bigint[], $3::boolean[], $4::boolean[], $5::text[], $6::text[])
              AS u(column_id, contains_null, contains_nan, min_value, max_value)",
    )
    .bind(table_id)
    .bind(&column_ids)
    .bind(&contains_null)
    .bind(&contains_nan)
    .bind(&mins)
    .bind(&maxes)
    .execute(&mut **tx)
    .await?;
    Ok(())
}

#[allow(clippy::too_many_arguments)]
async fn finalize_snapshot(
    catalog_id: i64,
    schema_name: &str,
    table_name: &str,
    table_id_hint: i64,
    columns: &[ColumnDef],
    column_ids: &[i64],
    mode: WriteMode,
    base_snapshot: i64,
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
) -> Result<(i64, i64, i64)> {
    let snapshot_id: i64 = sqlx::query(
        "INSERT INTO ducklake_snapshot (snapshot_time, schema_version)
         VALUES (NOW(), 0) RETURNING snapshot_id",
    )
    .fetch_one(&mut **tx)
    .await?
    .try_get(0)?;
    let mut schema_version = 0;
    let mut schema_changed = false;
    let (schema_id, table_id) = finalize_table_snapshot(
        catalog_id,
        snapshot_id,
        &mut schema_version,
        &mut schema_changed,
        schema_name,
        table_name,
        table_id_hint,
        columns,
        column_ids,
        mode,
        base_snapshot,
        tx,
    )
    .await?;
    Ok((snapshot_id, schema_id, table_id))
}

#[allow(clippy::too_many_arguments)]
#[tracing::instrument(
    name = "ducklake.finalize_table_snapshot",
    level = "info",
    skip_all,
    fields(catalog_id, schema_name, table_name)
)]
async fn finalize_table_snapshot(
    catalog_id: i64,
    snapshot_id: i64,
    schema_version: &mut i64,
    schema_changed: &mut bool,
    schema_name: &str,
    table_name: &str,
    table_id_hint: i64,
    columns: &[ColumnDef],
    column_ids: &[i64],
    mode: WriteMode,
    base_snapshot: i64,
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
) -> Result<(i64, i64)> {
    // 1. Resolve the live schema id under the lock. Reuse it if present; else
    //    reserve a fresh id from the sequence (the row is inserted in step 4 once
    //    the snapshot id exists for begin_snapshot).
    let (schema_id, schema_was_created): (i64, bool) = {
        let existing = sqlx::query(
            "SELECT s.schema_id FROM ducklake_schema s
             JOIN ducklake_catalog_schema_map m ON m.schema_id = s.schema_id
             WHERE m.catalog_id = $1 AND s.schema_name = $2 AND s.end_snapshot IS NULL",
        )
        .bind(catalog_id)
        .bind(schema_name)
        .fetch_optional(&mut **tx)
        .await?;
        if let Some(row) = existing {
            (row.try_get(0)?, false)
        } else {
            let id = reserve_ids("ducklake_schema", "schema_id", 1, tx).await?[0];
            (id, true)
        }
    };

    // 2. Conflict check for Replace, BEFORE inserting any of this writer's rows.
    //    Resolve the table id first; a brand-new table has no prior generation to
    //    conflict with, so skip the check (base == head, no rows exist yet).
    if mode == WriteMode::Replace && !schema_was_created {
        let existing_table_id: Option<i64> = sqlx::query(
            "SELECT table_id FROM ducklake_table
             WHERE schema_id = $1 AND table_name = $2 AND end_snapshot IS NULL",
        )
        .bind(schema_id)
        .bind(table_name)
        .fetch_optional(&mut **tx)
        .await?
        .map(|r| r.try_get(0))
        .transpose()?;
        if let Some(tid) = existing_table_id {
            detect_replace_conflict(tid, base_snapshot, tx).await?;
        }
    }

    // 3. Insert the schema row (with its reserved id) if it is new. The catalog id
    //    is encoded into the schema's *path* (not its name) so two catalogs holding
    //    their own `public` land in disjoint physical subtrees: the reader's
    //    resolution chain (`data_path + schema.path + table.path + file.path`) then
    //    puts files under `cat_{id}/{schema}/{table}/…`, matching the upload
    //    location.
    if schema_was_created {
        let scoped_schema_path = format!("cat_{catalog_id}/{schema_name}");
        sqlx::query(
            "INSERT INTO ducklake_schema
                 (schema_id, schema_name, path, path_is_relative, begin_snapshot)
             OVERRIDING SYSTEM VALUE
             VALUES ($1, $2, $3, TRUE, $4)",
        )
        .bind(schema_id)
        .bind(schema_name)
        .bind(&scoped_schema_path)
        .bind(snapshot_id)
        .execute(&mut **tx)
        .await?;
        sqlx::query(
            "INSERT INTO ducklake_catalog_schema_map (catalog_id, schema_id)
             VALUES ($1, $2)",
        )
        .bind(catalog_id)
        .bind(schema_id)
        .execute(&mut **tx)
        .await?;
    }

    // 4. Get or create the table under the lock.
    let (table_id, table_was_created): (i64, bool) = {
        let existing = sqlx::query(
            "SELECT table_id FROM ducklake_table
             WHERE schema_id = $1 AND table_name = $2 AND end_snapshot IS NULL",
        )
        .bind(schema_id)
        .bind(table_name)
        .fetch_optional(&mut **tx)
        .await?;
        if let Some(row) = existing {
            (row.try_get(0)?, false)
        } else {
            sqlx::query(
                "INSERT INTO ducklake_table
                     (table_id, schema_id, table_name, path, path_is_relative, begin_snapshot)
                 OVERRIDING SYSTEM VALUE
                 VALUES ($1, $2, $3, $4, TRUE, $5)",
            )
            .bind(table_id_hint)
            .bind(schema_id)
            .bind(table_name)
            .bind(table_name)
            .bind(snapshot_id)
            .execute(&mut **tx)
            .await?;
            (table_id_hint, true)
        }
    };

    // 5. Read the columns live at commit to classify DDL vs DML
    //    and drive the surgical column update below.
    let proposed = catalog_column_defs(columns)?;
    if proposed.len() != column_ids.len() {
        return Err(crate::DuckLakeError::InvalidConfig(format!(
            "column_ids has {} entries for {} catalog column nodes",
            column_ids.len(),
            proposed.len()
        )));
    }
    let existing_column_rows = sqlx::query(
        "SELECT column_name, column_type, nulls_allowed, column_order, column_id, parent_column
         FROM ducklake_column
         WHERE table_id = $1 AND end_snapshot IS NULL
         ORDER BY column_order",
    )
    .bind(table_id)
    .fetch_all(&mut **tx)
    .await?;
    let existing_catalog_columns = existing_column_rows
        .iter()
        .map(|row| {
            Ok::<_, sqlx::Error>(ExistingCatalogColumn {
                column_id: row.try_get(4)?,
                name: row.try_get(0)?,
                ducklake_type: row.try_get(1)?,
                parent_column: row.try_get(5)?,
            })
        })
        .collect::<std::result::Result<Vec<_>, _>>()?;
    let existing_nullability = existing_column_rows
        .iter()
        .map(|row| Ok::<_, sqlx::Error>(row.try_get::<Option<bool>, _>(2)?.unwrap_or(true)))
        .collect::<std::result::Result<Vec<_>, _>>()?;
    let committed_ids = assign_column_ids(&proposed, &existing_catalog_columns, column_ids)?;
    if committed_ids != column_ids {
        return Err(crate::DuckLakeError::Conflict(
            "table columns were created concurrently with different field ids; retry the write"
                .to_string(),
        ));
    }
    // column_id -> (column_order, nullable, type) for the surgical update. The
    // column_id is used to detect field-id drift: if the caller's staged parquet
    // baked a column_id (at begin) that no longer matches the committed column
    // (e.g. an Append whose table was created by a concurrent writer with
    // different ids), the file's field-ids would resolve to NULL — so we abort.
    let mut current_by_id: std::collections::HashMap<i64, (i64, bool, String)> =
        std::collections::HashMap::new();
    for row in &existing_column_rows {
        let nullable: bool = row.try_get::<Option<bool>, _>(2)?.unwrap_or(true);
        let order: i64 = row.try_get(3)?;
        let id: i64 = row.try_get(4)?;
        let ducklake_type: String = row.try_get(1)?;
        current_by_id.insert(id, (order, nullable, ducklake_type));
    }

    let is_ddl = table_was_created
        || catalog_columns_differ(
            &existing_catalog_columns,
            &existing_nullability,
            &proposed,
            column_ids,
        );

    // No `< S` window: ids are commit-ordered, so MAX over mapped predecessors is
    // the immediately-preceding version. DDL bumps; DML carries forward (with a v1
    // floor for the very first write to the catalog).
    let prev_max: i64 = sqlx::query(
        "SELECT COALESCE(MAX(s.schema_version), 0) FROM ducklake_snapshot s
         JOIN ducklake_catalog_snapshot_map m ON m.snapshot_id = s.snapshot_id
         WHERE m.catalog_id = $1",
    )
    .bind(catalog_id)
    .fetch_one(&mut **tx)
    .await?
    .try_get(0)?;
    if *schema_version == 0 {
        *schema_version = prev_max.max(1);
    }
    if is_ddl && !*schema_changed {
        *schema_version = prev_max + 1;
        *schema_changed = true;
    }
    sqlx::query("UPDATE ducklake_snapshot SET schema_version = $1 WHERE snapshot_id = $2")
        .bind(*schema_version)
        .bind(snapshot_id)
        .execute(&mut **tx)
        .await?;

    // 6. Write the column generation surgically (mode-independent, matching the
    //    SQLite writer) so each kept column keeps a STABLE column_id (== parquet
    //    field_id). End only removed columns, insert only genuinely-new ones (with
    //    their reserved ids), and sync order/nullability on the rest in place.
    //    Stable ids are required even for Replace: a concurrent in-flight Append
    //    baked the kept columns' ids into its parquet, so re-minting them would make
    //    that Append's rows read back as all-NULL. The prior generation's retired
    //    files keep their old ids for time travel. (The Replace conflict check does
    //    not depend on a column re-mint — see the data-file/column scan above.)
    {
        use std::collections::HashSet;
        let proposed_ids = column_ids.iter().copied().collect::<HashSet<_>>();
        for column_id in current_by_id.keys() {
            if !proposed_ids.contains(column_id) {
                sqlx::query(
                    "UPDATE ducklake_column SET end_snapshot = $1
                     WHERE table_id = $2 AND column_id = $3 AND end_snapshot IS NULL",
                )
                .bind(snapshot_id)
                .bind(table_id)
                .bind(column_id)
                .execute(&mut **tx)
                .await?;
            }
        }
        for (order, (column, column_id)) in proposed.iter().zip(column_ids).enumerate() {
            let parent_id = column.parent_index.map(|index| column_ids[index]);
            match current_by_id.get(column_id) {
                Some((cur_order, cur_nullable, cur_type)) => {
                    // Field-id drift: the staged parquet baked `*column_id` for this
                    // column at begin, but the committed column now has a different
                    // id (a concurrent writer created the table/column with other
                    // ids between this writer's begin and commit). Registering the
                    // file would make this column read back as all-NULL, so abort —
                    // the caller retries against the now-committed schema. (Append
                    // is otherwise not conflict-checked; this guards correctness,
                    // not isolation.)
                    let migrate_type = catalog_column_type_requires_migration(cur_type, column);
                    if migrate_type {
                        sqlx::query(
                            "UPDATE ducklake_column SET end_snapshot = $1
                             WHERE table_id = $2 AND column_id = $3 AND end_snapshot IS NULL",
                        )
                        .bind(snapshot_id)
                        .bind(table_id)
                        .bind(column_id)
                        .execute(&mut **tx)
                        .await?;
                        sqlx::query(
                            "INSERT INTO ducklake_column
                                 (column_id, table_id, column_name, column_type, column_order,
                                  nulls_allowed, parent_column, begin_snapshot)
                             OVERRIDING SYSTEM VALUE
                             VALUES ($1, $2, $3, $4, $5, $6, $7, $8)",
                        )
                        .bind(column_id)
                        .bind(table_id)
                        .bind(&column.name)
                        .bind(&column.ducklake_type)
                        .bind(order as i64)
                        .bind(column.is_nullable)
                        .bind(parent_id)
                        .bind(snapshot_id)
                        .execute(&mut **tx)
                        .await?;
                    } else if *cur_order != order as i64 || *cur_nullable != column.is_nullable {
                        sqlx::query(
                            "UPDATE ducklake_column
                             SET column_order = $1, nulls_allowed = $2
                             WHERE table_id = $3 AND column_id = $4 AND end_snapshot IS NULL",
                        )
                        .bind(order as i64)
                        .bind(column.is_nullable)
                        .bind(table_id)
                        .bind(column_id)
                        .execute(&mut **tx)
                        .await?;
                    }
                },
                None => {
                    sqlx::query(
                        "INSERT INTO ducklake_column
                               (column_id, table_id, column_name, column_type, column_order,
                                nulls_allowed, parent_column, begin_snapshot, initial_default,
                                default_value, default_value_type, default_value_dialect)
                          OVERRIDING SYSTEM VALUE
                          VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12)",
                    )
                    .bind(column_id)
                    .bind(table_id)
                    .bind(&column.name)
                    .bind(&column.ducklake_type)
                    .bind(order as i64)
                    .bind(column.is_nullable)
                    .bind(parent_id)
                    .bind(snapshot_id)
                    .bind(&column.initial_default)
                    .bind(&column.default_value)
                    .bind(&column.default_value_type)
                    .bind(&column.default_value_dialect)
                    .execute(&mut **tx)
                    .await?;
                },
            }
        }
    }

    // 7. Write one schema-version row per DDL
    if is_ddl {
        sqlx::query(
            "INSERT INTO ducklake_schema_versions (begin_snapshot, schema_version, table_id)
             VALUES ($1, $2, $3)",
        )
        .bind(snapshot_id)
        .bind(*schema_version)
        .bind(table_id)
        .execute(&mut **tx)
        .await?;
    }

    // 8. Retire the prior generation and zero visible totals for Replace
    //    totals. The `begin_snapshot < S` guard spares this write's own files.
    if mode == WriteMode::Replace {
        sqlx::query(
            "INSERT INTO ducklake_table_stats (table_id, record_count, next_row_id, file_size_bytes)
             VALUES ($1, 0, 0, 0)
             ON CONFLICT (table_id) DO NOTHING",
        )
        .bind(table_id)
        .execute(&mut **tx)
        .await?;
        retire_prior_generation(table_id, snapshot_id, tx).await?;
    }

    let mut ddl_changes = Vec::new();
    if schema_was_created {
        let schema_entry = format!(
            "created_schema:{}",
            crate::metadata_writer::quote_snapshot_name(schema_name)
        );
        let recorded: Option<String> = sqlx::query_scalar(
            "SELECT changes_made FROM ducklake_snapshot_changes WHERE snapshot_id = $1",
        )
        .bind(snapshot_id)
        .fetch_optional(&mut **tx)
        .await?
        .flatten();
        if !snapshot_has_change(recorded.as_deref().unwrap_or_default(), &schema_entry) {
            ddl_changes.push(schema_entry);
        }
    }
    if table_was_created {
        ddl_changes.push(format!(
            "created_table:{}",
            crate::metadata_writer::quote_snapshot_table(schema_name, table_name),
        ));
    } else if is_ddl {
        ddl_changes.push(format!("altered_table:{table_id}"));
    }
    if !ddl_changes.is_empty() {
        record_snapshot_changes(
            tx,
            snapshot_id,
            &ddl_changes.join(","),
            &SnapshotCommitMetadata::default(),
        )
        .await?;
    }

    Ok((schema_id, table_id))
}

// The flags report retired Parquet files and inlined rows, respectively
pub(crate) async fn replaced_storage(
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
    table_id: i64,
    snapshot_id: i64,
) -> Result<(bool, bool)> {
    let replaced: bool = sqlx::query_scalar(
        "SELECT EXISTS(
            SELECT 1 FROM ducklake_data_file
            WHERE table_id = $1 AND end_snapshot = $2
         )",
    )
    .bind(table_id)
    .bind(snapshot_id)
    .fetch_one(&mut **tx)
    .await?;
    let inlined_tables: Vec<String> = sqlx::query_scalar(
        "SELECT table_name FROM ducklake_inlined_data_tables WHERE table_id = $1",
    )
    .bind(table_id)
    .fetch_all(&mut **tx)
    .await?;
    for table_name in inlined_tables {
        let sql = format!(
            "SELECT EXISTS(SELECT 1 FROM {} WHERE end_snapshot = $1)",
            quote_ident(&table_name)
        );
        if sqlx::query_scalar(AssertSqlSafe(sql))
            .bind(snapshot_id)
            .fetch_one(&mut **tx)
            .await?
        {
            return Ok((replaced, true));
        }
    }
    Ok((replaced, false))
}

async fn record_snapshot_changes(
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
    snapshot_id: i64,
    changes_made: &str,
    commit_metadata: &SnapshotCommitMetadata,
) -> Result<()> {
    sqlx::query(
        "INSERT INTO ducklake_snapshot_changes
             (snapshot_id, changes_made, author, commit_message, commit_extra_info)
         VALUES ($1, $2, $3, $4, $5)
         ON CONFLICT (snapshot_id) DO UPDATE SET
             changes_made = CASE
                 WHEN ducklake_snapshot_changes.changes_made IS NULL THEN EXCLUDED.changes_made
                 WHEN EXCLUDED.changes_made IS NULL THEN ducklake_snapshot_changes.changes_made
                 ELSE ducklake_snapshot_changes.changes_made || ',' || EXCLUDED.changes_made
             END,
             author = EXCLUDED.author,
             commit_message = EXCLUDED.commit_message,
             commit_extra_info = EXCLUDED.commit_extra_info",
    )
    .bind(snapshot_id)
    .bind((!changes_made.is_empty()).then_some(changes_made))
    .bind(commit_metadata.author())
    .bind(commit_metadata.message())
    .bind(commit_metadata.extra_info())
    .execute(&mut **tx)
    .await?;
    Ok(())
}

pub(crate) async fn validate_staged_table(
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
    write: &StagedTableWrite,
) -> Result<()> {
    let ended: Option<Option<i64>> =
        sqlx::query_scalar("SELECT end_snapshot FROM ducklake_table WHERE table_id = $1")
            .bind(write.table_id)
            .fetch_optional(&mut **tx)
            .await?;
    if matches!(ended, Some(Some(_))) {
        return Err(crate::DuckLakeError::Conflict(
            "staged table was dropped".to_string(),
        ));
    }
    let current = sqlx::query("SELECT column_id, column_type, nulls_allowed, parent_column FROM ducklake_column WHERE table_id = $1 AND end_snapshot IS NULL ORDER BY column_order").bind(write.table_id).fetch_all(&mut **tx).await?;
    let proposed = catalog_column_defs(&write.columns)?;
    if current.is_empty() && proposed.len() == write.column_ids.len() {
        return Ok(());
    }
    let mut matches = current.len() == proposed.len() && proposed.len() == write.column_ids.len();
    if matches {
        for (index, (row, column)) in current.iter().zip(&proposed).enumerate() {
            let column_id: i64 = row.try_get("column_id")?;
            let column_type: String = row.try_get("column_type")?;
            let nullable: Option<bool> = row.try_get("nulls_allowed")?;
            let parent: Option<i64> = row.try_get("parent_column")?;
            matches &= column_id == write.column_ids[index]
                && catalog_column_type_equal(&column_type, column)
                && nullable.unwrap_or(true) == column.is_nullable
                && parent == column.parent_index.map(|i| write.column_ids[i]);
        }
    }
    if !matches {
        return Err(crate::DuckLakeError::Conflict(
            "table schema changed after staging".to_string(),
        ));
    }
    Ok(())
}

pub(crate) async fn commit_files_at_snapshot(
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
    snapshot_id: i64,
    write: &StagedTableWrite,
    files: &[DataFileInfo],
) -> Result<()> {
    if files.is_empty() {
        return Err(crate::DuckLakeError::InvalidConfig(
            "multi-table file stage must contain at least one file".to_string(),
        ));
    }
    let live_partition_id: Option<i64> = sqlx::query_scalar(
        "SELECT partition_id FROM ducklake_partition_info
         WHERE table_id = $1 AND end_snapshot IS NULL",
    )
    .bind(write.table_id)
    .fetch_optional(&mut **tx)
    .await?;
    for file in files {
        crate::metadata_writer::enforce_partition_fence(write.table_id, live_partition_id, file)?;
    }
    sqlx::query(
        "INSERT INTO ducklake_table_stats
             (table_id, record_count, next_row_id, file_size_bytes)
         VALUES ($1, 0, 0, 0) ON CONFLICT DO NOTHING",
    )
    .bind(write.table_id)
    .execute(&mut **tx)
    .await?;
    let mut next_row_id: i64 =
        sqlx::query_scalar("SELECT next_row_id FROM ducklake_table_stats WHERE table_id = $1")
            .bind(write.table_id)
            .fetch_one(&mut **tx)
            .await?;
    let mut total_records = 0;
    let mut total_bytes = 0;
    for file in files {
        let data_file_id: i64 = sqlx::query_scalar(
            "INSERT INTO ducklake_data_file
                 (table_id, path, path_is_relative, file_size_bytes, footer_size,
                  record_count, row_id_start, begin_snapshot)
             VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
             RETURNING data_file_id",
        )
        .bind(write.table_id)
        .bind(&file.path)
        .bind(file.path_is_relative)
        .bind(file.file_size_bytes)
        .bind(file.footer_size)
        .bind(file.record_count)
        .bind(next_row_id)
        .bind(snapshot_id)
        .fetch_one(&mut **tx)
        .await?;
        insert_file_column_stats(tx, write.table_id, data_file_id, &file.column_stats).await?;
        insert_partition_metadata(tx, write.table_id, data_file_id, file).await?;
        next_row_id += file.record_count;
        total_records += file.record_count;
        total_bytes += file.file_size_bytes;
    }
    recompute_table_column_stats(tx, write.table_id, &write.columns, &write.column_ids).await?;
    sqlx::query(
        "UPDATE ducklake_table_stats
         SET next_row_id = next_row_id + $1, record_count = record_count + $2,
             file_size_bytes = file_size_bytes + $3
         WHERE table_id = $4",
    )
    .bind(total_records)
    .bind(total_records)
    .bind(total_bytes)
    .bind(write.table_id)
    .execute(&mut **tx)
    .await?;
    Ok(())
}

pub(crate) async fn commit_inlined_at_snapshot(
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
    snapshot_id: i64,
    write: &StagedTableWrite,
    batches: &[RecordBatch],
) -> Result<()> {
    enforce_inline_partition_fence(tx, write.table_id, write.base_snapshot_id).await?;
    let record_count: usize = batches.iter().map(RecordBatch::num_rows).sum();
    if record_count == 0 {
        return Err(crate::DuckLakeError::InvalidConfig(
            "multi-table inline stage must contain at least one row".to_string(),
        ));
    }
    let schema_version: i64 =
        sqlx::query_scalar("SELECT schema_version FROM ducklake_snapshot WHERE snapshot_id = $1")
            .bind(snapshot_id)
            .fetch_one(&mut **tx)
            .await?;
    let physical_name = format!(
        "ducklake_inlined_data_{}_{}",
        write.table_id, schema_version
    );
    let sql_types = write
        .columns
        .iter()
        .zip(batches[0].schema().fields())
        .map(|(column, field)| inlined_postgres_type(field.data_type(), column.ducklake_type()))
        .collect::<Vec<_>>();
    let mut ddl = format!(
        "CREATE TABLE IF NOT EXISTS {} (\
         row_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT",
        quote_ident(&physical_name)
    );
    for ((column, _field), sql_type) in write
        .columns
        .iter()
        .zip(batches[0].schema().fields())
        .zip(&sql_types)
    {
        ddl.push_str(", ");
        ddl.push_str(&quote_ident(column.name()));
        ddl.push(' ');
        ddl.push_str(sql_type);
    }
    ddl.push(')');
    sqlx::query(AssertSqlSafe(ddl)).execute(&mut **tx).await?;
    ensure_postgres_physical_indexes(tx, write.table_id, &physical_name).await?;
    sqlx::query(
        "INSERT INTO ducklake_inlined_data_tables (table_id, table_name, schema_version)
         SELECT $1, $2, $3
         WHERE NOT EXISTS (
             SELECT 1 FROM ducklake_inlined_data_tables
             WHERE table_id = $1 AND schema_version = $3
         )",
    )
    .bind(write.table_id)
    .bind(&physical_name)
    .bind(schema_version)
    .execute(&mut **tx)
    .await?;
    sqlx::query(
        "INSERT INTO ducklake_table_stats
             (table_id, record_count, next_row_id, file_size_bytes)
         VALUES ($1, 0, 0, 0) ON CONFLICT DO NOTHING",
    )
    .bind(write.table_id)
    .execute(&mut **tx)
    .await?;
    let mut row_id: i64 =
        sqlx::query_scalar("SELECT next_row_id FROM ducklake_table_stats WHERE table_id = $1")
            .bind(write.table_id)
            .fetch_one(&mut **tx)
            .await?;
    let column_list = write
        .columns
        .iter()
        .map(|column| quote_ident(column.name()))
        .collect::<Vec<_>>()
        .join(", ");
    for batch in batches {
        for batch_row in 0..batch.num_rows() {
            let mut query = QueryBuilder::<Postgres>::new(format!(
                "INSERT INTO {} (row_id, begin_snapshot, end_snapshot, {}) VALUES (",
                quote_ident(&physical_name),
                column_list
            ));
            query.push_bind(row_id);
            query.push(", ").push_bind(snapshot_id);
            query.push(", NULL");
            for ((array, sql_type), column) in
                batch.columns().iter().zip(&sql_types).zip(&write.columns)
            {
                query.push(", ");
                if write
                    .snapshot_id_columns
                    .iter()
                    .any(|name| name == column.name())
                    && array.is_null(batch_row)
                {
                    query.push_bind(snapshot_id);
                } else {
                    push_inlined_postgres_value(&mut query, array.as_ref(), batch_row, sql_type)?;
                }
            }
            query.push(')');
            query.build().execute(&mut **tx).await?;
            row_id += 1;
        }
    }
    let record_count = i64::try_from(record_count).map_err(|_| {
        crate::DuckLakeError::InvalidConfig("multi-table inline row count exceeds i64".to_string())
    })?;
    sqlx::query(
        "UPDATE ducklake_table_stats
         SET next_row_id = next_row_id + $1, record_count = record_count + $2
         WHERE table_id = $3",
    )
    .bind(record_count)
    .bind(record_count)
    .bind(write.table_id)
    .execute(&mut **tx)
    .await?;
    Ok(())
}

pub(crate) async fn apply_positional_deletes_at_snapshot(
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
    table_id: i64,
    snapshot_id: i64,
    base_snapshot: i64,
    deletes: &[DeleteFileEntry],
) -> Result<()> {
    for entry in deletes {
        let target_live: Option<i64> = sqlx::query_scalar(
            "SELECT data_file_id FROM ducklake_data_file
             WHERE data_file_id = $1 AND end_snapshot IS NULL",
        )
        .bind(entry.data_file_id)
        .fetch_optional(&mut **tx)
        .await?;
        if target_live.is_none() {
            return Err(crate::DuckLakeError::Conflict(format!(
                "DELETE on data file {} could not commit: the file is no longer live since snapshot {base_snapshot}",
                entry.data_file_id
            )));
        }
        let current_previous: Option<i64> = sqlx::query_scalar(
            "SELECT delete_file_id FROM ducklake_delete_file
             WHERE data_file_id = $1 AND end_snapshot IS NULL",
        )
        .bind(entry.data_file_id)
        .fetch_optional(&mut **tx)
        .await?;
        if current_previous != entry.expected_prev_delete_file {
            return Err(crate::DuckLakeError::Conflict(format!(
                "DELETE on data file {} could not commit: its live delete file changed from {:?} to {current_previous:?} since snapshot {base_snapshot}",
                entry.data_file_id, entry.expected_prev_delete_file
            )));
        }
        if let Some(previous) = entry.expected_prev_delete_file {
            sqlx::query(
                "UPDATE ducklake_delete_file SET end_snapshot = $1
                 WHERE delete_file_id = $2 AND end_snapshot IS NULL",
            )
            .bind(snapshot_id)
            .bind(previous)
            .execute(&mut **tx)
            .await?;
        }
        sqlx::query(
            "INSERT INTO ducklake_delete_file
                 (data_file_id, table_id, path, path_is_relative, file_size_bytes,
                  footer_size, delete_count, begin_snapshot)
             VALUES ($1, $2, $3, $4, $5, $6, $7, $8)",
        )
        .bind(entry.data_file_id)
        .bind(table_id)
        .bind(&entry.delete.path)
        .bind(entry.delete.path_is_relative)
        .bind(entry.delete.file_size_bytes)
        .bind(entry.delete.footer_size)
        .bind(entry.delete.delete_count)
        .bind(snapshot_id)
        .execute(&mut **tx)
        .await?;
    }
    Ok(())
}

pub(crate) async fn apply_inlined_deletes_at_snapshot(
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
    table_id: i64,
    snapshot_id: i64,
    base_snapshot: i64,
    deletes: &[InlinedRowRef],
) -> Result<()> {
    if deletes.is_empty() {
        return Ok(());
    }
    let registered: Vec<String> = sqlx::query_scalar(
        "SELECT table_name FROM ducklake_inlined_data_tables WHERE table_id = $1",
    )
    .bind(table_id)
    .fetch_all(&mut **tx)
    .await?;
    let changes: Vec<Option<String>> = sqlx::query_scalar("SELECT changes_made FROM ducklake_snapshot_changes WHERE snapshot_id > $1 AND snapshot_id < $2").bind(base_snapshot).bind(snapshot_id).fetch_all(&mut **tx).await?;
    if changes
        .iter()
        .flatten()
        .any(|changes| inlined_delete_conflicts(changes, table_id))
    {
        return Err(crate::DuckLakeError::Conflict(
            "table changed since the inlined delete snapshot".to_string(),
        ));
    }
    for name in &registered {
        let sql = format!(
            "SELECT EXISTS(SELECT 1 FROM {} WHERE (begin_snapshot > $1 AND begin_snapshot < $2) OR (end_snapshot > $1 AND end_snapshot < $2))",
            quote_ident(name)
        );
        let changed: bool = sqlx::query_scalar(AssertSqlSafe(sql))
            .bind(base_snapshot)
            .bind(snapshot_id)
            .fetch_one(&mut **tx)
            .await?;
        if changed {
            return Err(crate::DuckLakeError::Conflict(
                "inlined rows changed since the delete snapshot".to_string(),
            ));
        }
    }
    for (name, row_ids) in inlined_delete_groups(deletes) {
        if !registered.iter().any(|table| table == name) {
            return Err(crate::DuckLakeError::Conflict(format!(
                "inlined deletes reference unregistered table '{name}'"
            )));
        }
        let ids = row_ids
            .iter()
            .map(i64::to_string)
            .collect::<Vec<_>>()
            .join(",");
        let sql = format!(
            "UPDATE {} SET end_snapshot = $1 WHERE row_id IN ({ids}) AND end_snapshot IS NULL AND begin_snapshot <> $2",
            quote_ident(name)
        );
        sqlx::query(AssertSqlSafe(sql))
            .bind(snapshot_id)
            .bind(snapshot_id)
            .execute(&mut **tx)
            .await?;
    }
    Ok(())
}

pub(crate) fn set_postgres_table_setting(
    pool: &PgPool,
    catalog_id: Option<i64>,
    table_id: i64,
    key: &str,
    value: &str,
) -> Result<()> {
    let key = validate_table_setting(key)?;
    block_on(async {
        let mut tx = pool.begin().await?;
        if let Some(catalog_id) = catalog_id {
            assert_table_in_catalog(catalog_id, table_id, &mut tx).await?;
        }
        let live: Option<i64> = sqlx::query_scalar("SELECT table_id FROM ducklake_table WHERE table_id = $1 AND end_snapshot IS NULL FOR UPDATE").bind(table_id).fetch_optional(&mut *tx).await?;
        if live.is_none() {
            return Err(crate::DuckLakeError::TableNotFound(table_id.to_string()));
        }
        sqlx::query(
            "DELETE FROM ducklake_metadata WHERE key = $1 AND scope = 'table' AND scope_id = $2",
        )
        .bind(&key)
        .bind(table_id)
        .execute(&mut *tx)
        .await?;
        sqlx::query("INSERT INTO ducklake_metadata (key, value, scope, scope_id) VALUES ($1, $2, 'table', $3)").bind(&key).bind(value).bind(table_id).execute(&mut *tx).await?;
        tx.commit().await?;
        Ok(())
    })
}

pub(crate) fn with_postgres_commit_lock(
    pool: &PgPool,
    identity: &str,
    operation: Box<dyn FnOnce() -> Result<()> + '_>,
) -> Result<()> {
    block_on(async {
        let mut tx = pool.begin().await?;
        sqlx::query("SELECT pg_advisory_xact_lock(hashtextextended($1, 0))")
            .bind(identity)
            .execute(&mut *tx)
            .await?;
        let result = operation();
        let unlock = tx.rollback().await;
        match (result, unlock) {
            (Err(e), _) => Err(e),
            (Ok(()), Ok(())) => Ok(()),
            (Ok(()), Err(e)) => Err(e.into()),
        }
    })
}

impl MetadataWriter for PostgresMetadataWriter {
    fn set_table_setting(&self, table_id: i64, key: &str, value: &str) -> Result<()> {
        set_postgres_table_setting(&self.pool, Some(self.catalog_id), table_id, key, value)
    }

    fn set_inlined_index_columns(&self, table_id: i64, columns: &[String]) -> Result<()> {
        let value = encode_inlined_index_columns(columns)?;
        block_on(async {
            // Fast path without the catalog lock: callers re-declare the same
            // columns on every catalog open, and a matching stored value needs
            // neither the lock nor a DELETE+INSERT rewrite.
            let existing: Option<String> = sqlx::query_scalar(
                "SELECT value FROM ducklake_metadata
                 WHERE key = $1 AND scope = 'table' AND scope_id = $2",
            )
            .bind(INLINED_INDEX_COLUMNS_SETTING)
            .bind(table_id)
            .fetch_optional(&self.pool)
            .await?;
            if existing.as_deref() == Some(value.as_str()) {
                return Ok(());
            }
            let mut transaction = self.pool.begin().await?;
            lock_catalog(self.catalog_id, self.lock_timeout_ms, &mut transaction).await?;
            assert_table_in_catalog(self.catalog_id, table_id, &mut transaction).await?;
            let available = sqlx::query_scalar::<_, String>(
                "SELECT column_name FROM ducklake_column
                 WHERE table_id = $1 AND end_snapshot IS NULL AND parent_column IS NULL",
            )
            .bind(table_id)
            .fetch_all(&mut *transaction)
            .await?
            .into_iter()
            .collect::<std::collections::HashSet<_>>();
            validate_inlined_index_columns(columns, &available)?;
            sqlx::query(
                "DELETE FROM ducklake_metadata
                 WHERE key = $1 AND scope = 'table' AND scope_id = $2",
            )
            .bind(INLINED_INDEX_COLUMNS_SETTING)
            .bind(table_id)
            .execute(&mut *transaction)
            .await?;
            sqlx::query(
                "INSERT INTO ducklake_metadata (key, value, scope, scope_id)
                 VALUES ($1, $2, 'table', $3)",
            )
            .bind(INLINED_INDEX_COLUMNS_SETTING)
            .bind(value)
            .bind(table_id)
            .execute(&mut *transaction)
            .await?;
            transaction.commit().await?;
            Ok(())
        })
    }

    fn ensure_inlined_indexes(&self, table_id: i64) -> Result<()> {
        block_on(async {
            let mut transaction = self.pool.begin().await?;
            lock_catalog(self.catalog_id, self.lock_timeout_ms, &mut transaction).await?;
            assert_table_in_catalog(self.catalog_id, table_id, &mut transaction).await?;
            let physical_tables = sqlx::query_scalar::<_, String>(
                "SELECT table_name FROM ducklake_inlined_data_tables WHERE table_id = $1",
            )
            .bind(table_id)
            .fetch_all(&mut *transaction)
            .await?;
            postgres_inlined_index_columns(&mut transaction, table_id).await?;
            for physical_table in physical_tables {
                ensure_postgres_physical_indexes(&mut transaction, table_id, &physical_table)
                    .await?;
            }
            transaction.commit().await?;
            Ok(())
        })
    }

    fn with_commit_lock(
        &self,
        identity: &str,
        operation: Box<dyn FnOnce() -> Result<()> + '_>,
    ) -> Result<()> {
        with_postgres_commit_lock(
            &self.pool,
            &format!("{}:{identity}", self.catalog_id),
            operation,
        )
    }

    fn create_snapshot(&self) -> Result<i64> {
        block_on(async {
            let mut tx = self.pool.begin().await?;

            let row = sqlx::query(
                "INSERT INTO ducklake_snapshot (snapshot_time, schema_version)
                 VALUES (CURRENT_TIMESTAMP, 0) RETURNING snapshot_id",
            )
            .fetch_one(&mut *tx)
            .await?;
            let snapshot_id: i64 = row.try_get(0)?;

            sqlx::query(
                "INSERT INTO ducklake_catalog_snapshot_map (catalog_id, snapshot_id)
                 VALUES ($1, $2)",
            )
            .bind(self.catalog_id)
            .bind(snapshot_id)
            .execute(&mut *tx)
            .await?;

            record_snapshot_changes(&mut tx, snapshot_id, "", &SnapshotCommitMetadata::default())
                .await?;
            tx.commit().await?;
            Ok(snapshot_id)
        })
    }

    fn promote_column_type(
        &self,
        table_id: i64,
        column_name: &str,
        new_ducklake_type: &str,
    ) -> Result<i64> {
        // Reject an unknown target type before opening a transaction.
        crate::types::ducklake_to_arrow_type(new_ducklake_type)?;
        block_on(async {
            let mut tx = self.pool.begin().await?;
            lock_catalog(self.catalog_id, self.lock_timeout_ms, &mut tx).await?;
            // Ownership guard (matches every other table_id-taking mutator,
            // e.g. set_columns / end_table_files): table_ids are global across the
            // multicatalog store, so refuse a table_id that belongs to a different
            // catalog — otherwise a promote scoped to this catalog could silently
            // mutate another catalog's column.
            assert_table_in_catalog(self.catalog_id, table_id, &mut tx).await?;

            // Live version of the column.
            let row = sqlx::query(
                "SELECT column_id, column_type, column_order, nulls_allowed, parent_column,
                        initial_default, default_value, default_value_type, default_value_dialect
                  FROM ducklake_column
                 WHERE table_id = $1 AND column_name = $2 AND end_snapshot IS NULL
                   AND parent_column IS NULL",
            )
            .bind(table_id)
            .bind(column_name)
            .fetch_optional(&mut *tx)
            .await?
            .ok_or_else(|| {
                crate::DuckLakeError::InvalidConfig(format!(
                    "promote_column_type: no live column '{column_name}' in table {table_id}"
                ))
            })?;
            let column_id: i64 = row.try_get("column_id")?;
            let cur_type: String = row.try_get("column_type")?;
            let column_order: i64 = row.try_get("column_order")?;
            let nulls_allowed: bool = row
                .try_get::<Option<bool>, _>("nulls_allowed")?
                .unwrap_or(true);
            let parent_column: Option<i64> = row.try_get("parent_column")?;
            let initial_default: Option<String> = row.try_get("initial_default")?;
            let default_value: Option<String> = row.try_get("default_value")?;
            let default_value_type: Option<String> = row.try_get("default_value_type")?;
            let default_value_dialect: Option<String> = row.try_get("default_value_dialect")?;

            // No-op / not-a-widening guards (canonical first so an alias-only
            // restatement is "no change", not attempted).
            if crate::types::types_equal_canonical(&cur_type, new_ducklake_type) {
                return Err(crate::DuckLakeError::InvalidConfig(format!(
                    "promote_column_type: column '{column_name}' is already type '{cur_type}' (no change)"
                )));
            }
            if !crate::types::is_promotable(&cur_type, new_ducklake_type) {
                return Err(crate::DuckLakeError::UnsupportedTypeChange {
                    operation: TypeChangeOperation::PromoteColumnType,
                    column: column_name.to_string(),
                    from: cur_type,
                    to: new_ducklake_type.to_string(),
                });
            }

            // New snapshot + advance this catalog's head.
            let snapshot_id: i64 = sqlx::query(
                "INSERT INTO ducklake_snapshot (snapshot_time, schema_version)
                 VALUES (NOW(), 0) RETURNING snapshot_id",
            )
            .fetch_one(&mut *tx)
            .await?
            .try_get(0)?;
            sqlx::query(
                "INSERT INTO ducklake_catalog_snapshot_map (catalog_id, snapshot_id) VALUES ($1, $2)",
            )
            .bind(self.catalog_id)
            .bind(snapshot_id)
            .execute(&mut *tx)
            .await?;

            // A promote IS schema evolution → bump schema_version (per-catalog dense)
            // and record the ledger row (same model as a DDL data-write commit).
            let prev_max: i64 = sqlx::query(
                "SELECT COALESCE(MAX(s.schema_version), 0) FROM ducklake_snapshot s
                 JOIN ducklake_catalog_snapshot_map m ON m.snapshot_id = s.snapshot_id
                 WHERE m.catalog_id = $1 AND s.snapshot_id <> $2",
            )
            .bind(self.catalog_id)
            .bind(snapshot_id)
            .fetch_one(&mut *tx)
            .await?
            .try_get(0)?;
            let new_schema_version = prev_max + 1;
            sqlx::query("UPDATE ducklake_snapshot SET schema_version = $1 WHERE snapshot_id = $2")
                .bind(new_schema_version)
                .bind(snapshot_id)
                .execute(&mut *tx)
                .await?;
            sqlx::query(
                "INSERT INTO ducklake_schema_versions (begin_snapshot, schema_version, table_id)
                 VALUES ($1, $2, $3)",
            )
            .bind(snapshot_id)
            .bind(new_schema_version)
            .bind(table_id)
            .execute(&mut *tx)
            .await?;

            record_snapshot_changes(
                &mut tx,
                snapshot_id,
                &format!("altered_table:{table_id}"),
                &SnapshotCommitMetadata::default(),
            )
            .await?;

            // Retire the live row, insert the new version with the SAME column_id
            // (OVERRIDING SYSTEM VALUE — column_id is IDENTITY). Retire-before-insert
            // keeps the live-version partial unique index satisfied at all times.
            sqlx::query(
                "UPDATE ducklake_column SET end_snapshot = $1
                 WHERE table_id = $2 AND column_id = $3 AND end_snapshot IS NULL",
            )
            .bind(snapshot_id)
            .bind(table_id)
            .bind(column_id)
            .execute(&mut *tx)
            .await?;
            sqlx::query(
                "INSERT INTO ducklake_column
                     (column_id, table_id, column_name, column_type, column_order, nulls_allowed,
                      parent_column, begin_snapshot, initial_default, default_value,
                      default_value_type, default_value_dialect)
                 OVERRIDING SYSTEM VALUE
                 VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12)",
            )
            .bind(column_id)
            .bind(table_id)
            .bind(column_name)
            .bind(new_ducklake_type)
            .bind(column_order)
            .bind(nulls_allowed)
            .bind(parent_column)
            .bind(snapshot_id)
            .bind(initial_default)
            .bind(default_value)
            .bind(default_value_type)
            .bind(default_value_dialect)
            .execute(&mut *tx)
            .await?;

            tx.commit().await?;
            Ok(snapshot_id)
        })
    }

    fn set_global_setting(&self, key: &str, value: &str) -> Result<()> {
        block_on(async {
            let mut transaction = self.pool.begin().await?;
            lock_catalog(self.catalog_id, self.lock_timeout_ms, &mut transaction).await?;
            sqlx::query(
                "DELETE FROM ducklake_metadata
                 WHERE key = $1 AND scope = 'catalog' AND scope_id = $2",
            )
            .bind(key)
            .bind(self.catalog_id)
            .execute(&mut *transaction)
            .await?;
            sqlx::query(
                "INSERT INTO ducklake_metadata (key, value, scope, scope_id)
                 VALUES ($1, $2, 'catalog', $3)",
            )
            .bind(key)
            .bind(value)
            .bind(self.catalog_id)
            .execute(&mut *transaction)
            .await?;
            transaction.commit().await?;
            Ok(())
        })
    }

    fn get_or_create_schema(
        &self,
        name: &str,
        path: Option<&str>,
        snapshot_id: i64,
    ) -> Result<(i64, bool)> {
        validate_name(name, "Schema")?;
        block_on(async {
            let mut tx = self.pool.begin().await?;
            lock_catalog(self.catalog_id, self.lock_timeout_ms, &mut tx).await?;

            let existing = sqlx::query(
                "SELECT s.schema_id FROM ducklake_schema s
                 JOIN ducklake_catalog_schema_map m ON m.schema_id = s.schema_id
                 WHERE m.catalog_id = $1 AND s.schema_name = $2 AND s.end_snapshot IS NULL",
            )
            .bind(self.catalog_id)
            .bind(name)
            .fetch_optional(&mut *tx)
            .await?;

            if let Some(row) = existing {
                let id: i64 = row.try_get(0)?;
                tx.commit().await?;
                return Ok((id, false));
            }

            let schema_path = path.unwrap_or(name);
            let row = sqlx::query(
                "INSERT INTO ducklake_schema (schema_name, path, path_is_relative, begin_snapshot)
                 VALUES ($1, $2, TRUE, $3) RETURNING schema_id",
            )
            .bind(name)
            .bind(schema_path)
            .bind(snapshot_id)
            .fetch_one(&mut *tx)
            .await?;
            let schema_id: i64 = row.try_get(0)?;

            sqlx::query(
                "INSERT INTO ducklake_catalog_schema_map (catalog_id, schema_id)
                 VALUES ($1, $2)",
            )
            .bind(self.catalog_id)
            .bind(schema_id)
            .execute(&mut *tx)
            .await?;

            record_snapshot_changes(
                &mut tx,
                snapshot_id,
                &format!(
                    "created_schema:{}",
                    crate::metadata_writer::quote_snapshot_name(name),
                ),
                &SnapshotCommitMetadata::default(),
            )
            .await?;
            tx.commit().await?;
            Ok((schema_id, true))
        })
    }

    fn get_or_create_table(
        &self,
        schema_id: i64,
        name: &str,
        path: Option<&str>,
        snapshot_id: i64,
    ) -> Result<(i64, bool)> {
        validate_name(name, "Table")?;
        block_on(async {
            let mut tx = self.pool.begin().await?;
            lock_catalog(self.catalog_id, self.lock_timeout_ms, &mut tx).await?;
            assert_schema_in_catalog(self.catalog_id, schema_id, &mut tx).await?;

            let existing = sqlx::query(
                "SELECT table_id FROM ducklake_table
                 WHERE schema_id = $1 AND table_name = $2 AND end_snapshot IS NULL",
            )
            .bind(schema_id)
            .bind(name)
            .fetch_optional(&mut *tx)
            .await?;

            if let Some(row) = existing {
                let id: i64 = row.try_get(0)?;
                tx.commit().await?;
                return Ok((id, false));
            }

            let schema_name: String =
                sqlx::query_scalar("SELECT schema_name FROM ducklake_schema WHERE schema_id = $1")
                    .bind(schema_id)
                    .fetch_one(&mut *tx)
                    .await?;

            let table_path = path.unwrap_or(name);
            let row = sqlx::query(
                "INSERT INTO ducklake_table (schema_id, table_name, path, path_is_relative, begin_snapshot)
                 VALUES ($1, $2, $3, TRUE, $4) RETURNING table_id",
            )
            .bind(schema_id)
            .bind(name)
            .bind(table_path)
            .bind(snapshot_id)
            .fetch_one(&mut *tx)
            .await?;
            let id: i64 = row.try_get(0)?;

            record_snapshot_changes(
                &mut tx,
                snapshot_id,
                &format!(
                    "created_table:{}",
                    crate::metadata_writer::quote_snapshot_table(&schema_name, name),
                ),
                &SnapshotCommitMetadata::default(),
            )
            .await?;
            tx.commit().await?;
            Ok((id, true))
        })
    }

    fn set_columns(
        &self,
        table_id: i64,
        columns: &[ColumnDef],
        snapshot_id: i64,
    ) -> Result<Vec<i64>> {
        if columns.is_empty() {
            return Err(crate::DuckLakeError::InvalidConfig(
                "Table must have at least one column".to_string(),
            ));
        }
        block_on(async {
            let mut tx = self.pool.begin().await?;
            lock_catalog(self.catalog_id, self.lock_timeout_ms, &mut tx).await?;
            assert_table_in_catalog(self.catalog_id, table_id, &mut tx).await?;
            let table_begin_snapshot: i64 =
                sqlx::query_scalar("SELECT begin_snapshot FROM ducklake_table WHERE table_id = $1")
                    .bind(table_id)
                    .fetch_one(&mut *tx)
                    .await?;

            sqlx::query(
                "UPDATE ducklake_column SET end_snapshot = $1
                 WHERE table_id = $2 AND end_snapshot IS NULL",
            )
            .bind(snapshot_id)
            .bind(table_id)
            .execute(&mut *tx)
            .await?;

            let catalog_columns = catalog_column_defs(columns)?;
            let mut column_ids = Vec::with_capacity(catalog_columns.len());
            for (order, column) in catalog_columns.iter().enumerate() {
                let parent_id = column.parent_index.map(|index| column_ids[index]);
                let row = sqlx::query(
                    "INSERT INTO ducklake_column
                           (table_id, column_name, column_type, column_order, nulls_allowed,
                            parent_column, begin_snapshot, initial_default, default_value,
                            default_value_type, default_value_dialect)
                       VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)
                       RETURNING column_id",
                )
                .bind(table_id)
                .bind(&column.name)
                .bind(&column.ducklake_type)
                .bind(order as i64)
                .bind(column.is_nullable)
                .bind(parent_id)
                .bind(snapshot_id)
                .bind(&column.initial_default)
                .bind(&column.default_value)
                .bind(&column.default_value_type)
                .bind(&column.default_value_dialect)
                .fetch_one(&mut *tx)
                .await?;
                column_ids.push(row.try_get(0)?);
            }

            if table_begin_snapshot != snapshot_id {
                record_snapshot_changes(
                    &mut tx,
                    snapshot_id,
                    &format!("altered_table:{table_id}"),
                    &SnapshotCommitMetadata::default(),
                )
                .await?;
            }
            tx.commit().await?;
            top_level_column_ids(&catalog_columns, &column_ids)
        })
    }

    fn register_data_file(
        &self,
        table_id: i64,
        schema_name: &str,
        table_name: &str,
        snapshot_id: i64,
        file: &DataFileInfo,
        mode: WriteMode,
        base_snapshot: i64,
        columns: &[ColumnDef],
        column_ids: &[i64],
    ) -> Result<CommitIds> {
        self.register_data_file_with_commit_metadata(
            table_id,
            schema_name,
            table_name,
            snapshot_id,
            file,
            mode,
            base_snapshot,
            columns,
            column_ids,
            &SnapshotCommitMetadata::default(),
            None,
        )
    }

    fn register_data_file_with_commit_metadata(
        &self,
        table_id: i64,
        schema_name: &str,
        table_name: &str,
        _snapshot_id: i64,
        file: &DataFileInfo,
        mode: WriteMode,
        base_snapshot: i64,
        columns: &[ColumnDef],
        column_ids: &[i64],
        commit_metadata: &SnapshotCommitMetadata,
        expected_base_snapshot_id: Option<i64>,
    ) -> Result<CommitIds> {
        block_on(async {
            // Single atomic commit. finalize_snapshot writes ALL metadata (the
            // snapshot row, get-or-create schema/table, the column generation, the
            // schema_versions row, and the Replace retirement) and returns the
            // committed snapshot id + real table id. We then register the file and
            // advance the catalog head LAST, so nothing is visible until the head
            // maps the snapshot. row_id_start is drawn from the table's monotonic
            // counter under the catalog lock so concurrent writers hand out
            // non-overlapping ranges; the stats row is seeded for tables created
            // before this writer maintained it. The passed `table_id` is the id
            // reserved at begin (== the committed id); we tolerate it not existing
            // yet (first write) but reject another catalog's id.
            let mut tx = self.pool.begin().await?;
            lock_catalog(self.catalog_id, self.lock_timeout_ms, &mut tx).await?;
            assert_table_not_in_other_catalog(self.catalog_id, table_id, &mut tx).await?;
            if mode != WriteMode::Replace
                && let Some(expected_base_snapshot_id) = expected_base_snapshot_id
            {
                detect_replace_conflict(table_id, expected_base_snapshot_id, &mut tx).await?;
            }
            let (snapshot_id, schema_id, table_id) = finalize_snapshot(
                self.catalog_id,
                schema_name,
                table_name,
                table_id,
                columns,
                column_ids,
                mode,
                base_snapshot,
                &mut tx,
            )
            .await?;

            // Partition-spec fence: this file must be consistent with the table's live
            // partition generation at commit time (both directions — see
            // enforce_partition_fence). Runs inside the lock_catalog-serialized tx;
            // rolls back on a Conflict.
            let live_partition_id: Option<i64> = sqlx::query_scalar(
                "SELECT partition_id FROM ducklake_partition_info
                 WHERE table_id = $1 AND end_snapshot IS NULL",
            )
            .bind(table_id)
            .fetch_optional(&mut *tx)
            .await?;
            crate::metadata_writer::enforce_partition_fence(table_id, live_partition_id, file)?;

            sqlx::query(
                "INSERT INTO ducklake_table_stats (table_id, record_count, next_row_id, file_size_bytes)
                 VALUES ($1, 0, 0, 0)
                 ON CONFLICT (table_id) DO NOTHING",
            )
            .bind(table_id)
            .execute(&mut *tx)
            .await?;

            let stats_row =
                sqlx::query("SELECT next_row_id FROM ducklake_table_stats WHERE table_id = $1")
                    .bind(table_id)
                    .fetch_one(&mut *tx)
                    .await?;
            let next_row_id: i64 = stats_row.try_get(0)?;
            let row_ids = crate::metadata_writer::allocate_row_id_start(next_row_id, file);

            let inserted = sqlx::query(
                "INSERT INTO ducklake_data_file
                     (table_id, path, path_is_relative, file_size_bytes,
                      footer_size, record_count, row_id_start, begin_snapshot)
                 VALUES ($1, $2, $3, $4, $5, $6, $7, $8) RETURNING data_file_id",
            )
            .bind(table_id)
            .bind(&file.path)
            .bind(file.path_is_relative)
            .bind(file.file_size_bytes)
            .bind(file.footer_size)
            .bind(file.record_count)
            .bind(row_ids.stored)
            .bind(snapshot_id)
            .fetch_one(&mut *tx)
            .await?;
            let data_file_id: i64 = inserted.try_get(0)?;

            // Persist the file's per-column zone maps + refresh the table roll-up,
            // in the same commit as the data-file row.
            insert_file_column_stats(&mut tx, table_id, data_file_id, &file.column_stats).await?;
            insert_partition_metadata(&mut tx, table_id, data_file_id, file).await?;
            recompute_table_column_stats(&mut tx, table_id, columns, column_ids).await?;

            // Advance the counter and accumulate stats. `next_row_id`
            // monotonically increases over the table's lifetime — rowids
            // are never reused, even after end-snapshot. For Replace the
            // record/byte totals were just zeroed, so this leaves them at the
            // new file's values.
            sqlx::query(
                "UPDATE ducklake_table_stats
                 SET next_row_id     = next_row_id + $1,
                     record_count    = record_count + $2,
                     file_size_bytes = file_size_bytes + $3
                 WHERE table_id = $4",
            )
            .bind(row_ids.advance)
            .bind(file.record_count)
            .bind(file.file_size_bytes)
            .bind(table_id)
            .execute(&mut *tx)
            .await?;

            let replaced_existing_data = replaced_storage(&mut tx, table_id, snapshot_id).await?;
            let changes_made = table_write_changes(table_id, mode, false, replaced_existing_data);
            record_snapshot_changes(&mut tx, snapshot_id, &changes_made, commit_metadata).await?;

            // advance_catalog_head MUST be the last write before commit.
            advance_catalog_head(self.catalog_id, snapshot_id, &mut tx).await?;

            tx.commit().await?;
            Ok(CommitIds {
                snapshot_id,
                schema_id,
                table_id,
            })
        })
    }

    #[allow(clippy::too_many_arguments)]
    fn register_data_files(
        &self,
        table_id: i64,
        schema_name: &str,
        table_name: &str,
        snapshot_id: i64,
        files: &[DataFileInfo],
        mode: WriteMode,
        base_snapshot: i64,
        columns: &[ColumnDef],
        column_ids: &[i64],
    ) -> Result<CommitIds> {
        self.register_data_files_with_commit_metadata(
            table_id,
            schema_name,
            table_name,
            snapshot_id,
            files,
            mode,
            base_snapshot,
            columns,
            column_ids,
            &SnapshotCommitMetadata::default(),
            None,
        )
    }

    #[allow(clippy::too_many_arguments)]
    #[tracing::instrument(
        name = "ducklake.register_data_files",
        level = "info",
        skip_all,
        fields(table_id, schema_name, table_name, files = files.len(), columns = columns.len())
    )]
    fn register_data_files_with_commit_metadata(
        &self,
        table_id: i64,
        schema_name: &str,
        table_name: &str,
        _snapshot_id: i64,
        files: &[DataFileInfo],
        mode: WriteMode,
        base_snapshot: i64,
        columns: &[ColumnDef],
        column_ids: &[i64],
        commit_metadata: &SnapshotCommitMetadata,
        expected_base_snapshot_id: Option<i64>,
    ) -> Result<CommitIds> {
        if files.is_empty() {
            return Err(crate::DuckLakeError::InvalidConfig(
                "register_data_files: files must be non-empty".to_string(),
            ));
        }
        block_on(async {
            let mut tx = self.pool.begin().await?;
            lock_catalog(self.catalog_id, self.lock_timeout_ms, &mut tx).await?;
            assert_table_not_in_other_catalog(self.catalog_id, table_id, &mut tx).await?;
            if mode != WriteMode::Replace
                && let Some(expected_base_snapshot_id) = expected_base_snapshot_id
            {
                detect_replace_conflict(table_id, expected_base_snapshot_id, &mut tx).await?;
            }
            let (snapshot_id, schema_id, table_id) = finalize_snapshot(
                self.catalog_id,
                schema_name,
                table_name,
                table_id,
                columns,
                column_ids,
                mode,
                base_snapshot,
                &mut tx,
            )
            .await?;
            // Partition-spec fence (both directions, every file): each file must be
            // consistent with the table's live partition generation at commit time.
            // Runs inside the lock_catalog-serialized tx; rolls back on a Conflict.
            // table_ids are global across the multicatalog store, so
            // ducklake_partition_info scopes by table_id alone.
            let live_partition_id: Option<i64> = sqlx::query_scalar(
                "SELECT partition_id FROM ducklake_partition_info
                 WHERE table_id = $1 AND end_snapshot IS NULL",
            )
            .bind(table_id)
            .fetch_optional(&mut *tx)
            .await?;
            for file in files {
                crate::metadata_writer::enforce_partition_fence(table_id, live_partition_id, file)?;
            }
            sqlx::query(
                "INSERT INTO ducklake_table_stats (table_id, record_count, next_row_id, file_size_bytes)
                 VALUES ($1, 0, 0, 0)
                 ON CONFLICT (table_id) DO NOTHING",
            )
            .bind(table_id)
            .execute(&mut *tx)
            .await?;
            let mut next_row_id: i64 =
                sqlx::query("SELECT next_row_id FROM ducklake_table_stats WHERE table_id = $1")
                    .bind(table_id)
                    .fetch_one(&mut *tx)
                    .await?
                    .try_get(0)?;
            let mut total_advance: i64 = 0;
            let mut total_records: i64 = 0;
            let mut total_bytes: i64 = 0;
            for file in files {
                let row_ids = crate::metadata_writer::allocate_row_id_start(next_row_id, file);
                let inserted = sqlx::query(
                    "INSERT INTO ducklake_data_file
                         (table_id, path, path_is_relative, file_size_bytes,
                          footer_size, record_count, row_id_start, begin_snapshot)
                     VALUES ($1, $2, $3, $4, $5, $6, $7, $8) RETURNING data_file_id",
                )
                .bind(table_id)
                .bind(&file.path)
                .bind(file.path_is_relative)
                .bind(file.file_size_bytes)
                .bind(file.footer_size)
                .bind(file.record_count)
                .bind(row_ids.stored)
                .bind(snapshot_id)
                .fetch_one(&mut *tx)
                .await?;
                let data_file_id: i64 = inserted.try_get(0)?;
                insert_file_column_stats(&mut tx, table_id, data_file_id, &file.column_stats)
                    .await?;
                insert_partition_metadata(&mut tx, table_id, data_file_id, file).await?;
                next_row_id += row_ids.advance;
                total_advance += row_ids.advance;
                total_records += file.record_count;
                total_bytes += file.file_size_bytes;
            }
            recompute_table_column_stats(&mut tx, table_id, columns, column_ids).await?;
            sqlx::query(
                "UPDATE ducklake_table_stats
                 SET next_row_id     = next_row_id + $1,
                     record_count    = record_count + $2,
                     file_size_bytes = file_size_bytes + $3
                 WHERE table_id = $4",
            )
            .bind(total_advance)
            .bind(total_records)
            .bind(total_bytes)
            .bind(table_id)
            .execute(&mut *tx)
            .await?;
            let replaced_existing_data = replaced_storage(&mut tx, table_id, snapshot_id).await?;
            let changes_made = table_write_changes(table_id, mode, false, replaced_existing_data);
            record_snapshot_changes(&mut tx, snapshot_id, &changes_made, commit_metadata).await?;
            // advance_catalog_head MUST be the last write before commit.
            advance_catalog_head(self.catalog_id, snapshot_id, &mut tx).await?;
            tx.commit().await?;
            Ok(CommitIds {
                snapshot_id,
                schema_id,
                table_id,
            })
        })
    }

    #[allow(clippy::too_many_arguments)]
    fn supports_data_inlining(&self, schema: &arrow::datatypes::Schema) -> bool {
        if schema
            .fields()
            .iter()
            .any(|field| is_inlined_system_column(field.name()) || field.name().len() > 63)
        {
            return false;
        }
        schema
            .fields()
            .iter()
            .all(|field| postgres_type_inlines(field.data_type()))
    }

    fn supports_data_inlining_values(&self, batches: &[RecordBatch]) -> bool {
        batches
            .iter()
            .flat_map(RecordBatch::columns)
            .all(|array| crate::nested_inline::values_support_inlining(array.as_ref()))
    }

    #[allow(clippy::too_many_arguments)]
    fn register_inlined_data(
        &self,
        table_id: i64,
        schema_name: &str,
        table_name: &str,
        _snapshot_id: i64,
        batches: &[RecordBatch],
        mode: WriteMode,
        base_snapshot: i64,
        columns: &[ColumnDef],
        column_ids: &[i64],
        commit_metadata: &SnapshotCommitMetadata,
        expected_base_snapshot_id: Option<i64>,
    ) -> Result<CommitIds> {
        let record_count: usize = batches.iter().map(RecordBatch::num_rows).sum();
        if record_count == 0 {
            return Err(crate::DuckLakeError::InvalidConfig(
                "register_inlined_data: batches must contain at least one row".to_string(),
            ));
        }
        if batches
            .iter()
            .any(|batch| batch.num_columns() != columns.len())
        {
            return Err(crate::DuckLakeError::InvalidConfig(
                "register_inlined_data: batch schema does not match table columns".to_string(),
            ));
        }

        block_on(async {
            let mut tx = self.pool.begin().await?;
            lock_catalog(self.catalog_id, self.lock_timeout_ms, &mut tx).await?;
            assert_table_not_in_other_catalog(self.catalog_id, table_id, &mut tx).await?;
            enforce_inline_partition_fence(&mut tx, table_id, base_snapshot).await?;
            let (snapshot_id, schema_id, table_id) = finalize_snapshot(
                self.catalog_id,
                schema_name,
                table_name,
                table_id,
                columns,
                column_ids,
                mode,
                base_snapshot,
                &mut tx,
            )
            .await?;
            if mode != WriteMode::Replace
                && let Some(expected_base_snapshot_id) = expected_base_snapshot_id
            {
                detect_replace_conflict(table_id, expected_base_snapshot_id, &mut tx).await?;
            }
            let schema_version: i64 = sqlx::query_scalar(
                "SELECT schema_version FROM ducklake_snapshot WHERE snapshot_id = $1",
            )
            .bind(snapshot_id)
            .fetch_one(&mut *tx)
            .await?;
            let physical_name = format!("ducklake_inlined_data_{table_id}_{schema_version}");
            let sql_types = columns
                .iter()
                .zip(batches[0].schema().fields())
                .map(|(column, field)| {
                    inlined_postgres_type(field.data_type(), column.ducklake_type())
                })
                .collect::<Vec<_>>();
            let mut ddl = format!(
                "CREATE TABLE IF NOT EXISTS {} (\
                 row_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT",
                quote_ident(&physical_name)
            );
            for ((column, _field), sql_type) in columns
                .iter()
                .zip(batches[0].schema().fields())
                .zip(&sql_types)
            {
                ddl.push_str(", ");
                ddl.push_str(&quote_ident(column.name()));
                ddl.push(' ');
                ddl.push_str(sql_type);
            }
            ddl.push(')');
            sqlx::query(AssertSqlSafe(ddl)).execute(&mut *tx).await?;
            ensure_postgres_physical_indexes(&mut tx, table_id, &physical_name).await?;
            sqlx::query(
                "INSERT INTO ducklake_inlined_data_tables
                     (table_id, table_name, schema_version)
                 SELECT $1, $2, $3
                 WHERE NOT EXISTS (
                     SELECT 1 FROM ducklake_inlined_data_tables
                     WHERE table_id = $1 AND schema_version = $3)",
            )
            .bind(table_id)
            .bind(&physical_name)
            .bind(schema_version)
            .execute(&mut *tx)
            .await?;
            sqlx::query(
                "INSERT INTO ducklake_table_stats
                     (table_id, record_count, next_row_id, file_size_bytes)
                 VALUES ($1, 0, 0, 0)
                 ON CONFLICT (table_id) DO NOTHING",
            )
            .bind(table_id)
            .execute(&mut *tx)
            .await?;
            let mut row_id: i64 = sqlx::query_scalar(
                "SELECT next_row_id FROM ducklake_table_stats WHERE table_id = $1",
            )
            .bind(table_id)
            .fetch_one(&mut *tx)
            .await?;

            let column_list = columns
                .iter()
                .map(|column| quote_ident(column.name()))
                .collect::<Vec<_>>()
                .join(", ");
            for batch in batches {
                for batch_row in 0..batch.num_rows() {
                    let mut query = QueryBuilder::<Postgres>::new(format!(
                        "INSERT INTO {} (row_id, begin_snapshot, end_snapshot, {}) VALUES (",
                        quote_ident(&physical_name),
                        column_list
                    ));
                    query.push_bind(row_id);
                    query.push(", ").push_bind(snapshot_id);
                    query.push(", NULL");
                    for ((array, sql_type), _column) in
                        batch.columns().iter().zip(&sql_types).zip(columns)
                    {
                        query.push(", ");
                        push_inlined_postgres_value(
                            &mut query,
                            array.as_ref(),
                            batch_row,
                            sql_type,
                        )?;
                    }
                    query.push(')');
                    query.build().execute(&mut *tx).await?;
                    row_id += 1;
                }
            }

            let record_count = i64::try_from(record_count).map_err(|_| {
                crate::DuckLakeError::InvalidConfig(
                    "register_inlined_data: record count exceeds i64".to_string(),
                )
            })?;
            sqlx::query(
                "UPDATE ducklake_table_stats
                 SET next_row_id = next_row_id + $1, record_count = record_count + $2
                 WHERE table_id = $3",
            )
            .bind(record_count)
            .bind(record_count)
            .bind(table_id)
            .execute(&mut *tx)
            .await?;
            // Append to any ledger entries already recorded for this snapshot
            // (created_schema:/created_table:) instead of replacing them, and
            // record a Replace over prior data — Parquet or inlined — as delete
            // + insert, the same semantics the Parquet path records.
            let (replaced_files, replaced_inlined) =
                replaced_storage(&mut tx, table_id, snapshot_id).await?;
            let changes_made = table_storage_changes(
                table_id,
                Some(true),
                mode == WriteMode::Replace && replaced_files,
                mode == WriteMode::Replace && replaced_inlined,
            );
            record_snapshot_changes(&mut tx, snapshot_id, &changes_made, commit_metadata).await?;
            advance_catalog_head(self.catalog_id, snapshot_id, &mut tx).await?;
            tx.commit().await?;
            Ok(CommitIds {
                snapshot_id,
                schema_id,
                table_id,
            })
        })
    }

    fn commit_multi_table(
        &self,
        writes: &[StagedTableWrite],
        commit_metadata: &SnapshotCommitMetadata,
        expected_base_snapshot_id: Option<i64>,
    ) -> Result<MultiTableCommit> {
        if writes.is_empty() {
            return Err(crate::DuckLakeError::InvalidConfig(
                "commit_multi_table requires at least one table stage".to_string(),
            ));
        }
        let table_ids = writes
            .iter()
            .map(|write| write.table_id)
            .collect::<std::collections::HashSet<_>>();
        if table_ids.len() != writes.len() {
            return Err(crate::DuckLakeError::InvalidConfig(
                "commit_multi_table requires one stage per table".to_string(),
            ));
        }

        block_on(async {
            let mut tx = self.pool.begin().await?;
            lock_catalog(self.catalog_id, self.lock_timeout_ms, &mut tx).await?;
            for write in writes {
                validate_staged_table(&mut tx, write).await?;
                assert_table_not_in_other_catalog(self.catalog_id, write.table_id, &mut tx).await?;
            }
            if let Some(expected) = expected_base_snapshot_id {
                for write in writes {
                    detect_replace_conflict(write.table_id, expected, &mut tx).await?;
                }
            }
            let mut had_live_data = Vec::with_capacity(writes.len());
            for write in writes {
                let files: bool = sqlx::query_scalar(
                    "SELECT EXISTS(
                         SELECT 1 FROM ducklake_data_file
                         WHERE table_id = $1 AND end_snapshot IS NULL
                     )",
                )
                .bind(write.table_id)
                .fetch_one(&mut *tx)
                .await?;
                let mut inline_rows = 0i64;
                let inline_tables = sqlx::query(
                    "SELECT table_name FROM ducklake_inlined_data_tables WHERE table_id = $1",
                )
                .bind(write.table_id)
                .fetch_all(&mut *tx)
                .await?;
                for row in inline_tables {
                    let table_name: String = row.try_get(0)?;
                    let sql = format!(
                        "SELECT COUNT(*)::BIGINT FROM {} WHERE end_snapshot IS NULL",
                        quote_ident(&table_name)
                    );
                    inline_rows += sqlx::query_scalar::<_, i64>(AssertSqlSafe(sql))
                        .fetch_one(&mut *tx)
                        .await?;
                }
                had_live_data.push((files, inline_rows > 0));
            }
            let snapshot_id: i64 = sqlx::query_scalar(
                "INSERT INTO ducklake_snapshot (snapshot_time, schema_version)
                 VALUES (NOW(), 0) RETURNING snapshot_id",
            )
            .fetch_one(&mut *tx)
            .await?;
            let mut schema_version = 0;
            let mut schema_changed = false;
            let mut committed_writes = Vec::with_capacity(writes.len());
            let mut tables = Vec::with_capacity(writes.len());
            for write in writes {
                let (schema_id, table_id) = finalize_table_snapshot(
                    self.catalog_id,
                    snapshot_id,
                    &mut schema_version,
                    &mut schema_changed,
                    &write.schema_name,
                    &write.table_name,
                    write.table_id,
                    &write.columns,
                    &write.column_ids,
                    write.mode,
                    write.base_snapshot_id,
                    &mut tx,
                )
                .await?;
                let mut committed_write = write.clone();
                committed_write.table_id = table_id;
                committed_writes.push(committed_write);
                tables.push(CommitIds {
                    snapshot_id,
                    schema_id,
                    table_id,
                });
            }

            for (write, (had_files, had_inlined_rows)) in committed_writes.iter().zip(had_live_data)
            {
                match &write.data {
                    StagedTableData::Files(files) => {
                        commit_files_at_snapshot(&mut tx, snapshot_id, write, files).await?;
                    },
                    StagedTableData::Inlined(batches) => {
                        commit_inlined_at_snapshot(&mut tx, snapshot_id, write, batches).await?;
                    },
                    StagedTableData::None => {},
                }
                apply_positional_deletes_at_snapshot(
                    &mut tx,
                    write.table_id,
                    snapshot_id,
                    write.base_snapshot_id,
                    &write.positional_deletes,
                )
                .await?;
                apply_inlined_deletes_at_snapshot(
                    &mut tx,
                    write.table_id,
                    snapshot_id,
                    write.base_snapshot_id,
                    &write.inlined_deletes,
                )
                .await?;
                let changes_made = staged_table_write_changes(write, had_files, had_inlined_rows);
                record_snapshot_changes(&mut tx, snapshot_id, &changes_made, commit_metadata)
                    .await?;
            }
            advance_catalog_head(self.catalog_id, snapshot_id, &mut tx).await?;
            tx.commit().await?;
            Ok(MultiTableCommit {
                snapshot_id,
                tables,
            })
        })
    }

    fn set_partition_spec(
        &self,
        table_id: i64,
        columns: &[(String, PartitionTransform)],
    ) -> Result<i64> {
        if columns.is_empty() {
            return Err(crate::DuckLakeError::InvalidConfig(
                "set_partition_spec: partition spec must have at least one column; \
                 use reset_partition_spec to remove partitioning"
                    .to_string(),
            ));
        }
        block_on(async {
            let mut tx = self.pool.begin().await?;
            lock_catalog(self.catalog_id, self.lock_timeout_ms, &mut tx).await?;
            assert_table_in_catalog(self.catalog_id, table_id, &mut tx).await?;

            // Resolve each partition-key column NAME to its live column_id.
            let mut column_ids: Vec<i64> = Vec::with_capacity(columns.len());
            for (name, _transform) in columns {
                let column_id: i64 = sqlx::query_scalar(
                    "SELECT column_id FROM ducklake_column
                     WHERE table_id = $1 AND column_name = $2 AND end_snapshot IS NULL
                       AND parent_column IS NULL",
                )
                .bind(table_id)
                .bind(name)
                .fetch_optional(&mut *tx)
                .await?
                .ok_or_else(|| {
                    crate::DuckLakeError::InvalidConfig(format!(
                        "set_partition_spec: no live column '{name}' in table {table_id}"
                    ))
                })?;
                column_ids.push(column_id);
            }

            // New snapshot + advance this catalog's head.
            let snapshot_id: i64 = sqlx::query(
                "INSERT INTO ducklake_snapshot (snapshot_time, schema_version)
                 VALUES (NOW(), 0) RETURNING snapshot_id",
            )
            .fetch_one(&mut *tx)
            .await?
            .try_get(0)?;
            sqlx::query(
                "INSERT INTO ducklake_catalog_snapshot_map (catalog_id, snapshot_id) VALUES ($1, $2)",
            )
            .bind(self.catalog_id)
            .bind(snapshot_id)
            .execute(&mut *tx)
            .await?;

            // Setting a spec is DDL → bump the per-catalog schema_version + ledger.
            let prev_max: i64 = sqlx::query(
                "SELECT COALESCE(MAX(s.schema_version), 0) FROM ducklake_snapshot s
                 JOIN ducklake_catalog_snapshot_map m ON m.snapshot_id = s.snapshot_id
                 WHERE m.catalog_id = $1 AND s.snapshot_id <> $2",
            )
            .bind(self.catalog_id)
            .bind(snapshot_id)
            .fetch_one(&mut *tx)
            .await?
            .try_get(0)?;
            let new_schema_version = prev_max + 1;
            sqlx::query("UPDATE ducklake_snapshot SET schema_version = $1 WHERE snapshot_id = $2")
                .bind(new_schema_version)
                .bind(snapshot_id)
                .execute(&mut *tx)
                .await?;
            sqlx::query(
                "INSERT INTO ducklake_schema_versions (begin_snapshot, schema_version, table_id)
                 VALUES ($1, $2, $3)",
            )
            .bind(snapshot_id)
            .bind(new_schema_version)
            .bind(table_id)
            .execute(&mut *tx)
            .await?;

            // End the currently-live spec generation (if any), then insert the new
            // one (partition_id is IDENTITY → RETURNING) and its per-key columns.
            sqlx::query(
                "UPDATE ducklake_partition_info SET end_snapshot = $1
                 WHERE table_id = $2 AND end_snapshot IS NULL",
            )
            .bind(snapshot_id)
            .bind(table_id)
            .execute(&mut *tx)
            .await?;
            let partition_id: i64 = sqlx::query(
                "INSERT INTO ducklake_partition_info (table_id, begin_snapshot, end_snapshot)
                 VALUES ($1, $2, NULL) RETURNING partition_id",
            )
            .bind(table_id)
            .bind(snapshot_id)
            .fetch_one(&mut *tx)
            .await?
            .try_get(0)?;
            for (key_index, column_id) in column_ids.iter().enumerate() {
                sqlx::query(
                    "INSERT INTO ducklake_partition_column
                         (partition_id, table_id, partition_key_index, column_id, transform)
                     VALUES ($1, $2, $3, $4, $5)",
                )
                .bind(partition_id)
                .bind(table_id)
                .bind(key_index as i64)
                .bind(*column_id)
                .bind(columns[key_index].1.to_catalog_string())
                .execute(&mut *tx)
                .await?;
            }

            record_snapshot_changes(
                &mut tx,
                snapshot_id,
                &format!("altered_table:{table_id}"),
                &SnapshotCommitMetadata::default(),
            )
            .await?;
            tx.commit().await?;
            Ok(snapshot_id)
        })
    }

    fn live_partition_spec(
        &self,
        table_id: i64,
    ) -> Result<Option<crate::partition::PartitionSpec>> {
        block_on(async {
            let rows = sqlx::query(
                "SELECT pi.partition_id, pc.partition_key_index, pc.column_id, pc.transform
                 FROM ducklake_partition_info AS pi
                 JOIN ducklake_partition_column AS pc
                   ON pc.partition_id = pi.partition_id AND pc.table_id = pi.table_id
                 WHERE pi.table_id = $1 AND pi.end_snapshot IS NULL
                 ORDER BY pc.partition_key_index",
            )
            .bind(table_id)
            .fetch_all(&self.pool)
            .await?;
            let parsed = rows
                .iter()
                .map(|row| {
                    Ok::<_, crate::DuckLakeError>((
                        row.try_get::<i64, _>(0)?,
                        i32::try_from(row.try_get::<i64, _>(1)?).unwrap_or(0),
                        row.try_get::<i64, _>(2)?,
                        row.try_get::<String, _>(3)?,
                    ))
                })
                .collect::<Result<Vec<_>>>()?;
            // prune_safe = false: this spec is for laying out a write, never pruning.
            Ok(crate::partition::PartitionSpec::from_rows(parsed, false))
        })
    }

    fn reset_partition_spec(&self, table_id: i64) -> Result<i64> {
        block_on(async {
            let mut tx = self.pool.begin().await?;
            lock_catalog(self.catalog_id, self.lock_timeout_ms, &mut tx).await?;
            assert_table_in_catalog(self.catalog_id, table_id, &mut tx).await?;

            // Nothing to reset → report the current head without a new snapshot.
            let has_live: Option<i32> = sqlx::query_scalar(
                "SELECT 1 FROM ducklake_partition_info
                 WHERE table_id = $1 AND end_snapshot IS NULL LIMIT 1",
            )
            .bind(table_id)
            .fetch_optional(&mut *tx)
            .await?;
            if has_live.is_none() {
                let head: i64 = sqlx::query_scalar(
                    "SELECT COALESCE(MAX(snapshot_id), 0) FROM ducklake_catalog_snapshot_map
                     WHERE catalog_id = $1",
                )
                .bind(self.catalog_id)
                .fetch_one(&mut *tx)
                .await?;
                return Ok(head);
            }

            let snapshot_id: i64 = sqlx::query(
                "INSERT INTO ducklake_snapshot (snapshot_time, schema_version)
                 VALUES (NOW(), 0) RETURNING snapshot_id",
            )
            .fetch_one(&mut *tx)
            .await?
            .try_get(0)?;
            sqlx::query(
                "INSERT INTO ducklake_catalog_snapshot_map (catalog_id, snapshot_id) VALUES ($1, $2)",
            )
            .bind(self.catalog_id)
            .bind(snapshot_id)
            .execute(&mut *tx)
            .await?;
            let prev_max: i64 = sqlx::query(
                "SELECT COALESCE(MAX(s.schema_version), 0) FROM ducklake_snapshot s
                 JOIN ducklake_catalog_snapshot_map m ON m.snapshot_id = s.snapshot_id
                 WHERE m.catalog_id = $1 AND s.snapshot_id <> $2",
            )
            .bind(self.catalog_id)
            .bind(snapshot_id)
            .fetch_one(&mut *tx)
            .await?
            .try_get(0)?;
            let new_schema_version = prev_max + 1;
            sqlx::query("UPDATE ducklake_snapshot SET schema_version = $1 WHERE snapshot_id = $2")
                .bind(new_schema_version)
                .bind(snapshot_id)
                .execute(&mut *tx)
                .await?;
            sqlx::query(
                "INSERT INTO ducklake_schema_versions (begin_snapshot, schema_version, table_id)
                 VALUES ($1, $2, $3)",
            )
            .bind(snapshot_id)
            .bind(new_schema_version)
            .bind(table_id)
            .execute(&mut *tx)
            .await?;
            sqlx::query(
                "UPDATE ducklake_partition_info SET end_snapshot = $1
                 WHERE table_id = $2 AND end_snapshot IS NULL",
            )
            .bind(snapshot_id)
            .bind(table_id)
            .execute(&mut *tx)
            .await?;

            record_snapshot_changes(
                &mut tx,
                snapshot_id,
                &format!("altered_table:{table_id}"),
                &SnapshotCommitMetadata::default(),
            )
            .await?;
            tx.commit().await?;
            Ok(snapshot_id)
        })
    }

    fn live_sort_spec(&self, table_id: i64) -> Result<Option<crate::sort::SortSpec>> {
        block_on(async {
            let rows = sqlx::query(
                "SELECT si.sort_id, se.sort_key_index, se.expression, se.dialect,
                        se.sort_direction, se.null_order
                 FROM ducklake_sort_info AS si
                 JOIN ducklake_sort_expression AS se
                   ON se.sort_id = si.sort_id AND se.table_id = si.table_id
                 WHERE si.table_id = $1 AND si.end_snapshot IS NULL
                 ORDER BY se.sort_key_index",
            )
            .bind(table_id)
            .fetch_all(&self.pool)
            .await?;
            let parsed = rows
                .iter()
                .map(|row| {
                    Ok::<_, crate::DuckLakeError>((
                        row.try_get::<i64, _>(0)?,
                        i32::try_from(row.try_get::<i64, _>(1)?).unwrap_or(0),
                        row.try_get::<String, _>(2)?,
                        row.try_get::<String, _>(3)?,
                        row.try_get::<String, _>(4)?,
                        row.try_get::<String, _>(5)?,
                    ))
                })
                .collect::<Result<Vec<_>>>()?;
            Ok(crate::sort::SortSpec::from_rows(parsed))
        })
    }

    fn set_sort_spec(&self, table_id: i64, fields: &[crate::sort::SortField]) -> Result<i64> {
        if fields.is_empty() {
            return Err(crate::DuckLakeError::InvalidConfig(
                "set_sort_spec: at least one sort key is required (use reset_sort_spec to clear)"
                    .to_string(),
            ));
        }
        block_on(async {
            let mut tx = self.pool.begin().await?;
            lock_catalog(self.catalog_id, self.lock_timeout_ms, &mut tx).await?;
            assert_table_in_catalog(self.catalog_id, table_id, &mut tx).await?;

            let snapshot_id = insert_sort_snapshot(self.catalog_id, &mut tx).await?;

            // Validate every sort key resolves to a live column (v1 supports bare
            // column keys only), so a bad SET fails here rather than silently
            // producing unsorted writes later.
            for field in fields {
                let column = field.column_candidate().ok_or_else(|| {
                    crate::DuckLakeError::InvalidConfig(format!(
                        "set_sort_spec: sort key '{}' is not a bare column; only column \
                         sort keys are supported",
                        field.expression
                    ))
                })?;
                let exists: Option<i64> = sqlx::query_scalar(
                    "SELECT column_id FROM ducklake_column
                     WHERE table_id = $1 AND column_name = $2 AND end_snapshot IS NULL
                       AND parent_column IS NULL",
                )
                .bind(table_id)
                .bind(&column)
                .fetch_optional(&mut *tx)
                .await?;
                if exists.is_none() {
                    return Err(crate::DuckLakeError::InvalidConfig(format!(
                        "set_sort_spec: no live column '{column}' in table {table_id}"
                    )));
                }
            }

            // End the currently-live spec generation (if any), then insert the new
            // one (sort_id is IDENTITY → RETURNING) and its per-key expressions.
            sqlx::query(
                "UPDATE ducklake_sort_info SET end_snapshot = $1
                 WHERE table_id = $2 AND end_snapshot IS NULL",
            )
            .bind(snapshot_id)
            .bind(table_id)
            .execute(&mut *tx)
            .await?;
            let sort_id: i64 = sqlx::query(
                "INSERT INTO ducklake_sort_info (table_id, begin_snapshot, end_snapshot)
                 VALUES ($1, $2, NULL) RETURNING sort_id",
            )
            .bind(table_id)
            .bind(snapshot_id)
            .fetch_one(&mut *tx)
            .await?
            .try_get(0)?;
            for field in fields {
                sqlx::query(
                    "INSERT INTO ducklake_sort_expression
                         (sort_id, table_id, sort_key_index, expression, dialect,
                          sort_direction, null_order)
                     VALUES ($1, $2, $3, $4, $5, $6, $7)",
                )
                .bind(sort_id)
                .bind(table_id)
                .bind(field.sort_key_index as i64)
                .bind(&field.expression)
                .bind(&field.dialect)
                .bind(field.direction.to_catalog_string())
                .bind(field.null_order.to_catalog_string())
                .execute(&mut *tx)
                .await?;
            }

            record_snapshot_changes(
                &mut tx,
                snapshot_id,
                &format!("altered_table:{table_id}"),
                &SnapshotCommitMetadata::default(),
            )
            .await?;
            tx.commit().await?;
            Ok(snapshot_id)
        })
    }

    fn reset_sort_spec(&self, table_id: i64) -> Result<i64> {
        block_on(async {
            let mut tx = self.pool.begin().await?;
            lock_catalog(self.catalog_id, self.lock_timeout_ms, &mut tx).await?;
            assert_table_in_catalog(self.catalog_id, table_id, &mut tx).await?;

            // Nothing to reset → report the current head without a new snapshot.
            let has_live: Option<i32> = sqlx::query_scalar(
                "SELECT 1 FROM ducklake_sort_info
                 WHERE table_id = $1 AND end_snapshot IS NULL LIMIT 1",
            )
            .bind(table_id)
            .fetch_optional(&mut *tx)
            .await?;
            if has_live.is_none() {
                let head: i64 = sqlx::query_scalar(
                    "SELECT COALESCE(MAX(snapshot_id), 0) FROM ducklake_catalog_snapshot_map
                     WHERE catalog_id = $1",
                )
                .bind(self.catalog_id)
                .fetch_one(&mut *tx)
                .await?;
                return Ok(head);
            }

            let snapshot_id = insert_sort_snapshot(self.catalog_id, &mut tx).await?;
            sqlx::query(
                "UPDATE ducklake_sort_info SET end_snapshot = $1
                 WHERE table_id = $2 AND end_snapshot IS NULL",
            )
            .bind(snapshot_id)
            .bind(table_id)
            .execute(&mut *tx)
            .await?;
            record_snapshot_changes(
                &mut tx,
                snapshot_id,
                &format!("altered_table:{table_id}"),
                &SnapshotCommitMetadata::default(),
            )
            .await?;
            tx.commit().await?;
            Ok(snapshot_id)
        })
    }

    fn register_existing_data_file_with_delete(
        &self,
        schema_name: &str,
        table_name: &str,
        columns: &[ColumnDef],
        column_ids: &[i64],
        file: &DataFileInfo,
        delete: Option<&DeleteFileInfo>,
        mode: WriteMode,
    ) -> Result<CommitIds> {
        // This method bypasses begin_write_transaction, so it must do begin's
        // input validation itself. `column_ids[i]` is inserted for `columns[i]`
        // (finalize_snapshot zips them), so a length mismatch would silently drop
        // the trailing columns.
        if columns.is_empty() {
            return Err(crate::DuckLakeError::InvalidConfig(
                "register_existing_data_file requires at least one column".to_string(),
            ));
        }
        let catalog_column_count = catalog_column_defs(columns)?.len();
        if column_ids.len() != catalog_column_count {
            return Err(crate::DuckLakeError::InvalidConfig(format!(
                "register_existing_data_file: column_ids (len {}) must match catalog column nodes (len {})",
                column_ids.len(),
                catalog_column_count
            )));
        }
        // Both ways of getting ownership wrong are refused here, where the caller can
        // still fix them.
        //
        // Naming THIS catalog as owner is a contradiction, and a silent leak: the row
        // would be a reference nobody schedules, while the owner cleanup that would
        // have reclaimed it is this very catalog, holding itself back for as long as
        // its own row stands.
        //
        // A reference with a RELATIVE path is worse — it deletes rather than leaks.
        // `all_reference_paths` hands cleanup the stored path verbatim, which the
        // owner then compares against its own fully-resolved key; a relative spelling
        // can never match, so the owner's cleanup is not deferred and reclaims the
        // object while this row still names it. (The row does not even read back
        // correctly here: a relative path resolves through THIS catalog's schema and
        // table rows, not the owner's.)
        for (kind, owner, is_relative) in [
            ("file", file.owner_catalog_id, file.path_is_relative),
            (
                "delete file",
                delete.and_then(|d| d.owner_catalog_id),
                delete.is_some_and(|d| d.path_is_relative),
            ),
        ] {
            let Some(owner) = owner else {
                continue;
            };
            if owner == self.catalog_id {
                return Err(crate::DuckLakeError::InvalidConfig(format!(
                    "register_existing_data_file: {kind} owner_catalog_id {owner} is this \
                     catalog; a row that references its own catalog's file is never reclaimed \
                     by anyone. Leave it unset for a file this catalog owns."
                )));
            }
            if is_relative {
                return Err(crate::DuckLakeError::InvalidConfig(format!(
                    "register_existing_data_file: {kind} references catalog {owner} but its \
                     path is relative, so it resolves through this catalog's layout and never \
                     matches the object the owner's cleanup defers against. Mark it absolute \
                     with `with_absolute_path()`."
                )));
            }
        }
        block_on(async {
            let mut tx = self.pool.begin().await?;
            lock_catalog(self.catalog_id, self.lock_timeout_ms, &mut tx).await?;

            // No begin_write_transaction: derive the conflict base (catalog head)
            // and a table-id hint (used only if the table doesn't exist yet) here.
            let base_snapshot: i64 = sqlx::query_scalar(
                "SELECT COALESCE(MAX(snapshot_id), 0) FROM ducklake_catalog_snapshot_map
                 WHERE catalog_id = $1",
            )
            .bind(self.catalog_id)
            .fetch_one(&mut *tx)
            .await?;
            let table_id_hint = reserve_ids("ducklake_table", "table_id", 1, &mut tx).await?[0];

            // finalize inserts the columns with the adopted `column_ids`
            // (OVERRIDING SYSTEM VALUE), so the file's field-ids match.
            let (snapshot_id, schema_id, table_id) = finalize_snapshot(
                self.catalog_id,
                schema_name,
                table_name,
                table_id_hint,
                columns,
                column_ids,
                mode,
                base_snapshot,
                &mut tx,
            )
            .await?;

            // Rowids: a fresh range by default, or the source's carried value
            // when the caller supplied one. See `allocate_row_id_start`.
            sqlx::query(
                "INSERT INTO ducklake_table_stats (table_id, record_count, next_row_id, file_size_bytes)
                 VALUES ($1, 0, 0, 0)
                 ON CONFLICT (table_id) DO NOTHING",
            )
            .bind(table_id)
            .execute(&mut *tx)
            .await?;
            let next_row_id: i64 = sqlx::query_scalar(
                "SELECT next_row_id FROM ducklake_table_stats WHERE table_id = $1",
            )
            .bind(table_id)
            .fetch_one(&mut *tx)
            .await?;
            let row_ids = crate::metadata_writer::allocate_row_id_start(next_row_id, file);

            // Partition-spec fence + validation. A promoted file is registered
            // as-is, so the caller is asserting it already holds rows of exactly one
            // partition (official DuckLake's ducklake_add_data_files makes the same
            // assumption, deriving the values from the file's Hive path). We check
            // everything checkable without reading the data: the file agrees with the
            // live generation, and its values fit that generation's keys.
            let live_partition_id: Option<i64> = sqlx::query_scalar(
                "SELECT partition_id FROM ducklake_partition_info
                 WHERE table_id = $1 AND end_snapshot IS NULL",
            )
            .bind(table_id)
            .fetch_optional(&mut *tx)
            .await?;
            // Diagnose the promote-specific case before the shared fence: a caller
            // that supplied no partition assignment for a partitioned table has not
            // lost a race, so the fence's "concurrent SET PARTITIONED BY … retry"
            // wording would send them chasing a problem that does not exist. Tell
            // them what to actually do instead.
            //
            // Deliberately NOT exempting a 0-row file, unlike the shared fence.
            // That exemption exists for the empty-Replace truncate marker a write
            // session emits, which has no promote equivalent — a promoted file is one
            // the caller actually produced. Official agrees: `AddFileToTable` compares
            // the derived value count against the spec's key count with no row-count
            // exception, so an empty file with no assignment is rejected there too.
            if file.partition_id.is_none() && live_partition_id.is_some() {
                return Err(crate::DuckLakeError::InvalidConfig(format!(
                    "cannot promote {} into table {table_id}: the table is partitioned, so a \
                     registered file must declare the single partition its rows belong to. \
                     Attach it with DataFileInfo::with_partition (copy the values from the \
                     source catalog, or derive them from the file's Hive path).",
                    file.path
                )));
            }
            crate::metadata_writer::enforce_partition_fence(table_id, live_partition_id, file)?;
            if let Some(partition_id) = live_partition_id.filter(|_| file.partition_id.is_some()) {
                let key_rows = sqlx::query(
                    "SELECT transform, column_id FROM ducklake_partition_column
                     WHERE table_id = $1 AND partition_id = $2
                     ORDER BY partition_key_index",
                )
                .bind(table_id)
                .bind(partition_id)
                .fetch_all(&mut *tx)
                .await?;
                let mut transforms = Vec::with_capacity(key_rows.len());
                let mut key_column_types = Vec::with_capacity(key_rows.len());
                for row in &key_rows {
                    transforms.push(row.try_get::<String, _>(0)?);
                    // Resolve each key's column_id to the Arrow type of the matching
                    // promoted column, so an `identity` value can be cast-checked the
                    // way official's MapHiveColumn does.
                    let column_id: i64 = row.try_get(1)?;
                    key_column_types.push(
                        column_ids
                            .iter()
                            .position(|id| *id == column_id)
                            .and_then(|index| columns.get(index))
                            .and_then(|column| {
                                crate::types::ducklake_to_arrow_type(column.ducklake_type()).ok()
                            }),
                    );
                }
                crate::metadata_writer::validate_promoted_partition_values(
                    table_id,
                    &transforms,
                    &key_column_types,
                    file,
                )?;
            }

            // `owner_catalog_id` is the one thing registration decides that no
            // other write path does: NULL for a file this catalog owns (including
            // one promoted from outside its own layout), the source's catalog id
            // for a reference. Every reclaim path reads it.
            let data_file_id: i64 = sqlx::query_scalar(
                "INSERT INTO ducklake_data_file
                     (table_id, path, path_is_relative, file_size_bytes,
                      footer_size, record_count, row_id_start, begin_snapshot,
                      owner_catalog_id)
                 VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) RETURNING data_file_id",
            )
            .bind(table_id)
            .bind(&file.path)
            .bind(file.path_is_relative)
            .bind(file.file_size_bytes)
            .bind(file.footer_size)
            .bind(file.record_count)
            .bind(row_ids.stored)
            .bind(snapshot_id)
            .bind(file.owner_catalog_id)
            .fetch_one(&mut *tx)
            .await?;
            // Persist the caller-supplied partition assignment. Without this a
            // promoted file lands with no partition_id and no
            // ducklake_file_partition_value rows, i.e. unprunable and inconsistent
            // with the table's spec.
            insert_partition_metadata(&mut tx, table_id, data_file_id, file).await?;

            // An existing positional delete file travels with its data file: same
            // commit, same snapshot, hanging off the id just assigned. The path is
            // stored as given (a fork marks it absolute, pointing at the source's
            // object), like the data file's.
            if let Some(delete) = delete {
                sqlx::query(
                    "INSERT INTO ducklake_delete_file
                         (data_file_id, table_id, path, path_is_relative, file_size_bytes,
                          footer_size, delete_count, begin_snapshot, owner_catalog_id)
                     VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)",
                )
                .bind(data_file_id)
                .bind(table_id)
                .bind(&delete.path)
                .bind(delete.path_is_relative)
                .bind(delete.file_size_bytes)
                .bind(delete.footer_size)
                .bind(delete.delete_count)
                .bind(snapshot_id)
                .bind(delete.owner_catalog_id)
                .execute(&mut *tx)
                .await?;
            }

            sqlx::query(
                "UPDATE ducklake_table_stats
                 SET next_row_id     = next_row_id + $1,
                     record_count    = record_count + $2,
                     file_size_bytes = file_size_bytes + $3
                 WHERE table_id = $4",
            )
            .bind(row_ids.advance)
            .bind(file.record_count)
            .bind(file.file_size_bytes)
            .bind(table_id)
            .execute(&mut *tx)
            .await?;

            record_snapshot_changes(
                &mut tx,
                snapshot_id,
                &format!("inserted_into_table:{table_id}"),
                &SnapshotCommitMetadata::default(),
            )
            .await?;
            advance_catalog_head(self.catalog_id, snapshot_id, &mut tx).await?;
            tx.commit().await?;
            Ok(CommitIds {
                snapshot_id,
                schema_id,
                table_id,
            })
        })
    }

    #[allow(clippy::too_many_arguments)]
    fn set_delete_file(
        &self,
        table_id: i64,
        _schema_name: &str,
        _table_name: &str,
        _snapshot_id: i64,
        data_file_id: i64,
        expected_prev_delete_file: Option<i64>,
        base_snapshot: i64,
        delete: &DeleteFileInfo,
    ) -> Result<CommitIds> {
        block_on(async {
            // Single atomic commit under the catalog lock: fence on
            // `base_snapshot`, compare-and-swap the currently-live delete file
            // for this data file, allocate the snapshot, retire the prior delete
            // file, insert the new cumulative one, and advance the catalog head
            // LAST — so at most one delete file is ever live per data file and
            // nothing is visible until the head maps the snapshot.
            let mut tx = self.pool.begin().await?;
            lock_catalog(self.catalog_id, self.lock_timeout_ms, &mut tx).await?;
            assert_table_not_in_other_catalog(self.catalog_id, table_id, &mut tx).await?;

            // Target-file fence: the resolved positions are physical row indices
            // in `data_file_id`, and a parquet data file is immutable — so only a
            // concurrent write that RETIRED this file (a Replace/compaction) since
            // `base_snapshot` can invalidate them. An append that adds *other*
            // files does not move this file's rows, and a concurrent delete on
            // THIS file is caught by the compare-and-swap below; neither must
            // block the delete. Abort iff the target is no longer the live file.
            // Select the BIGINT `data_file_id` (not a literal `1`, which Postgres
            // types as INT4 and cannot decode into i64) — we only need existence.
            let target_live: Option<i64> = sqlx::query_scalar(
                "SELECT data_file_id FROM ducklake_data_file
                 WHERE data_file_id = $1 AND end_snapshot IS NULL",
            )
            .bind(data_file_id)
            .fetch_optional(&mut *tx)
            .await?;
            if target_live.is_none() {
                return Err(crate::DuckLakeError::Conflict(format!(
                    "delete targets data file {data_file_id}, which was retired by a \
                     concurrent write since snapshot {base_snapshot}; retry against the \
                     new generation"
                )));
            }

            // Compare-and-swap on the currently-live delete file for this data
            // file (`end_snapshot IS NULL`); a concurrent delete on the same data
            // file makes it differ from what the caller saw.
            let current_prev: Option<i64> = sqlx::query_scalar(
                "SELECT delete_file_id FROM ducklake_delete_file
                 WHERE data_file_id = $1 AND end_snapshot IS NULL",
            )
            .bind(data_file_id)
            .fetch_optional(&mut *tx)
            .await?;
            if current_prev != expected_prev_delete_file {
                return Err(crate::DuckLakeError::Conflict(format!(
                    "delete on data file {data_file_id} conflicts with a concurrent delete \
                     (expected live delete file {expected_prev_delete_file:?}, found \
                     {current_prev:?}); retry against the new generation"
                )));
            }

            // Allocate the snapshot (commit-ordered IDENTITY). A delete is
            // non-DDL, so carry the per-catalog `schema_version` forward.
            let snapshot_id: i64 = sqlx::query(
                "INSERT INTO ducklake_snapshot (snapshot_time, schema_version)
                 VALUES (NOW(), 0) RETURNING snapshot_id",
            )
            .fetch_one(&mut *tx)
            .await?
            .try_get(0)?;
            let prev_max: i64 = sqlx::query(
                "SELECT COALESCE(MAX(s.schema_version), 0) FROM ducklake_snapshot s
                 JOIN ducklake_catalog_snapshot_map m ON m.snapshot_id = s.snapshot_id
                 WHERE m.catalog_id = $1",
            )
            .bind(self.catalog_id)
            .fetch_one(&mut *tx)
            .await?
            .try_get(0)?;
            sqlx::query("UPDATE ducklake_snapshot SET schema_version = $1 WHERE snapshot_id = $2")
                .bind(prev_max.max(1))
                .bind(snapshot_id)
                .execute(&mut *tx)
                .await?;

            // Retire the prior delete file (cumulative: the new file carries all
            // still-deleted positions, so the old one is superseded).
            if let Some(prev) = expected_prev_delete_file {
                sqlx::query(
                    "UPDATE ducklake_delete_file SET end_snapshot = $1
                     WHERE delete_file_id = $2 AND end_snapshot IS NULL",
                )
                .bind(snapshot_id)
                .bind(prev)
                .execute(&mut *tx)
                .await?;
            }

            sqlx::query(
                "INSERT INTO ducklake_delete_file
                     (data_file_id, table_id, path, path_is_relative, file_size_bytes,
                      footer_size, delete_count, begin_snapshot)
                 VALUES ($1, $2, $3, $4, $5, $6, $7, $8)",
            )
            .bind(data_file_id)
            .bind(table_id)
            .bind(&delete.path)
            .bind(delete.path_is_relative)
            .bind(delete.file_size_bytes)
            .bind(delete.footer_size)
            .bind(delete.delete_count)
            .bind(snapshot_id)
            .execute(&mut *tx)
            .await?;

            let schema_id: i64 =
                sqlx::query_scalar("SELECT schema_id FROM ducklake_table WHERE table_id = $1")
                    .bind(table_id)
                    .fetch_one(&mut *tx)
                    .await?;

            record_snapshot_changes(
                &mut tx,
                snapshot_id,
                &format!("deleted_from_table:{table_id}"),
                &SnapshotCommitMetadata::default(),
            )
            .await?;
            // advance_catalog_head MUST be the last write before commit.
            advance_catalog_head(self.catalog_id, snapshot_id, &mut tx).await?;

            tx.commit().await?;
            Ok(CommitIds {
                snapshot_id,
                schema_id,
                table_id,
            })
        })
    }

    #[allow(clippy::too_many_arguments)]
    fn register_data_file_with_deletes(
        &self,
        table_id: i64,
        schema_name: &str,
        table_name: &str,
        snapshot_id: i64,
        file: &DataFileInfo,
        deletes: &[DeleteFileEntry],
        mode: WriteMode,
        base_snapshot: i64,
        columns: &[ColumnDef],
        column_ids: &[i64],
    ) -> Result<CommitIds> {
        self.register_data_file_with_deletes_and_commit_metadata(
            table_id,
            schema_name,
            table_name,
            snapshot_id,
            file,
            deletes,
            mode,
            base_snapshot,
            columns,
            column_ids,
            &SnapshotCommitMetadata::default(),
            None,
        )
    }

    #[allow(clippy::too_many_arguments)]
    fn register_data_file_with_deletes_and_commit_metadata(
        &self,
        table_id: i64,
        schema_name: &str,
        table_name: &str,
        _snapshot_id: i64,
        file: &DataFileInfo,
        deletes: &[DeleteFileEntry],
        mode: WriteMode,
        base_snapshot: i64,
        columns: &[ColumnDef],
        column_ids: &[i64],
        commit_metadata: &SnapshotCommitMetadata,
        expected_base_snapshot_id: Option<i64>,
    ) -> Result<CommitIds> {
        validate_delete_entries(mode, deletes)?;
        block_on(async {
            // One atomic commit for a combined append + positional deletes (an
            // update/upsert). finalize_snapshot writes the snapshot + schema/columns
            // and returns the committed snapshot id; the new data file AND every
            // delete file are stamped with that one id, and advance_catalog_head runs
            // LAST — so the whole mutation becomes visible together, never a
            // half-applied intermediate.
            let mut tx = self.pool.begin().await?;
            lock_catalog(self.catalog_id, self.lock_timeout_ms, &mut tx).await?;
            assert_table_not_in_other_catalog(self.catalog_id, table_id, &mut tx).await?;

            if mode != WriteMode::Replace
                && let Some(expected_base_snapshot_id) = expected_base_snapshot_id
            {
                detect_replace_conflict(table_id, expected_base_snapshot_id, &mut tx).await?;
            }
            if !deletes.is_empty() {
                let file_ids = deletes
                    .iter()
                    .map(|entry| entry.data_file_id)
                    .collect::<Vec<_>>();
                detect_new_inlined_deletes(table_id, base_snapshot, &file_ids, &mut tx).await?;
            }

            let (snapshot_id, schema_id, table_id) = finalize_snapshot(
                self.catalog_id,
                schema_name,
                table_name,
                table_id,
                columns,
                column_ids,
                mode,
                base_snapshot,
                &mut tx,
            )
            .await?;

            // Partition-spec fence: the new row versions are a NEW write, so they
            // must agree with the table's live partition generation exactly as
            // register_data_file's do.
            let live_partition_id: Option<i64> = sqlx::query_scalar(
                "SELECT partition_id FROM ducklake_partition_info
                 WHERE table_id = $1 AND end_snapshot IS NULL",
            )
            .bind(table_id)
            .fetch_optional(&mut *tx)
            .await?;
            crate::metadata_writer::enforce_partition_fence(table_id, live_partition_id, file)?;

            // Register the new data file (the inserted row versions), exactly as
            // register_data_file: seed stats, draw the row-id range, insert, and
            // accumulate. Deletes are accounted at read time (delete_count), so the
            // stats record_count stays gross — do not adjust it for the deletes.
            sqlx::query(
                "INSERT INTO ducklake_table_stats (table_id, record_count, next_row_id, file_size_bytes)
                 VALUES ($1, 0, 0, 0)
                 ON CONFLICT (table_id) DO NOTHING",
            )
            .bind(table_id)
            .execute(&mut *tx)
            .await?;

            let stats_row =
                sqlx::query("SELECT next_row_id FROM ducklake_table_stats WHERE table_id = $1")
                    .bind(table_id)
                    .fetch_one(&mut *tx)
                    .await?;
            let next_row_id: i64 = stats_row.try_get(0)?;
            let row_ids = crate::metadata_writer::allocate_row_id_start(next_row_id, file);

            let inserted = sqlx::query(
                "INSERT INTO ducklake_data_file
                     (table_id, path, path_is_relative, file_size_bytes,
                      footer_size, record_count, row_id_start, begin_snapshot)
                 VALUES ($1, $2, $3, $4, $5, $6, $7, $8) RETURNING data_file_id",
            )
            .bind(table_id)
            .bind(&file.path)
            .bind(file.path_is_relative)
            .bind(file.file_size_bytes)
            .bind(file.footer_size)
            .bind(file.record_count)
            .bind(row_ids.stored)
            .bind(snapshot_id)
            .fetch_one(&mut *tx)
            .await?;
            let data_file_id: i64 = inserted.try_get(0)?;
            // Persist the file's partition assignment, exactly as every other commit
            // path does. Without this an append+delete commit — the update/upsert
            // path — silently drops `partition_id` and the per-key
            // `ducklake_file_partition_value` rows: the commit succeeds, the rows read
            // back correctly, and the file is simply unprunable forever, an island in
            // an otherwise partitioned table. The partition fence does not catch it,
            // because the `DataFileInfo` it validates DOES carry the assignment; only
            // the persistence was missing.
            insert_partition_metadata(&mut tx, table_id, data_file_id, file).await?;
            insert_file_column_stats(&mut tx, table_id, data_file_id, &file.column_stats).await?;
            recompute_table_column_stats(&mut tx, table_id, columns, column_ids).await?;

            sqlx::query(
                "UPDATE ducklake_table_stats
                 SET next_row_id     = next_row_id + $1,
                     record_count    = record_count + $2,
                     file_size_bytes = file_size_bytes + $3
                 WHERE table_id = $4",
            )
            .bind(row_ids.advance)
            .bind(file.record_count)
            .bind(file.file_size_bytes)
            .bind(table_id)
            .execute(&mut *tx)
            .await?;

            // Apply each positional delete with the same fence + compare-and-swap as
            // set_delete_file, stamped with this snapshot. Each entry targets a
            // distinct data file, so there is no intra-transaction CAS contention.
            for entry in deletes {
                let target_live: Option<i64> = sqlx::query_scalar(
                    "SELECT data_file_id FROM ducklake_data_file
                     WHERE data_file_id = $1 AND end_snapshot IS NULL",
                )
                .bind(entry.data_file_id)
                .fetch_optional(&mut *tx)
                .await?;
                if target_live.is_none() {
                    return Err(crate::DuckLakeError::Conflict(format!(
                        "UPDATE/DELETE on data file {} could not commit: the file is no longer \
                         live as of the catalog's current head (retired since snapshot \
                         {base_snapshot}). This happens when another writer committed a \
                         Replace/compaction, OR when an earlier write in THIS session already \
                         advanced the catalog (the catalog pins its snapshot at creation and does \
                         not refresh). Re-open the catalog at the latest snapshot and retry.",
                        entry.data_file_id
                    )));
                }

                let current_prev: Option<i64> = sqlx::query_scalar(
                    "SELECT delete_file_id FROM ducklake_delete_file
                     WHERE data_file_id = $1 AND end_snapshot IS NULL",
                )
                .bind(entry.data_file_id)
                .fetch_optional(&mut *tx)
                .await?;
                if current_prev != entry.expected_prev_delete_file {
                    return Err(crate::DuckLakeError::Conflict(format!(
                        "UPDATE/DELETE on data file {} could not commit: its live delete file \
                         changed from {:?} to {current_prev:?} since snapshot {base_snapshot}. \
                         Another writer committed a delete on this file, OR an earlier \
                         UPDATE/DELETE in THIS session did (the catalog pins its snapshot at \
                         creation and does not refresh). Re-open the catalog at the latest \
                         snapshot and retry.",
                        entry.data_file_id, entry.expected_prev_delete_file
                    )));
                }

                if let Some(prev) = entry.expected_prev_delete_file {
                    sqlx::query(
                        "UPDATE ducklake_delete_file SET end_snapshot = $1
                         WHERE delete_file_id = $2 AND end_snapshot IS NULL",
                    )
                    .bind(snapshot_id)
                    .bind(prev)
                    .execute(&mut *tx)
                    .await?;
                }

                sqlx::query(
                    "INSERT INTO ducklake_delete_file
                         (data_file_id, table_id, path, path_is_relative, file_size_bytes,
                          footer_size, delete_count, begin_snapshot)
                     VALUES ($1, $2, $3, $4, $5, $6, $7, $8)",
                )
                .bind(entry.data_file_id)
                .bind(table_id)
                .bind(&entry.delete.path)
                .bind(entry.delete.path_is_relative)
                .bind(entry.delete.file_size_bytes)
                .bind(entry.delete.footer_size)
                .bind(entry.delete.delete_count)
                .bind(snapshot_id)
                .execute(&mut *tx)
                .await?;
            }

            let replaced_existing_data = replaced_storage(&mut tx, table_id, snapshot_id).await?;
            let changes_made =
                table_write_changes(table_id, mode, !deletes.is_empty(), replaced_existing_data);
            record_snapshot_changes(&mut tx, snapshot_id, &changes_made, commit_metadata).await?;

            // advance_catalog_head MUST be the last write before commit.
            advance_catalog_head(self.catalog_id, snapshot_id, &mut tx).await?;

            tx.commit().await?;
            Ok(CommitIds {
                snapshot_id,
                schema_id,
                table_id,
            })
        })
    }

    #[allow(clippy::too_many_arguments)]
    fn register_data_files_with_deletes(
        &self,
        table_id: i64,
        schema_name: &str,
        table_name: &str,
        snapshot_id: i64,
        files: &[DataFileInfo],
        deletes: &[DeleteFileEntry],
        mode: WriteMode,
        base_snapshot: i64,
        columns: &[ColumnDef],
        column_ids: &[i64],
    ) -> Result<CommitIds> {
        self.register_data_files_with_deletes_and_commit_metadata(
            table_id,
            schema_name,
            table_name,
            snapshot_id,
            files,
            deletes,
            mode,
            base_snapshot,
            columns,
            column_ids,
            &SnapshotCommitMetadata::default(),
            None,
        )
    }

    #[allow(clippy::too_many_arguments)]
    fn register_data_files_with_deletes_and_commit_metadata(
        &self,
        table_id: i64,
        schema_name: &str,
        table_name: &str,
        _snapshot_id: i64,
        files: &[DataFileInfo],
        deletes: &[DeleteFileEntry],
        mode: WriteMode,
        base_snapshot: i64,
        columns: &[ColumnDef],
        column_ids: &[i64],
        commit_metadata: &SnapshotCommitMetadata,
        expected_base_snapshot_id: Option<i64>,
    ) -> Result<CommitIds> {
        if files.is_empty() {
            return Err(crate::DuckLakeError::InvalidConfig(
                "register_data_files_with_deletes: files must be non-empty".to_string(),
            ));
        }
        validate_delete_entries(mode, deletes)?;
        block_on(async {
            // One atomic commit for a combined N-file append + positional deletes
            // (a keyed mutation whose new row versions span several partitions or
            // rolled files). finalize_snapshot writes the snapshot + schema/columns
            // and returns the committed snapshot id; EVERY appended data file AND
            // every delete file are stamped with that one id, and
            // advance_catalog_head runs LAST — so the whole mutation becomes visible
            // together, never a half-applied intermediate.
            let mut tx = self.pool.begin().await?;
            lock_catalog(self.catalog_id, self.lock_timeout_ms, &mut tx).await?;
            assert_table_not_in_other_catalog(self.catalog_id, table_id, &mut tx).await?;

            if mode != WriteMode::Replace
                && let Some(expected_base_snapshot_id) = expected_base_snapshot_id
            {
                detect_replace_conflict(table_id, expected_base_snapshot_id, &mut tx).await?;
            }
            if !deletes.is_empty() {
                let file_ids = deletes
                    .iter()
                    .map(|entry| entry.data_file_id)
                    .collect::<Vec<_>>();
                detect_new_inlined_deletes(table_id, base_snapshot, &file_ids, &mut tx).await?;
            }

            let (snapshot_id, schema_id, table_id) = finalize_snapshot(
                self.catalog_id,
                schema_name,
                table_name,
                table_id,
                columns,
                column_ids,
                mode,
                base_snapshot,
                &mut tx,
            )
            .await?;

            // Partition-spec fence (both directions, every file): the new row
            // versions are a NEW write, so each must agree with the table's live
            // partition generation exactly as register_data_files' do.
            let live_partition_id: Option<i64> = sqlx::query_scalar(
                "SELECT partition_id FROM ducklake_partition_info
                 WHERE table_id = $1 AND end_snapshot IS NULL",
            )
            .bind(table_id)
            .fetch_optional(&mut *tx)
            .await?;
            for file in files {
                crate::metadata_writer::enforce_partition_fence(table_id, live_partition_id, file)?;
            }

            // Register the new data files (the inserted row versions), exactly as
            // register_data_files: seed stats, draw a distinct row-id range per
            // file, insert, and accumulate. Deletes are accounted at read time
            // (delete_count), so the stats record_count stays gross — do not adjust
            // it for the deletes.
            sqlx::query(
                "INSERT INTO ducklake_table_stats (table_id, record_count, next_row_id, file_size_bytes)
                 VALUES ($1, 0, 0, 0)
                 ON CONFLICT (table_id) DO NOTHING",
            )
            .bind(table_id)
            .execute(&mut *tx)
            .await?;

            let mut next_row_id: i64 =
                sqlx::query("SELECT next_row_id FROM ducklake_table_stats WHERE table_id = $1")
                    .bind(table_id)
                    .fetch_one(&mut *tx)
                    .await?
                    .try_get(0)?;
            let mut total_advance: i64 = 0;
            let mut total_records: i64 = 0;
            let mut total_bytes: i64 = 0;
            for file in files {
                let row_ids = crate::metadata_writer::allocate_row_id_start(next_row_id, file);
                let inserted = sqlx::query(
                    "INSERT INTO ducklake_data_file
                         (table_id, path, path_is_relative, file_size_bytes,
                          footer_size, record_count, row_id_start, begin_snapshot)
                     VALUES ($1, $2, $3, $4, $5, $6, $7, $8) RETURNING data_file_id",
                )
                .bind(table_id)
                .bind(&file.path)
                .bind(file.path_is_relative)
                .bind(file.file_size_bytes)
                .bind(file.footer_size)
                .bind(file.record_count)
                .bind(row_ids.stored)
                .bind(snapshot_id)
                .fetch_one(&mut *tx)
                .await?;
                let data_file_id: i64 = inserted.try_get(0)?;
                insert_partition_metadata(&mut tx, table_id, data_file_id, file).await?;
                insert_file_column_stats(&mut tx, table_id, data_file_id, &file.column_stats)
                    .await?;
                next_row_id += row_ids.advance;
                total_advance += row_ids.advance;
                total_records += file.record_count;
                total_bytes += file.file_size_bytes;
            }
            recompute_table_column_stats(&mut tx, table_id, columns, column_ids).await?;

            sqlx::query(
                "UPDATE ducklake_table_stats
                 SET next_row_id     = next_row_id + $1,
                     record_count    = record_count + $2,
                     file_size_bytes = file_size_bytes + $3
                 WHERE table_id = $4",
            )
            .bind(total_advance)
            .bind(total_records)
            .bind(total_bytes)
            .bind(table_id)
            .execute(&mut *tx)
            .await?;

            // Apply each positional delete with the same fence + compare-and-swap as
            // set_delete_file, stamped with this snapshot. Each entry targets a
            // distinct data file, so there is no intra-transaction CAS contention.
            for entry in deletes {
                let target_live: Option<i64> = sqlx::query_scalar(
                    "SELECT data_file_id FROM ducklake_data_file
                     WHERE data_file_id = $1 AND end_snapshot IS NULL",
                )
                .bind(entry.data_file_id)
                .fetch_optional(&mut *tx)
                .await?;
                if target_live.is_none() {
                    return Err(crate::DuckLakeError::Conflict(format!(
                        "UPDATE/DELETE on data file {} could not commit: the file is no longer \
                         live as of the catalog's current head (retired since snapshot \
                         {base_snapshot}). This happens when another writer committed a \
                         Replace/compaction, OR when an earlier write in THIS session already \
                         advanced the catalog (the catalog pins its snapshot at creation and does \
                         not refresh). Re-open the catalog at the latest snapshot and retry.",
                        entry.data_file_id
                    )));
                }

                let current_prev: Option<i64> = sqlx::query_scalar(
                    "SELECT delete_file_id FROM ducklake_delete_file
                     WHERE data_file_id = $1 AND end_snapshot IS NULL",
                )
                .bind(entry.data_file_id)
                .fetch_optional(&mut *tx)
                .await?;
                if current_prev != entry.expected_prev_delete_file {
                    return Err(crate::DuckLakeError::Conflict(format!(
                        "UPDATE/DELETE on data file {} could not commit: its live delete file \
                         changed from {:?} to {current_prev:?} since snapshot {base_snapshot}. \
                         Another writer committed a delete on this file, OR an earlier \
                         UPDATE/DELETE in THIS session did (the catalog pins its snapshot at \
                         creation and does not refresh). Re-open the catalog at the latest \
                         snapshot and retry.",
                        entry.data_file_id, entry.expected_prev_delete_file
                    )));
                }

                if let Some(prev) = entry.expected_prev_delete_file {
                    sqlx::query(
                        "UPDATE ducklake_delete_file SET end_snapshot = $1
                         WHERE delete_file_id = $2 AND end_snapshot IS NULL",
                    )
                    .bind(snapshot_id)
                    .bind(prev)
                    .execute(&mut *tx)
                    .await?;
                }

                sqlx::query(
                    "INSERT INTO ducklake_delete_file
                         (data_file_id, table_id, path, path_is_relative, file_size_bytes,
                          footer_size, delete_count, begin_snapshot)
                     VALUES ($1, $2, $3, $4, $5, $6, $7, $8)",
                )
                .bind(entry.data_file_id)
                .bind(table_id)
                .bind(&entry.delete.path)
                .bind(entry.delete.path_is_relative)
                .bind(entry.delete.file_size_bytes)
                .bind(entry.delete.footer_size)
                .bind(entry.delete.delete_count)
                .bind(snapshot_id)
                .execute(&mut *tx)
                .await?;
            }

            let replaced_existing_data = replaced_storage(&mut tx, table_id, snapshot_id).await?;
            let changes_made =
                table_write_changes(table_id, mode, !deletes.is_empty(), replaced_existing_data);
            record_snapshot_changes(&mut tx, snapshot_id, &changes_made, commit_metadata).await?;

            // advance_catalog_head MUST be the last write before commit.
            advance_catalog_head(self.catalog_id, snapshot_id, &mut tx).await?;

            tx.commit().await?;
            Ok(CommitIds {
                snapshot_id,
                schema_id,
                table_id,
            })
        })
    }

    fn commit_positional_deletes(
        &self,
        table_id: i64,
        _schema_name: &str,
        _table_name: &str,
        base_snapshot: i64,
        deletes: &[DeleteFileEntry],
    ) -> Result<CommitIds> {
        if deletes.is_empty() {
            return Err(crate::DuckLakeError::InvalidConfig(
                "commit_positional_deletes requires at least one delete entry".to_string(),
            ));
        }
        // A positional delete never retires the data files it targets, so it is
        // Append-semantics for validation (also enforces distinct data files).
        validate_delete_entries(WriteMode::Append, deletes)?;
        block_on(async {
            // Single atomic commit under the catalog lock for an N-file positional
            // DELETE with no append: fence + compare-and-swap + retire + insert per
            // entry, all stamped with one snapshot; advance_catalog_head LAST so
            // the whole multi-file delete becomes visible together.
            let mut tx = self.pool.begin().await?;
            lock_catalog(self.catalog_id, self.lock_timeout_ms, &mut tx).await?;
            assert_table_not_in_other_catalog(self.catalog_id, table_id, &mut tx).await?;
            let file_ids = deletes
                .iter()
                .map(|entry| entry.data_file_id)
                .collect::<Vec<_>>();
            detect_new_inlined_deletes(table_id, base_snapshot, &file_ids, &mut tx).await?;

            // Allocate the snapshot (commit-ordered IDENTITY). A delete is
            // non-DDL, so carry the per-catalog schema_version forward.
            let snapshot_id: i64 = sqlx::query(
                "INSERT INTO ducklake_snapshot (snapshot_time, schema_version)
                 VALUES (NOW(), 0) RETURNING snapshot_id",
            )
            .fetch_one(&mut *tx)
            .await?
            .try_get(0)?;
            let prev_max: i64 = sqlx::query(
                "SELECT COALESCE(MAX(s.schema_version), 0) FROM ducklake_snapshot s
                 JOIN ducklake_catalog_snapshot_map m ON m.snapshot_id = s.snapshot_id
                 WHERE m.catalog_id = $1",
            )
            .bind(self.catalog_id)
            .fetch_one(&mut *tx)
            .await?
            .try_get(0)?;
            sqlx::query("UPDATE ducklake_snapshot SET schema_version = $1 WHERE snapshot_id = $2")
                .bind(prev_max.max(1))
                .bind(snapshot_id)
                .execute(&mut *tx)
                .await?;

            for entry in deletes {
                // Target-file fence: abort iff the data file is no longer live.
                // Select the BIGINT data_file_id (not literal 1, which Postgres
                // types as INT4 and cannot decode into i64) — existence only.
                let target_live: Option<i64> = sqlx::query_scalar(
                    "SELECT data_file_id FROM ducklake_data_file
                     WHERE data_file_id = $1 AND end_snapshot IS NULL",
                )
                .bind(entry.data_file_id)
                .fetch_optional(&mut *tx)
                .await?;
                if target_live.is_none() {
                    return Err(crate::DuckLakeError::Conflict(format!(
                        "DELETE on data file {} could not commit: the file is no longer live as \
                         of the catalog's current head (retired since snapshot {base_snapshot}). \
                         This happens when another writer committed a Replace/compaction, OR when \
                         an earlier write in THIS session already advanced the catalog (the \
                         catalog pins its snapshot at creation and does not refresh). Re-open the \
                         catalog at the latest snapshot and retry.",
                        entry.data_file_id
                    )));
                }

                // Compare-and-swap on the currently-live delete file.
                let current_prev: Option<i64> = sqlx::query_scalar(
                    "SELECT delete_file_id FROM ducklake_delete_file
                     WHERE data_file_id = $1 AND end_snapshot IS NULL",
                )
                .bind(entry.data_file_id)
                .fetch_optional(&mut *tx)
                .await?;
                if current_prev != entry.expected_prev_delete_file {
                    return Err(crate::DuckLakeError::Conflict(format!(
                        "DELETE on data file {} could not commit: its live delete file changed \
                         from {:?} to {current_prev:?} since snapshot {base_snapshot}. Another \
                         writer committed a delete on this file, OR an earlier DELETE in THIS \
                         session did (the catalog pins its snapshot at creation and does not \
                         refresh). Re-open the catalog at the latest snapshot and retry.",
                        entry.data_file_id, entry.expected_prev_delete_file
                    )));
                }

                if let Some(prev) = entry.expected_prev_delete_file {
                    sqlx::query(
                        "UPDATE ducklake_delete_file SET end_snapshot = $1
                         WHERE delete_file_id = $2 AND end_snapshot IS NULL",
                    )
                    .bind(snapshot_id)
                    .bind(prev)
                    .execute(&mut *tx)
                    .await?;
                }

                sqlx::query(
                    "INSERT INTO ducklake_delete_file
                         (data_file_id, table_id, path, path_is_relative, file_size_bytes,
                          footer_size, delete_count, begin_snapshot)
                     VALUES ($1, $2, $3, $4, $5, $6, $7, $8)",
                )
                .bind(entry.data_file_id)
                .bind(table_id)
                .bind(&entry.delete.path)
                .bind(entry.delete.path_is_relative)
                .bind(entry.delete.file_size_bytes)
                .bind(entry.delete.footer_size)
                .bind(entry.delete.delete_count)
                .bind(snapshot_id)
                .execute(&mut *tx)
                .await?;
            }

            let schema_id: i64 =
                sqlx::query_scalar("SELECT schema_id FROM ducklake_table WHERE table_id = $1")
                    .bind(table_id)
                    .fetch_one(&mut *tx)
                    .await?;

            record_snapshot_changes(
                &mut tx,
                snapshot_id,
                &format!("deleted_from_table:{table_id}"),
                &SnapshotCommitMetadata::default(),
            )
            .await?;
            // advance_catalog_head MUST be the last write before commit.
            advance_catalog_head(self.catalog_id, snapshot_id, &mut tx).await?;

            tx.commit().await?;
            Ok(CommitIds {
                snapshot_id,
                schema_id,
                table_id,
            })
        })
    }

    fn commit_inlined_deletes(
        &self,
        table_id: i64,
        _schema_name: &str,
        _table_name: &str,
        base_snapshot: i64,
        rows: &[InlinedRowRef],
    ) -> Result<CommitIds> {
        if rows.is_empty() {
            return Err(crate::DuckLakeError::InvalidConfig(
                "commit_inlined_deletes requires at least one row".to_string(),
            ));
        }
        block_on(async {
            let mut tx = self.pool.begin().await?;
            lock_catalog(self.catalog_id, self.lock_timeout_ms, &mut tx).await?;
            assert_table_in_catalog(self.catalog_id, table_id, &mut tx).await?;
            let snapshot_id: i64 = sqlx::query(
                "INSERT INTO ducklake_snapshot (snapshot_time, schema_version)
                 VALUES (NOW(), 0) RETURNING snapshot_id",
            )
            .fetch_one(&mut *tx)
            .await?
            .try_get(0)?;
            let schema_version: i64 = sqlx::query_scalar(
                "SELECT COALESCE(MAX(s.schema_version), 0) FROM ducklake_snapshot s
                 JOIN ducklake_catalog_snapshot_map m ON m.snapshot_id = s.snapshot_id
                 WHERE m.catalog_id = $1",
            )
            .bind(self.catalog_id)
            .fetch_one(&mut *tx)
            .await?;
            sqlx::query("UPDATE ducklake_snapshot SET schema_version = $1 WHERE snapshot_id = $2")
                .bind(schema_version.max(1))
                .bind(snapshot_id)
                .execute(&mut *tx)
                .await?;
            apply_inlined_deletes_at_snapshot(&mut tx, table_id, snapshot_id, base_snapshot, rows)
                .await?;
            sqlx::query(
                "INSERT INTO ducklake_snapshot_changes (snapshot_id, changes_made)
                 VALUES ($1, $2)",
            )
            .bind(snapshot_id)
            .bind(format!("inlined_delete:{table_id}"))
            .execute(&mut *tx)
            .await?;
            let schema_id: i64 =
                sqlx::query_scalar("SELECT schema_id FROM ducklake_table WHERE table_id = $1")
                    .bind(table_id)
                    .fetch_one(&mut *tx)
                    .await?;
            advance_catalog_head(self.catalog_id, snapshot_id, &mut tx).await?;
            tx.commit().await?;
            Ok(CommitIds {
                snapshot_id,
                schema_id,
                table_id,
            })
        })
    }

    fn commit_deletes(
        &self,
        table_id: i64,
        schema_name: &str,
        table_name: &str,
        base_snapshot: i64,
        positional: &[DeleteFileEntry],
        inlined: &[InlinedRowRef],
    ) -> Result<CommitIds> {
        if inlined.is_empty() {
            return self.commit_positional_deletes(
                table_id,
                schema_name,
                table_name,
                base_snapshot,
                positional,
            );
        }
        if positional.is_empty() {
            return self.commit_inlined_deletes(
                table_id,
                schema_name,
                table_name,
                base_snapshot,
                inlined,
            );
        }
        validate_delete_entries(WriteMode::Append, positional)?;
        block_on(async {
            let mut tx = self.pool.begin().await?;
            lock_catalog(self.catalog_id, self.lock_timeout_ms, &mut tx).await?;
            assert_table_in_catalog(self.catalog_id, table_id, &mut tx).await?;
            let snapshot_id: i64 = sqlx::query(
                "INSERT INTO ducklake_snapshot (snapshot_time, schema_version)
                 VALUES (NOW(), 0) RETURNING snapshot_id",
            )
            .fetch_one(&mut *tx)
            .await?
            .try_get(0)?;
            let schema_version: i64 = sqlx::query_scalar(
                "SELECT COALESCE(MAX(s.schema_version), 0) FROM ducklake_snapshot s
                 JOIN ducklake_catalog_snapshot_map m ON m.snapshot_id = s.snapshot_id
                 WHERE m.catalog_id = $1",
            )
            .bind(self.catalog_id)
            .fetch_one(&mut *tx)
            .await?;
            sqlx::query("UPDATE ducklake_snapshot SET schema_version = $1 WHERE snapshot_id = $2")
                .bind(schema_version.max(1))
                .bind(snapshot_id)
                .execute(&mut *tx)
                .await?;

            for entry in positional {
                let target_live: Option<i64> = sqlx::query_scalar(
                    "SELECT data_file_id FROM ducklake_data_file
                     WHERE data_file_id = $1 AND end_snapshot IS NULL",
                )
                .bind(entry.data_file_id)
                .fetch_optional(&mut *tx)
                .await?;
                if target_live.is_none() {
                    return Err(crate::DuckLakeError::Conflict(format!(
                        "DELETE on data file {} could not commit: the file is no longer live \
                         since snapshot {base_snapshot}",
                        entry.data_file_id
                    )));
                }
                let current_previous: Option<i64> = sqlx::query_scalar(
                    "SELECT delete_file_id FROM ducklake_delete_file
                     WHERE data_file_id = $1 AND end_snapshot IS NULL",
                )
                .bind(entry.data_file_id)
                .fetch_optional(&mut *tx)
                .await?;
                if current_previous != entry.expected_prev_delete_file {
                    return Err(crate::DuckLakeError::Conflict(format!(
                        "DELETE on data file {} could not commit: its live delete file changed \
                         from {:?} to {current_previous:?} since snapshot {base_snapshot}",
                        entry.data_file_id, entry.expected_prev_delete_file
                    )));
                }
                if let Some(previous) = entry.expected_prev_delete_file {
                    sqlx::query(
                        "UPDATE ducklake_delete_file SET end_snapshot = $1
                         WHERE delete_file_id = $2 AND end_snapshot IS NULL",
                    )
                    .bind(snapshot_id)
                    .bind(previous)
                    .execute(&mut *tx)
                    .await?;
                }
                sqlx::query(
                    "INSERT INTO ducklake_delete_file
                         (data_file_id, table_id, path, path_is_relative, file_size_bytes,
                          footer_size, delete_count, begin_snapshot)
                     VALUES ($1, $2, $3, $4, $5, $6, $7, $8)",
                )
                .bind(entry.data_file_id)
                .bind(table_id)
                .bind(&entry.delete.path)
                .bind(entry.delete.path_is_relative)
                .bind(entry.delete.file_size_bytes)
                .bind(entry.delete.footer_size)
                .bind(entry.delete.delete_count)
                .bind(snapshot_id)
                .execute(&mut *tx)
                .await?;
            }

            apply_inlined_deletes_at_snapshot(
                &mut tx,
                table_id,
                snapshot_id,
                base_snapshot,
                inlined,
            )
            .await?;
            sqlx::query(
                "INSERT INTO ducklake_snapshot_changes (snapshot_id, changes_made)
                 VALUES ($1, $2)",
            )
            .bind(snapshot_id)
            .bind(table_storage_changes(
                table_id,
                None,
                !positional.is_empty(),
                !inlined.is_empty(),
            ))
            .execute(&mut *tx)
            .await?;
            let schema_id: i64 =
                sqlx::query_scalar("SELECT schema_id FROM ducklake_table WHERE table_id = $1")
                    .bind(table_id)
                    .fetch_one(&mut *tx)
                    .await?;
            advance_catalog_head(self.catalog_id, snapshot_id, &mut tx).await?;
            tx.commit().await?;
            Ok(CommitIds {
                snapshot_id,
                schema_id,
                table_id,
            })
        })
    }

    fn commit_compaction(
        &self,
        table_id: i64,
        base_snapshot: i64,
        sources: &[crate::metadata_writer::CompactionSourceFile],
        outputs: &[crate::metadata_writer::CompactionOutputFile],
        retirement: crate::metadata_writer::SourceRetirement,
    ) -> Result<CommitIds> {
        use crate::metadata_writer::SourceRetirement;
        if sources.is_empty() {
            return Err(crate::DuckLakeError::InvalidConfig(
                "commit_compaction requires at least one source file".to_string(),
            ));
        }
        block_on(async {
            // One atomic commit under the catalog lock. Mirrors
            // commit_positional_deletes' shape (allocate snapshot, carry
            // schema_version forward, fence + CAS), plus: schedule + retire the
            // sources (and their delete files), register the outputs, recompute
            // stats, record the change ledger, and advance_catalog_head LAST.
            let mut tx = self.pool.begin().await?;
            lock_catalog(self.catalog_id, self.lock_timeout_ms, &mut tx).await?;
            assert_table_not_in_other_catalog(self.catalog_id, table_id, &mut tx).await?;
            let source_data_ids = sources
                .iter()
                .map(|source| source.data_file_id)
                .collect::<Vec<_>>();
            detect_new_inlined_deletes(table_id, base_snapshot, &source_data_ids, &mut tx).await?;

            let snapshot_id: i64 = sqlx::query(
                "INSERT INTO ducklake_snapshot (snapshot_time, schema_version)
                 VALUES (NOW(), 0) RETURNING snapshot_id",
            )
            .fetch_one(&mut *tx)
            .await?
            .try_get(0)?;
            let prev_max: i64 = sqlx::query(
                "SELECT COALESCE(MAX(s.schema_version), 0) FROM ducklake_snapshot s
                 JOIN ducklake_catalog_snapshot_map m ON m.snapshot_id = s.snapshot_id
                 WHERE m.catalog_id = $1",
            )
            .bind(self.catalog_id)
            .fetch_one(&mut *tx)
            .await?
            .try_get(0)?;
            sqlx::query("UPDATE ducklake_snapshot SET schema_version = $1 WHERE snapshot_id = $2")
                .bind(prev_max.max(1))
                .bind(snapshot_id)
                .execute(&mut *tx)
                .await?;

            // Fence + compare-and-swap per source (see commit_compaction on the
            // SQLite writer for the rationale): abort — never resurrect rows —
            // if a source was retired or its live delete file changed since read.
            for src in sources {
                let target_live: Option<i64> = sqlx::query_scalar(
                    "SELECT data_file_id FROM ducklake_data_file
                     WHERE data_file_id = $1 AND table_id = $2 AND end_snapshot IS NULL",
                )
                .bind(src.data_file_id)
                .bind(table_id)
                .fetch_optional(&mut *tx)
                .await?;
                if target_live.is_none() {
                    return Err(crate::DuckLakeError::Conflict(format!(
                        "compaction of table {table_id} could not commit: source data file {} is \
                         no longer live (retired by a concurrent Replace/compaction since snapshot \
                         {base_snapshot}). Re-open the catalog at the latest snapshot and re-plan.",
                        src.data_file_id
                    )));
                }
                let current_delete: Option<i64> = sqlx::query_scalar(
                    "SELECT delete_file_id FROM ducklake_delete_file
                     WHERE data_file_id = $1 AND end_snapshot IS NULL",
                )
                .bind(src.data_file_id)
                .fetch_optional(&mut *tx)
                .await?;
                if current_delete != src.delete_file_id {
                    return Err(crate::DuckLakeError::Conflict(format!(
                        "compaction of table {table_id} could not commit: the live delete file of \
                         source data file {} changed from {:?} to {current_delete:?} since snapshot \
                         {base_snapshot} (a concurrent DELETE/UPDATE). Re-open the catalog at the \
                         latest snapshot and re-plan.",
                        src.data_file_id, src.delete_file_id
                    )));
                }
            }

            match retirement {
                SourceRetirement::Remove => {
                    // Merge: the partial output serves every snapshot the sources
                    // did, so schedule their physical files (resolving paths as the
                    // multicatalog expire path does) and REMOVE their catalog rows.
                    let dead_data = sqlx::query(AssertSqlSafe(format!(
                        "SELECT df.data_file_id, {COMPACTION_RESOLVED_PATH} AS resolved_path,
                                {COMPACTION_REL_FLAG} AS rel,
                                df.owner_catalog_id IS NULL AS owned
                         FROM ducklake_data_file df
                         JOIN ducklake_table t ON t.table_id = df.table_id
                         JOIN ducklake_schema s ON s.schema_id = t.schema_id
                         WHERE df.data_file_id = ANY($1)"
                    )))
                    .bind(&source_data_ids)
                    .fetch_all(&mut *tx)
                    .await?;
                    schedule_compaction_files(&mut tx, self.catalog_id, dead_data).await?;

                    let dead_del = sqlx::query(AssertSqlSafe(format!(
                        "SELECT df.delete_file_id, {COMPACTION_RESOLVED_PATH} AS resolved_path,
                                {COMPACTION_REL_FLAG} AS rel,
                                df.owner_catalog_id IS NULL AS owned
                         FROM ducklake_delete_file df
                         JOIN ducklake_table t ON t.table_id = df.table_id
                         JOIN ducklake_schema s ON s.schema_id = t.schema_id
                         WHERE df.data_file_id = ANY($1)"
                    )))
                    .bind(&source_data_ids)
                    .fetch_all(&mut *tx)
                    .await?;
                    schedule_compaction_files(&mut tx, self.catalog_id, dead_del).await?;

                    sqlx::query("DELETE FROM ducklake_delete_file WHERE data_file_id = ANY($1)")
                        .bind(&source_data_ids)
                        .execute(&mut *tx)
                        .await?;
                    sqlx::query("DELETE FROM ducklake_data_file WHERE data_file_id = ANY($1)")
                        .bind(&source_data_ids)
                        .execute(&mut *tx)
                        .await?;
                    // Hard-delete the removed sources' per-column stats too, as
                    // official DuckLake does on merge (otherwise they orphan).
                    sqlx::query(
                        "DELETE FROM ducklake_file_column_stats WHERE data_file_id = ANY($1)",
                    )
                    .bind(&source_data_ids)
                    .execute(&mut *tx)
                    .await?;
                    // Likewise the removed sources' per-file partition values
                    // (mirrors the SQLite compaction path).
                    sqlx::query(
                        "DELETE FROM ducklake_file_partition_value WHERE data_file_id = ANY($1)",
                    )
                    .bind(&source_data_ids)
                    .execute(&mut *tx)
                    .await?;
                },
                SourceRetirement::Retire => {
                    // Rewrite: the sources still serve time travel to pre-compaction
                    // snapshots, so retire them (end_snapshot) but do NOT schedule
                    // them; expire_snapshots reclaims them once their snapshots are
                    // gone.
                    sqlx::query(
                        "UPDATE ducklake_data_file SET end_snapshot = $1
                         WHERE data_file_id = ANY($2) AND end_snapshot IS NULL",
                    )
                    .bind(snapshot_id)
                    .bind(&source_data_ids)
                    .execute(&mut *tx)
                    .await?;
                    sqlx::query(
                        "UPDATE ducklake_delete_file SET end_snapshot = $1
                         WHERE data_file_id = ANY($2) AND end_snapshot IS NULL",
                    )
                    .bind(snapshot_id)
                    .bind(&source_data_ids)
                    .execute(&mut *tx)
                    .await?;
                },
            }

            // Register each rewritten output. begin_snapshot = the file's min
            // origin snapshot for a merged partial file (so historical reads see
            // it), else this compaction snapshot; row_id_start NULL (rowids come
            // from the embedded column); partial_max marks a merged partial file.
            for out in outputs {
                let begin = out.begin_snapshot.unwrap_or(snapshot_id);
                let inserted = sqlx::query(
                    "INSERT INTO ducklake_data_file
                         (table_id, path, path_is_relative, file_size_bytes,
                          footer_size, record_count, row_id_start, begin_snapshot, partial_max)
                     VALUES ($1, $2, $3, $4, $5, $6, NULL, $7, $8) RETURNING data_file_id",
                )
                .bind(table_id)
                .bind(&out.file.path)
                .bind(out.file.path_is_relative)
                .bind(out.file.file_size_bytes)
                .bind(out.file.footer_size)
                .bind(out.file.record_count)
                .bind(begin)
                .bind(out.partial_max)
                .fetch_one(&mut *tx)
                .await?;
                // Persist the compacted file's harvested zone maps (same tx),
                // mirroring official DuckLake's stats-recording for outputs.
                let data_file_id: i64 = inserted.try_get(0)?;
                insert_file_column_stats(&mut tx, table_id, data_file_id, &out.file.column_stats)
                    .await?;
                // Carry the output's partition assignment over from its sources.
                // Every file in a compaction group shares one partition, so the
                // output belongs to exactly that partition; without this the
                // merged file drops out of its partition and partition-value
                // pruning is lost for good.
                insert_partition_metadata(&mut tx, table_id, data_file_id, &out.file).await?;
            }

            // Recompute the visible stat totals from the surviving files (see the
            // SQLite writer for why this is correct for both merge and rewrite).
            // next_row_id is deliberately not advanced (no new logical rows).
            sqlx::query(
                "INSERT INTO ducklake_table_stats (table_id, record_count, next_row_id, file_size_bytes)
                 VALUES ($1, 0, 0, 0) ON CONFLICT (table_id) DO NOTHING",
            )
            .bind(table_id)
            .execute(&mut *tx)
            .await?;
            sqlx::query(
                "UPDATE ducklake_table_stats SET
                     record_count = (SELECT COALESCE(SUM(record_count), 0)
                                     FROM ducklake_data_file
                                     WHERE table_id = $1 AND end_snapshot IS NULL),
                     file_size_bytes = (SELECT COALESCE(SUM(file_size_bytes), 0)
                                        FROM ducklake_data_file
                                        WHERE table_id = $1 AND end_snapshot IS NULL)
                 WHERE table_id = $1",
            )
            .bind(table_id)
            .execute(&mut *tx)
            .await?;

            sqlx::query(
                "INSERT INTO ducklake_snapshot_changes (snapshot_id, changes_made, commit_message)
                 VALUES ($1, $2, $3)",
            )
            .bind(snapshot_id)
            .bind(format!("compacted_table:{table_id}"))
            .bind("datafusion compaction")
            .execute(&mut *tx)
            .await?;

            let schema_id: i64 =
                sqlx::query_scalar("SELECT schema_id FROM ducklake_table WHERE table_id = $1")
                    .bind(table_id)
                    .fetch_one(&mut *tx)
                    .await?;

            // advance_catalog_head MUST be the last write before commit.
            advance_catalog_head(self.catalog_id, snapshot_id, &mut tx).await?;

            tx.commit().await?;
            Ok(CommitIds {
                snapshot_id,
                schema_id,
                table_id,
            })
        })
    }

    fn commit_truncate(
        &self,
        table_id: i64,
        _schema_name: &str,
        _table_name: &str,
        _base_snapshot: i64,
    ) -> Result<u64> {
        block_on(async {
            // Metadata-only truncate in one snapshot under the catalog lock: end
            // every live data file and its live delete file, zero the visible stat
            // totals, and advance the head LAST. next_row_id is preserved.
            let mut tx = self.pool.begin().await?;
            lock_catalog(self.catalog_id, self.lock_timeout_ms, &mut tx).await?;
            assert_table_in_catalog(self.catalog_id, table_id, &mut tx).await?;

            // No-op guard: nothing to truncate if the table has no live data file.
            // Return Ok(0) BEFORE allocating a snapshot, so a repeated
            // `DELETE FROM t` under a pinned snapshot does not create a
            // content-free snapshot. lock_catalog above already serializes, so
            // this read is stable.
            let has_live_data: Option<i64> = sqlx::query_scalar(
                "SELECT data_file_id FROM ducklake_data_file
                 WHERE table_id = $1 AND end_snapshot IS NULL LIMIT 1",
            )
            .bind(table_id)
            .fetch_optional(&mut *tx)
            .await?;
            let inlined_tables = sqlx::query(
                "SELECT table_name FROM ducklake_inlined_data_tables WHERE table_id = $1",
            )
            .bind(table_id)
            .fetch_all(&mut *tx)
            .await?;
            let mut live_inlined = 0i64;
            for row in &inlined_tables {
                let table_name: String = row.try_get(0)?;
                let sql = format!(
                    "SELECT COUNT(*)::BIGINT FROM {} WHERE end_snapshot IS NULL",
                    quote_ident(&table_name)
                );
                live_inlined += sqlx::query_scalar::<_, i64>(AssertSqlSafe(sql))
                    .fetch_one(&mut *tx)
                    .await?;
            }
            if has_live_data.is_none() && live_inlined == 0 {
                return Ok(0);
            }

            let snapshot_id: i64 = sqlx::query(
                "INSERT INTO ducklake_snapshot (snapshot_time, schema_version)
                 VALUES (NOW(), 0) RETURNING snapshot_id",
            )
            .fetch_one(&mut *tx)
            .await?
            .try_get(0)?;
            let prev_max: i64 = sqlx::query(
                "SELECT COALESCE(MAX(s.schema_version), 0) FROM ducklake_snapshot s
                 JOIN ducklake_catalog_snapshot_map m ON m.snapshot_id = s.snapshot_id
                 WHERE m.catalog_id = $1",
            )
            .bind(self.catalog_id)
            .fetch_one(&mut *tx)
            .await?
            .try_get(0)?;
            sqlx::query("UPDATE ducklake_snapshot SET schema_version = $1 WHERE snapshot_id = $2")
                .bind(prev_max.max(1))
                .bind(snapshot_id)
                .execute(&mut *tx)
                .await?;

            // Rows removed = gross record_count minus still-live delete counts,
            // computed BEFORE ending anything (so it matches what we retire).
            // SUM(bigint) is NUMERIC in Postgres; cast back to BIGINT for i64.
            let gross: Option<i64> = sqlx::query_scalar(
                "SELECT COALESCE(SUM(record_count), 0)::BIGINT FROM ducklake_data_file WHERE table_id = $1 AND end_snapshot IS NULL",
            )
            .bind(table_id)
            .fetch_optional(&mut *tx)
            .await?;
            let deleted: i64 = sqlx::query_scalar(
                "SELECT COALESCE(SUM(delete_count), 0)::BIGINT FROM ducklake_delete_file
                 WHERE table_id = $1 AND end_snapshot IS NULL",
            )
            .bind(table_id)
            .fetch_one(&mut *tx)
            .await?;
            let inlined_table = crate::metadata_provider::inlined_delete_table_name(table_id)?;
            let inlined_exists: bool = sqlx::query_scalar("SELECT to_regclass($1) IS NOT NULL")
                .bind(&inlined_table)
                .fetch_one(&mut *tx)
                .await?;
            let inlined_deleted: i64 = if inlined_exists {
                sqlx::query(AssertSqlSafe(format!(
                    "LOCK TABLE \"{inlined_table}\" IN SHARE MODE"
                )))
                .execute(&mut *tx)
                .await?;
                sqlx::query_scalar(AssertSqlSafe(format!(
                    "SELECT COUNT(*)::BIGINT FROM (
                       SELECT DISTINCT d.file_id, d.row_id
                       FROM \"{inlined_table}\" d
                       JOIN ducklake_data_file f ON f.data_file_id = d.file_id
                       WHERE f.table_id = $1 AND f.end_snapshot IS NULL
                         AND d.begin_snapshot <= $2
                     ) counted"
                )))
                .bind(table_id)
                .bind(snapshot_id)
                .fetch_one(&mut *tx)
                .await?
            } else {
                0
            };
            let live_rows =
                (gross.unwrap_or(0) + live_inlined - deleted - inlined_deleted).max(0) as u64;

            sqlx::query(
                "UPDATE ducklake_data_file SET end_snapshot = $1
                 WHERE table_id = $2 AND end_snapshot IS NULL",
            )
            .bind(snapshot_id)
            .bind(table_id)
            .execute(&mut *tx)
            .await?;
            for row in inlined_tables {
                let table_name: String = row.try_get(0)?;
                let sql = format!(
                    "UPDATE {} SET end_snapshot = $1 WHERE end_snapshot IS NULL",
                    quote_ident(&table_name)
                );
                sqlx::query(AssertSqlSafe(sql))
                    .bind(snapshot_id)
                    .execute(&mut *tx)
                    .await?;
            }
            sqlx::query(
                "UPDATE ducklake_delete_file SET end_snapshot = $1
                 WHERE table_id = $2 AND end_snapshot IS NULL",
            )
            .bind(snapshot_id)
            .bind(table_id)
            .execute(&mut *tx)
            .await?;
            sqlx::query(
                "UPDATE ducklake_table_stats SET record_count = 0, file_size_bytes = 0
                 WHERE table_id = $1",
            )
            .bind(table_id)
            .execute(&mut *tx)
            .await?;

            record_snapshot_changes(
                &mut tx,
                snapshot_id,
                &format!("deleted_from_table:{table_id}"),
                &SnapshotCommitMetadata::default(),
            )
            .await?;
            // advance_catalog_head MUST be the last write before commit.
            advance_catalog_head(self.catalog_id, snapshot_id, &mut tx).await?;

            tx.commit().await?;
            Ok(live_rows)
        })
    }

    fn retire_appends_since(&self, table_id: i64, base_snapshot: i64) -> Result<Option<i64>> {
        block_on(async {
            // Metadata-only rollback of a pure-append delta, in one snapshot under
            // the catalog lock: end every live data file added after base_snapshot,
            // recompute the visible stats from the survivors, advance the head LAST.
            // next_row_id is preserved (rowids never reused).
            let mut tx = self.pool.begin().await?;
            lock_catalog(self.catalog_id, self.lock_timeout_ms, &mut tx).await?;
            assert_table_in_catalog(self.catalog_id, table_id, &mut tx).await?;

            // Purity guard — the ONLY change since base may be appended data files.
            // Checked before the no-op return so a delete-only / schema-only orphan
            // surfaces as Conflict instead of a silent no-op. A post-base delete
            // file, a base-era data file ended after base (replace/delete/update),
            // or any post-base column version (schema promotion / add) all mean a
            // non-append mutation we cannot faithfully revert with forward-only file
            // retirement — refuse so the caller keeps its read freeze.
            let impure_delete: Option<i64> = sqlx::query_scalar(
                "SELECT 1::BIGINT FROM ducklake_delete_file
                 WHERE table_id = $1 AND begin_snapshot > $2 LIMIT 1",
            )
            .bind(table_id)
            .bind(base_snapshot)
            .fetch_optional(&mut *tx)
            .await?;
            let impure_ended: Option<i64> = sqlx::query_scalar(
                "SELECT 1::BIGINT FROM ducklake_data_file
                 WHERE table_id = $1 AND begin_snapshot <= $2
                   AND end_snapshot IS NOT NULL AND end_snapshot > $2 LIMIT 1",
            )
            .bind(table_id)
            .bind(base_snapshot)
            .fetch_optional(&mut *tx)
            .await?;
            let impure_column: Option<i64> = sqlx::query_scalar(
                "SELECT 1::BIGINT FROM ducklake_column
                 WHERE table_id = $1 AND (begin_snapshot > $2 OR end_snapshot > $2) LIMIT 1",
            )
            .bind(table_id)
            .bind(base_snapshot)
            .fetch_optional(&mut *tx)
            .await?;
            if impure_delete.is_some() || impure_ended.is_some() || impure_column.is_some() {
                return Err(crate::DuckLakeError::Conflict(format!(
                    "table {table_id}: changes since snapshot {base_snapshot} are not a pure \
                     append (delete/replace/update or schema change present); refusing to retire"
                )));
            }

            // Appended files to retire (live, begin_snapshot > base). No-op → None
            // BEFORE allocating a snapshot, so repeated reconcile calls on a clean
            // table never mint a content-free snapshot.
            let has_appended: Option<i64> = sqlx::query_scalar(
                "SELECT 1::BIGINT FROM ducklake_data_file
                 WHERE table_id = $1 AND end_snapshot IS NULL AND begin_snapshot > $2 LIMIT 1",
            )
            .bind(table_id)
            .bind(base_snapshot)
            .fetch_optional(&mut *tx)
            .await?;
            if has_appended.is_none() {
                return Ok(None);
            }

            let snapshot_id: i64 = sqlx::query(
                "INSERT INTO ducklake_snapshot (snapshot_time, schema_version)
                 VALUES (NOW(), 0) RETURNING snapshot_id",
            )
            .fetch_one(&mut *tx)
            .await?
            .try_get(0)?;
            let prev_max: i64 = sqlx::query(
                "SELECT COALESCE(MAX(s.schema_version), 0) FROM ducklake_snapshot s
                 JOIN ducklake_catalog_snapshot_map m ON m.snapshot_id = s.snapshot_id
                 WHERE m.catalog_id = $1",
            )
            .bind(self.catalog_id)
            .fetch_one(&mut *tx)
            .await?
            .try_get(0)?;
            sqlx::query("UPDATE ducklake_snapshot SET schema_version = $1 WHERE snapshot_id = $2")
                .bind(prev_max.max(1))
                .bind(snapshot_id)
                .execute(&mut *tx)
                .await?;

            // Retire the appended files (append writes no delete files, so there are
            // none to end — the purity guard above rejected any delete file > base).
            sqlx::query(
                "UPDATE ducklake_data_file SET end_snapshot = $1
                 WHERE table_id = $2 AND end_snapshot IS NULL AND begin_snapshot > $3",
            )
            .bind(snapshot_id)
            .bind(table_id)
            .bind(base_snapshot)
            .execute(&mut *tx)
            .await?;

            // Recompute the visible stat totals from the surviving live files
            // (mirrors the compaction commit); next_row_id is deliberately not
            // touched. With the pure-append guarantee, the survivors are exactly
            // base_snapshot's files, so this restores base_snapshot's stats.
            sqlx::query(
                "UPDATE ducklake_table_stats SET
                     record_count = (SELECT COALESCE(SUM(record_count), 0)
                                     FROM ducklake_data_file
                                     WHERE table_id = $1 AND end_snapshot IS NULL),
                     file_size_bytes = (SELECT COALESCE(SUM(file_size_bytes), 0)
                                        FROM ducklake_data_file
                                        WHERE table_id = $1 AND end_snapshot IS NULL)
                 WHERE table_id = $1",
            )
            .bind(table_id)
            .execute(&mut *tx)
            .await?;

            // Recompute per-column stats from the survivors too (register_data_file
            // widened them when the appended file landed). Read the live column
            // generation to drive the numeric-vs-not classification.
            let (columns, column_ids) = live_columns_for_stats(table_id, &mut tx).await?;
            recompute_table_column_stats(&mut tx, table_id, &columns, &column_ids).await?;

            record_snapshot_changes(
                &mut tx,
                snapshot_id,
                &format!("deleted_from_table:{table_id}"),
                &SnapshotCommitMetadata::default(),
            )
            .await?;
            // advance_catalog_head MUST be the last write before commit.
            advance_catalog_head(self.catalog_id, snapshot_id, &mut tx).await?;

            tx.commit().await?;
            Ok(Some(snapshot_id))
        })
    }

    fn publish_snapshot(
        &self,
        table_id: i64,
        schema_name: &str,
        table_name: &str,
        _snapshot_id: i64,
        mode: WriteMode,
        base_snapshot: i64,
        columns: &[ColumnDef],
        column_ids: &[i64],
    ) -> Result<CommitIds> {
        block_on(async {
            // Fileless commit point (CREATE TABLE, zero-row Replace). Same atomic
            // model as register_data_file minus the data-file insert: write all
            // metadata via finalize_snapshot, then advance the head LAST.
            let mut tx = self.pool.begin().await?;
            lock_catalog(self.catalog_id, self.lock_timeout_ms, &mut tx).await?;
            assert_table_not_in_other_catalog(self.catalog_id, table_id, &mut tx).await?;

            let (snapshot_id, schema_id, table_id) = finalize_snapshot(
                self.catalog_id,
                schema_name,
                table_name,
                table_id,
                columns,
                column_ids,
                mode,
                base_snapshot,
                &mut tx,
            )
            .await?;

            let replaced_existing_data = replaced_storage(&mut tx, table_id, snapshot_id).await?;
            let changes_made = table_write_changes(table_id, mode, false, replaced_existing_data);
            record_snapshot_changes(
                &mut tx,
                snapshot_id,
                &changes_made,
                &SnapshotCommitMetadata::default(),
            )
            .await?;
            advance_catalog_head(self.catalog_id, snapshot_id, &mut tx).await?;

            tx.commit().await?;
            Ok(CommitIds {
                snapshot_id,
                schema_id,
                table_id,
            })
        })
    }

    fn end_table_files(&self, table_id: i64, snapshot_id: i64) -> Result<u64> {
        // Used by WriteMode::Replace. End-snapshotting every visible file
        // drops the table's currently-visible row count and byte total to
        // zero (the new files written next will rebuild them). `next_row_id`
        // is deliberately NOT reset: rowids must stay monotonic across the
        // table's lifetime so historical snapshots still resolve uniquely.
        block_on(async {
            let mut tx = self.pool.begin().await?;
            lock_catalog(self.catalog_id, self.lock_timeout_ms, &mut tx).await?;
            assert_table_in_catalog(self.catalog_id, table_id, &mut tx).await?;

            let result = sqlx::query(
                "UPDATE ducklake_data_file SET end_snapshot = $1
                 WHERE table_id = $2 AND end_snapshot IS NULL",
            )
            .bind(snapshot_id)
            .bind(table_id)
            .execute(&mut *tx)
            .await?;
            let n = result.rows_affected();

            sqlx::query(
                "UPDATE ducklake_table_stats
                 SET record_count = 0, file_size_bytes = 0
                 WHERE table_id = $1",
            )
            .bind(table_id)
            .execute(&mut *tx)
            .await?;

            tx.commit().await?;
            Ok(n)
        })
    }

    fn get_data_path(&self) -> Result<String> {
        block_on(async {
            let path: Option<String> =
                sqlx::query_scalar("SELECT data_path FROM ducklake_catalog WHERE catalog_id = $1")
                    .bind(self.catalog_id)
                    .fetch_one(&self.pool)
                    .await?;

            path.ok_or_else(|| {
                crate::error::DuckLakeError::InvalidConfig(
                    "Missing required catalog metadata: 'data_path' not configured.".to_string(),
                )
            })
        })
    }

    fn set_data_path(&self, path: &str) -> Result<()> {
        block_on(async {
            let mut tx = self.pool.begin().await?;
            lock_catalog(self.catalog_id, self.lock_timeout_ms, &mut tx).await?;

            let existing: Option<String> =
                sqlx::query("SELECT data_path FROM ducklake_catalog WHERE catalog_id = $1")
                    .bind(self.catalog_id)
                    .fetch_optional(&mut *tx)
                    .await?
                    .map(|r| r.try_get(0))
                    .transpose()?
                    .flatten();

            match existing {
                Some(cur) if cur == path => {
                    tx.commit().await?;
                    return Ok(());
                },
                Some(cur) => {
                    return Err(crate::error::DuckLakeError::InvalidConfig(format!(
                        "data_path for catalog_id {} already set to {:?}, refusing to overwrite with {:?}",
                        self.catalog_id, cur, path
                    )));
                },
                None => {},
            }

            sqlx::query("UPDATE ducklake_catalog SET data_path = $1 WHERE catalog_id = $2")
                .bind(path)
                .bind(self.catalog_id)
                .execute(&mut *tx)
                .await?;

            tx.commit().await?;
            Ok(())
        })
    }

    fn initialize_schema(&self) -> Result<()> {
        block_on(async {
            execute_ddl_statements(&self.pool, SQL_CREATE_STANDARD_TABLES).await?;
            execute_ddl_statements(&self.pool, SQL_CREATE_MULTICATALOG_TABLES).await?;
            migrate_column_default_metadata(&self.pool).await?;
            sqlx::query("ALTER TABLE ducklake_metadata ADD COLUMN IF NOT EXISTS scope VARCHAR")
                .execute(&self.pool)
                .await?;
            sqlx::query("ALTER TABLE ducklake_metadata ADD COLUMN IF NOT EXISTS scope_id BIGINT")
                .execute(&self.pool)
                .await?;
            // Upgrade a pre-existing store's ducklake_column to the composite PK
            // (legacy single-row column_id PK → versioned-capable). Idempotent.
            migrate_ducklake_column_to_composite_pk(&self.pool).await?;
            // Record file ownership on its own column instead of inferring it from
            // path relativity, and create the indexes that serve it. Idempotent.
            migrate_file_ownership_column(&self.pool).await?;
            Ok(())
        })
    }

    #[tracing::instrument(
        name = "ducklake.begin_write_transaction",
        level = "info",
        skip_all,
        fields(schema_name, table_name, columns = columns.len())
    )]
    fn begin_write_transaction(
        &self,
        schema_name: &str,
        table_name: &str,
        columns: &[ColumnDef],
        mode: WriteMode,
    ) -> Result<WriteSetupResult> {
        validate_name(schema_name, "Schema")?;
        validate_name(table_name, "Table")?;
        if columns.is_empty() {
            return Err(crate::DuckLakeError::InvalidConfig(
                "Table must have at least one column".to_string(),
            ));
        }
        let catalog_id = self.catalog_id;
        let lock_timeout_ms = self.lock_timeout_ms;
        block_on(async {
            // RESERVE ONLY: this transaction inserts NOTHING into ducklake_snapshot
            // / ducklake_schema / ducklake_table / ducklake_column /
            // ducklake_schema_versions / either map table. All of that is written
            // atomically at the commit point (register_data_file /
            // publish_snapshot → finalize_snapshot) and assigned a commit-ordered
            // snapshot id there, so no metadata row is ever readable before its
            // write has published. Here we only reserve ids from the IDENTITY
            // sequences (gaps are fine) and capture the conflict base.
            //
            // The catalog FOR UPDATE lock is held so the existing-column read and
            // the id-reuse map are consistent and the returned field-ids are
            // stable; it is released at tx.commit (which commits only the
            // non-transactional sequence advances).
            let mut tx = self.pool.begin().await?;
            lock_catalog(catalog_id, lock_timeout_ms, &mut tx).await?;

            // Look up (do NOT create) the live schema; reserve a fresh id if
            // absent. setup.schema_id is informational (no caller bakes it into a
            // file — the parquet path encodes the catalog id, not the schema id),
            // so for a brand-new schema this reserved id is NOT the committed id:
            // finalize_snapshot re-derives/reserves the schema id at the commit.
            // The reservation here keeps setup.schema_id distinct & non-zero across
            // concurrent new schemas (sequence gaps from the unused reservation are
            // expected and harmless).
            let schema_id: i64 = {
                let existing = sqlx::query(
                    "SELECT s.schema_id FROM ducklake_schema s
                     JOIN ducklake_catalog_schema_map m ON m.schema_id = s.schema_id
                     WHERE m.catalog_id = $1 AND s.schema_name = $2 AND s.end_snapshot IS NULL",
                )
                .bind(catalog_id)
                .bind(schema_name)
                .fetch_optional(&mut *tx)
                .await?;
                match existing {
                    Some(row) => row.try_get(0)?,
                    None => reserve_ids("ducklake_schema", "schema_id", 1, &mut tx).await?[0],
                }
            };

            // Look up (do NOT create) the live table; reserve an id if absent. The
            // reserved id IS used (threaded to finalize as table_id_hint).
            let table_id: i64 = {
                let existing = sqlx::query(
                    "SELECT table_id FROM ducklake_table
                     WHERE schema_id = $1 AND table_name = $2 AND end_snapshot IS NULL",
                )
                .bind(schema_id)
                .bind(table_name)
                .fetch_optional(&mut *tx)
                .await?;
                if let Some(row) = existing {
                    row.try_get(0)?
                } else {
                    reserve_ids("ducklake_table", "table_id", 1, &mut tx).await?[0]
                }
            };

            // Read existing columns (name, type, nullable, id) to drive (a) the
            // Append schema-evolution check and (b) id reuse: an unchanged column
            // keeps its column_id (== parquet field_id), so an already-written
            // file's field-ids stay valid.
            let rows = sqlx::query(
                "SELECT column_name, column_type, column_id, parent_column
                 FROM ducklake_column
                 WHERE table_id = $1 AND end_snapshot IS NULL
                 ORDER BY column_order",
            )
            .bind(table_id)
            .fetch_all(&mut *tx)
            .await?;
            let mut existing_catalog_columns = Vec::with_capacity(rows.len());
            for row in rows {
                let name: String = row.try_get(0)?;
                let ducklake_type: String = row.try_get(1)?;
                let column_id: i64 = row.try_get(2)?;
                let parent_column: Option<i64> = row.try_get(3)?;
                existing_catalog_columns.push(ExistingCatalogColumn {
                    column_id,
                    name,
                    ducklake_type,
                    parent_column,
                });
            }

            let catalog_columns = catalog_column_defs(columns)?;
            let fresh_ids = reserve_ids(
                "ducklake_column",
                "column_id",
                catalog_columns.len() as i64,
                &mut tx,
            )
            .await?;
            let field_ids =
                assign_column_ids(&catalog_columns, &existing_catalog_columns, &fresh_ids)?;

            // Data-write policy (§5, same rules as the SQLite writer): a data write
            // — Replace OR Append — must NOT change a column's type. A type change is
            // schema evolution and must go through `promote_column_type`, never a data
            // write; silently keeping the old catalog type (the "C" bug) corrupts
            // reads. Canonical comparison (`int64` ≡ `bigint`) so an alias-only
            // restatement is a no-op. Append additionally requires a genuinely new
            // column to be nullable (a Replace overwrites every row, so a new
            // non-nullable column is fine there).
            if !existing_catalog_columns.is_empty() {
                use std::collections::HashMap;
                let existing_map: HashMap<i64, &ExistingCatalogColumn> = existing_catalog_columns
                    .iter()
                    .map(|column| (column.column_id, column))
                    .collect();

                for (new_column, column_id) in catalog_columns.iter().zip(&field_ids) {
                    if let Some(existing_column) = existing_map.get(column_id) {
                        let same_type =
                            catalog_column_type_equal(&existing_column.ducklake_type, new_column);
                        if !same_type {
                            return Err(crate::error::DuckLakeError::UnsupportedTypeChange {
                                operation: TypeChangeOperation::DataWrite {
                                    mode: match mode {
                                        WriteMode::Replace => TypeChangeWriteMode::Replace,
                                        WriteMode::Append => TypeChangeWriteMode::Append,
                                    },
                                },
                                column: new_column.name.clone(),
                                from: existing_column.ducklake_type.clone(),
                                to: new_column.ducklake_type.clone(),
                            });
                        }
                    } else if mode == WriteMode::Append
                        && new_column.parent_index.is_none()
                        && !new_column.is_nullable
                    {
                        return Err(crate::error::DuckLakeError::InvalidConfig(format!(
                            "Schema evolution error: new column '{}' must be nullable. Adding non-nullable columns is not allowed.",
                            new_column.name
                        )));
                    }
                }
            }

            // Reserve N column ids, then compute the final per-column ids. These
            // are baked into the staged parquet's field_id metadata, so they must
            // equal what finalize_snapshot inserts at commit.
            //
            // Mode-independent (matching the SQLite writer): REUSE the existing id
            // for a column whose NAME already exists, consume a freshly-reserved id
            // only for a genuinely-new column. Stable ids are required for BOTH
            // modes — a concurrent in-flight Append bakes the kept columns' ids into
            // its parquet, so a Replace must NOT re-mint them (re-minting would make
            // that Append's rows read back as all-NULL). The Replace conflict check
            // does not rely on a column re-mint: a data Replace leaves a new data
            // file (begin > base) and a schema-changing Replace ends/inserts the
            // changed columns (begin/end > base); a fileless same-schema Replace
            // leaves no trace and resolves last-writer-wins, exactly like SQLite.
            // base = catalog head observed at begin. The Replace commit aborts if
            // any file/column of the table moved past it (a concurrent writer
            // committed a newer generation). Snapshot ids are commit-ordered, so
            // this scalar head is an exact conflict base.
            let base_snapshot_id: i64 = sqlx::query(
                "SELECT COALESCE(MAX(snapshot_id), 0) FROM ducklake_catalog_snapshot_map
                 WHERE catalog_id = $1",
            )
            .bind(catalog_id)
            .fetch_one(&mut *tx)
            .await?
            .try_get(0)?;

            // Commit the (sequence-only) reservation transaction.
            tx.commit().await?;

            Ok(WriteSetupResult {
                // snapshot_id is vestigial here (like SQLite's): the real id is
                // assigned at the commit by finalize_snapshot.
                snapshot_id: 0,
                base_snapshot_id,
                schema_id,
                table_id,
                column_ids: top_level_column_ids(&catalog_columns, &field_ids)?,
                field_ids,
            })
        })
    }

    fn catalog_id(&self) -> Option<i64> {
        Some(self.catalog_id)
    }

    /// Multicatalog Postgres implements the atomic append-with-deletes commit,
    /// so it supports row-level `UPDATE`.
    fn supports_update(&self) -> bool {
        true
    }
}

async fn enforce_inline_partition_fence(
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
    table_id: i64,
    base_snapshot: i64,
) -> Result<()> {
    let changed: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM ducklake_partition_info WHERE table_id = $1 AND (begin_snapshot > $2 OR end_snapshot > $2)")
        .bind(table_id)
        .bind(base_snapshot)
        .fetch_one(&mut **tx).await?;
    if changed != 0 {
        return Err(crate::DuckLakeError::Conflict(format!(
            "partition spec for table {table_id} changed during inline write; retry the write"
        )));
    }
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::{inlined_postgres_type, push_inlined_postgres_value};
    use crate::metadata_writer::{ColumnDef, columns_differ};
    use arrow::array::types::IntervalMonthDayNano;
    use arrow::array::{Array, FixedSizeBinaryArray, IntervalMonthDayNanoArray, UInt64Array};
    use arrow::datatypes::{DataType, Field, IntervalUnit};
    use sqlx::{Postgres, QueryBuilder};
    use std::sync::Arc;

    #[test]
    fn postgres_inlined_types_follow_ducklake_encodings() {
        assert_eq!(inlined_postgres_type(&DataType::Int8, "int8"), "SMALLINT");
        assert_eq!(inlined_postgres_type(&DataType::UInt32, "uint32"), "BIGINT");
        assert_eq!(
            inlined_postgres_type(&DataType::UInt64, "uint64"),
            "VARCHAR"
        );
        assert_eq!(inlined_postgres_type(&DataType::Utf8, "varchar"), "BYTEA");
        assert_eq!(
            inlined_postgres_type(&DataType::FixedSizeBinary(16), "uuid"),
            "UUID"
        );
        assert_eq!(
            inlined_postgres_type(&DataType::FixedSizeBinary(16), "fixed_size_binary(16)"),
            "BYTEA"
        );
        assert_eq!(
            inlined_postgres_type(&DataType::FixedSizeBinary(32), "fixed_size_binary(32)"),
            "BYTEA"
        );
        assert_eq!(inlined_postgres_type(&DataType::Date32, "date"), "VARCHAR");
        assert_eq!(
            inlined_postgres_type(
                &DataType::Timestamp(arrow::datatypes::TimeUnit::Nanosecond, None),
                "timestamp_ns",
            ),
            "VARCHAR"
        );
        assert_eq!(
            inlined_postgres_type(
                &DataType::List(Arc::new(Field::new("item", DataType::Int32, true,))),
                "list"
            ),
            "VARCHAR"
        );
    }

    #[test]
    fn postgres_uint64_inline_insert_casts_the_parameter() {
        let values = UInt64Array::from(vec![u64::MAX]);
        let mut query = QueryBuilder::<Postgres>::new("VALUES (");
        push_inlined_postgres_value(&mut query, &values, 0, "VARCHAR").unwrap();
        query.push(')');

        assert_eq!(query.sql(), "VALUES (CAST($1 AS VARCHAR))");
    }

    #[test]
    fn postgres_rejects_submicrosecond_interval_inlining() {
        let values = IntervalMonthDayNanoArray::from(vec![IntervalMonthDayNano::new(0, 0, 1)]);
        let mut query = QueryBuilder::<Postgres>::new("VALUES (");
        let error = push_inlined_postgres_value(&mut query, &values, 0, "INTERVAL").unwrap_err();

        assert!(matches!(error, crate::DuckLakeError::Unsupported(_)));
        assert!(error.to_string().contains("sub-microsecond"));
        assert_eq!(
            values.data_type(),
            &DataType::Interval(IntervalUnit::MonthDayNano)
        );
    }

    #[test]
    fn postgres_non_uuid_fixed_16_binds_raw_bytes() {
        let bytes = [0x00_u8, 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07];
        let bytes = [bytes.as_slice(), bytes.as_slice()].concat();
        let values = FixedSizeBinaryArray::try_from_iter([bytes.as_slice()].into_iter()).unwrap();
        let mut query = QueryBuilder::<Postgres>::new("VALUES (");
        push_inlined_postgres_value(&mut query, &values, 0, "BYTEA").unwrap();
        query.push(')');

        assert_eq!(query.sql(), "VALUES ($1)");
        assert_eq!(
            inlined_postgres_type(values.data_type(), "fixed_size_binary(16)"),
            "BYTEA"
        );
    }

    #[test]
    fn test_columns_differ_identical() {
        let existing =
            vec![("id".into(), "int64".into(), false), ("name".into(), "varchar".into(), true)];
        let proposed = vec![
            ColumnDef::new("id", "int64", false).unwrap(),
            ColumnDef::new("name", "varchar", true).unwrap(),
        ];
        assert!(!columns_differ(&existing, &proposed));
    }

    #[test]
    fn test_columns_differ_added_column() {
        let existing = vec![("id".into(), "int64".into(), false)];
        let proposed = vec![
            ColumnDef::new("id", "int64", false).unwrap(),
            ColumnDef::new("name", "varchar", true).unwrap(),
        ];
        assert!(columns_differ(&existing, &proposed));
    }

    #[test]
    fn test_columns_differ_renamed_column() {
        let existing = vec![("id".into(), "int64".into(), false)];
        let proposed = vec![ColumnDef::new("user_id", "int64", false).unwrap()];
        assert!(columns_differ(&existing, &proposed));
    }

    #[test]
    fn test_columns_differ_type_change() {
        let existing = vec![("id".into(), "int32".into(), false)];
        let proposed = vec![ColumnDef::new("id", "varchar", false).unwrap()];
        assert!(columns_differ(&existing, &proposed));
    }

    #[test]
    fn test_columns_differ_forward_widening_is_a_change() {
        // existing int32 -> proposed int64 is a forward widening. Since #149 this
        // is NOT "the same column": on a data write it's rejected at begin-time
        // (widenings must go through promote_column_type), and if it reaches
        // columns_differ it must classify as DDL. Only the benign promote-race
        // direction below is treated as same-type.
        let existing = vec![("id".into(), "int32".into(), false)];
        let proposed = vec![ColumnDef::new("id", "int64", false).unwrap()];
        assert!(columns_differ(&existing, &proposed));
    }

    #[test]
    fn test_columns_differ_benign_promote_race_is_not_ddl() {
        // The Append-vs-promote race: the committed column was already widened to
        // int64 by a concurrent promote, while this write staged the narrower
        // int32 (which passed the begin-time reject against the type AT BEGIN).
        // The staged int32 losslessly widens to the committed int64 and is served
        // via cast-on-read, so it is NOT a schema change and must not bump
        // schema_version.
        let existing = vec![("id".into(), "int64".into(), false)];
        let proposed = vec![ColumnDef::new("id", "int32", false).unwrap()];
        assert!(!columns_differ(&existing, &proposed));
    }

    #[test]
    fn test_columns_differ_nullability_change() {
        let existing = vec![("id".into(), "int64".into(), false)];
        let proposed = vec![ColumnDef::new("id", "int64", true).unwrap()];
        assert!(columns_differ(&existing, &proposed));
    }

    #[test]
    fn test_columns_differ_alias_canonical() {
        // bigint and int64 normalize to the same canonical type.
        let existing = vec![("id".into(), "bigint".into(), false)];
        let proposed = vec![ColumnDef::new("id", "int64", false).unwrap()];
        assert!(!columns_differ(&existing, &proposed));
    }
}