use arrow::record_batch::RecordBatch;
#[cfg(any(feature = "postgres", feature = "quack"))]
use async_trait::async_trait;
#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
use ducklake_catalog::DeleteFileId;
use ducklake_catalog::{
    ColumnEntry, ColumnId, DeleteFormat, SchemaEntry, SchemaId, SnapshotId, TableEntry, TableId,
    Validity, ViewEntry, ViewId, try_build_schema,
};
use std::borrow::Cow;

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
use crate::InlinedDataAppend;
use crate::{
    ActivePartitionSpec, ActiveSortSpec, AddColumn, AddColumnCommit, AppendFile,
    AppendFilesSnapshotPlan, CatalogSnapshotIds, CleanupMetadataFile, ColumnCloseRequest,
    ColumnDefaultCommit, ColumnNullabilityCommit, ColumnSubtreeCloseRequest, ColumnTypeCommit,
    ColumnTypeTreeCommit, CreateSchemaCommit, CreateTableColumn, CreateTableCommit,
    CreateTableWithFilesCommit, CreateTableWithInlinedDataCommit, CreateViewCommit,
    DeleteFileCommit, DeleteFilesSnapshotPlan, DropColumnCommit, DropColumnFacts, DropSchemaCommit,
    DropTableCommit, DropViewCommit, FlushedInlinedDataCommit, FlushedInlinedFileDeletionCommit,
    InlinedDataAppendCommit, InlinedDataDeletionCommit, InlinedFileDeletionCommit,
    MaintenanceCommitPlan, MergeAdjacentCompactionCommit, MetadataMutationPlan,
    PartitionColumnCommit, PartitioningBatchCommit, PartitioningTableFacts,
    PlannedColumnReplacement, PlannedColumnTypeChange, PlannedColumnTypeTreeChange,
    PlannedColumnTypeTreeRow, RenameColumnCommit, RenameTableCommit, Result,
    RewriteDataFilesCommit, SortExpressionCommit, SortedByCommit, SortedByCommitPlan, StorageError,
    TableChangeCommit, TableChangeSnapshotPlan, TableColumnStats, TableSchema, TruncateTableCommit,
    TruncateTableSnapshotPlan, drop_column_sort_reference_lookup, final_partitioning_changes,
    plan_add_columns_commit, plan_append_files_commit, plan_column_default_mutation,
    plan_column_nullability_mutation, plan_column_type_mutation, plan_column_type_tree_mutation,
    plan_create_schema_catalog_mutation, plan_create_table_catalog_mutation,
    plan_create_table_with_files_commit, plan_create_table_with_inlined_data_commit,
    plan_create_view_catalog_mutation, plan_delete_files_commit, plan_drop_column_mutation,
    plan_drop_schema_catalog_mutation, plan_drop_table_catalog_mutation,
    plan_drop_view_catalog_mutation, plan_merge_adjacent_compaction_commit,
    plan_merge_adjacent_compaction_sources, plan_partitioning_batch_commit,
    plan_rename_column_mutation_start, plan_rename_table_catalog_mutation,
    plan_rewrite_data_files_commit, plan_rewrite_data_files_sources, plan_sorted_by_commit,
    plan_table_change_commit, plan_truncate_table_commit, renamed_table_path_for_target_schema,
    sort_expression_commit, sort_expression_references_unqualified_column, sort_key_index,
    validate_sort_expression_columns,
};

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct CommitSnapshotIds {
    pub(crate) snapshot_id: i64,
    pub(crate) schema_version: i64,
    pub(crate) next_catalog_id: i64,
    pub(crate) next_file_id: i64,
}

#[derive(Debug, Clone)]
struct PlannedSnapshotCommit<'a> {
    new_snapshot: i64,
    new_schema_version: i64,
    next_catalog_id: i64,
    next_file_id: i64,
    changes_made: Cow<'a, str>,
    commit_message: Cow<'a, str>,
}

impl PlannedSnapshotCommit<'_> {
    fn snapshot_id(self) -> SnapshotId {
        SnapshotId(self.new_snapshot)
    }
}

fn planned_snapshot_commit<'a>(
    new_snapshot: i64,
    new_schema_version: i64,
    next_catalog_id: i64,
    next_file_id: i64,
    changes_made: impl Into<Cow<'a, str>>,
    commit_message: impl Into<Cow<'a, str>>,
) -> PlannedSnapshotCommit<'a> {
    PlannedSnapshotCommit {
        new_snapshot,
        new_schema_version,
        next_catalog_id,
        next_file_id,
        changes_made: changes_made.into(),
        commit_message: commit_message.into(),
    }
}

fn metadata_snapshot_commit(metadata: &MetadataMutationPlan) -> PlannedSnapshotCommit<'_> {
    planned_snapshot_commit(
        metadata.new_snapshot,
        metadata.new_schema_version,
        metadata.next_catalog_id,
        metadata.next_file_id,
        &metadata.changes_made,
        &metadata.commit_message,
    )
}

fn catalog_object_snapshot_commit(
    plan: &crate::CatalogObjectMutationPlan,
) -> PlannedSnapshotCommit<'_> {
    planned_snapshot_commit(
        plan.new_snapshot,
        plan.new_schema_version,
        plan.next_catalog_id,
        plan.next_file_id,
        &plan.changes_made,
        &plan.commit_message,
    )
}

fn partitioning_snapshot_commit(plan: &crate::PartitioningBatchPlan) -> PlannedSnapshotCommit<'_> {
    planned_snapshot_commit(
        plan.new_snapshot,
        plan.new_schema_version,
        plan.next_catalog_id,
        plan.next_file_id,
        &plan.changes_made,
        &plan.commit_message,
    )
}

fn maintenance_snapshot_commit(plan: &MaintenanceCommitPlan) -> PlannedSnapshotCommit<'_> {
    planned_snapshot_commit(
        plan.new_snapshot,
        plan.schema_version,
        plan.next_catalog_id,
        plan.new_next_file_id,
        &plan.changes_made,
        &plan.commit_message,
    )
}

fn catalog_object_snapshot_commit_with_next_file_id(
    plan: &crate::CatalogObjectMutationPlan,
    next_file_id: i64,
) -> PlannedSnapshotCommit<'_> {
    planned_snapshot_commit(
        plan.new_snapshot,
        plan.new_schema_version,
        plan.next_catalog_id,
        next_file_id,
        &plan.changes_made,
        &plan.commit_message,
    )
}

fn append_files_snapshot_commit(
    ids: CommitSnapshotIds,
    table_id: TableId,
    plan: &AppendFilesSnapshotPlan,
) -> PlannedSnapshotCommit<'static> {
    planned_snapshot_commit(
        plan.new_snapshot.0,
        ids.schema_version,
        ids.next_catalog_id,
        plan.new_next_file_id,
        format!("inserted_into_table:{}", table_id.0),
        "datafusion append",
    )
}

fn inlined_append_snapshot_ids(ids: CommitSnapshotIds) -> Result<(i64, i64)> {
    let new_snapshot = ids
        .snapshot_id
        .checked_add(1)
        .ok_or_else(|| StorageError::Backend("DuckLake snapshot id overflow".into()))?;
    let next_file_id = ids
        .next_file_id
        .checked_add(1)
        .ok_or_else(|| StorageError::Backend("DuckLake next_file_id overflow".into()))?;
    Ok((new_snapshot, next_file_id))
}

fn inlined_append_snapshot_commit<'a>(
    ids: CommitSnapshotIds,
    append: &'a InlinedDataAppendCommit,
    new_snapshot: i64,
    next_file_id: i64,
) -> PlannedSnapshotCommit<'a> {
    planned_snapshot_commit(
        new_snapshot,
        ids.schema_version,
        ids.next_catalog_id,
        next_file_id,
        append.changes_made.as_str(),
        append.commit_message.as_str(),
    )
}

fn inlined_append_next_row_id(row_id_start: i64, rows_appended: i64) -> Result<i64> {
    row_id_start
        .checked_add(rows_appended)
        .ok_or_else(|| StorageError::Backend("DuckLake inlined append next_row_id overflow".into()))
}

#[derive(Debug, Clone, Copy)]
struct PlannedInlinedAppendCommit {
    new_snapshot: i64,
    next_file_id: i64,
    row_id_start: i64,
    has_table_stats: bool,
}

fn planned_inlined_append_commit(
    ids: CommitSnapshotIds,
    append: &InlinedDataAppendCommit,
    has_inlined_data_tables: bool,
    has_table_stats: bool,
    next_row_id: Option<i64>,
) -> Result<PlannedInlinedAppendCommit> {
    if ids.snapshot_id != append.expected_snapshot.0 {
        return Err(StorageError::conflict(
            "append",
            append.expected_snapshot,
            ids.snapshot_id,
        ));
    }
    require_inlined_data_table_registry(has_inlined_data_tables)?;
    let (new_snapshot, next_file_id) = inlined_append_snapshot_ids(ids)?;
    Ok(PlannedInlinedAppendCommit {
        new_snapshot,
        next_file_id,
        row_id_start: next_row_id.unwrap_or(0),
        has_table_stats,
    })
}

fn require_inlined_data_table_registry(exists: bool) -> Result<()> {
    if !exists {
        return Err(StorageError::Backend(
            "DuckLake inlined append requires ducklake_inlined_data_tables".into(),
        ));
    }
    Ok(())
}

#[derive(Debug, Clone, Copy)]
struct PlannedInlinedBatchInsert<'a> {
    table_id: TableId,
    schema_version: i64,
    schema: &'a TableSchema,
    batches: &'a [RecordBatch],
    preserved_row_ids: Option<&'a [i64]>,
    row_id_start: i64,
    new_snapshot: i64,
}

fn planned_inlined_append_batch_insert<'a>(
    append: &'a InlinedDataAppendCommit,
    plan: &PlannedInlinedAppendCommit,
) -> PlannedInlinedBatchInsert<'a> {
    PlannedInlinedBatchInsert {
        table_id: append.table_id,
        schema_version: append.schema_version,
        schema: &append.schema,
        batches: &append.batches,
        preserved_row_ids: None,
        row_id_start: plan.row_id_start,
        new_snapshot: plan.new_snapshot,
    }
}

fn create_table_schema_from_commit(
    table_id: TableId,
    new_snapshot: i64,
    columns: &[CreateTableColumn],
) -> Result<TableSchema> {
    let snapshot = SnapshotId(new_snapshot);
    let column_entries = columns
        .iter()
        .map(|column| ColumnEntry {
            column_id: column.column_id,
            validity: Validity {
                begin: snapshot,
                end: None,
            },
            table_id,
            column_order: column.column_order,
            column_name: column.column_name.clone(),
            column_type: column.column_type.clone(),
            initial_default: None,
            default_value: Some("NULL".to_string()),
            nulls_allowed: column.nulls_allowed,
            parent_column: column.parent_column,
            default_value_type: Some("literal".to_string()),
            default_value_dialect: Some("duckdb".to_string()),
        })
        .collect::<Vec<_>>();
    try_build_schema(&column_entries, snapshot).map_err(StorageError::Catalog)
}

fn planned_create_table_inlined_batch_insert<'a>(
    create: &'a CreateTableWithInlinedDataCommit,
    schema: &'a TableSchema,
    new_schema_version: i64,
    new_snapshot: i64,
) -> PlannedInlinedBatchInsert<'a> {
    PlannedInlinedBatchInsert {
        table_id: create.table_id,
        schema_version: new_schema_version,
        schema,
        batches: &create.batches,
        preserved_row_ids: None,
        row_id_start: 0,
        new_snapshot,
    }
}

fn planned_table_change_inlined_batch_inserts<'a>(
    changes: &'a TableChangeCommit,
    plans: &'a [crate::InlinedDataAppendPlan],
    new_snapshot: i64,
) -> Result<Vec<PlannedInlinedBatchInsert<'a>>> {
    if changes.inlined_data_appends.len() != plans.len() {
        return Err(StorageError::Backend(format!(
            "DuckLake table-change planned {} inlined append rows for {} input appends",
            plans.len(),
            changes.inlined_data_appends.len()
        )));
    }
    Ok(changes
        .inlined_data_appends
        .iter()
        .zip(plans.iter())
        .map(|(append, plan)| PlannedInlinedBatchInsert {
            table_id: changes.table_id,
            schema_version: append.schema_version,
            schema: &append.schema,
            batches: &append.batches,
            preserved_row_ids: append.row_ids.as_deref(),
            row_id_start: plan.row_id_start,
            new_snapshot,
        })
        .collect())
}

#[derive(Debug, Clone, Copy)]
struct PlannedAppendTableStatsUpdate {
    table_id: TableId,
    record_count: i64,
    allocated_row_id_count: i64,
    file_size_bytes: i64,
}

fn planned_append_table_stats_update(
    table_id: TableId,
    plan: &AppendFilesSnapshotPlan,
) -> PlannedAppendTableStatsUpdate {
    PlannedAppendTableStatsUpdate {
        table_id,
        record_count: plan.total_record_count,
        allocated_row_id_count: plan.total_allocated_row_id_count,
        file_size_bytes: plan.total_file_size_bytes,
    }
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
fn apply_append_table_stats_sync<'tx, O>(
    ops: &O,
    tx: &O::Tx<'tx>,
    enabled: bool,
    stats: PlannedAppendTableStatsUpdate,
) -> Result<()>
where
    O: SyncTableDataCommitOps,
{
    if !enabled {
        return Ok(());
    }
    let updated = ops.update_table_stats_for_append(
        tx,
        stats.table_id,
        stats.record_count,
        stats.allocated_row_id_count,
        stats.file_size_bytes,
    )?;
    if updated == 0 {
        ops.insert_table_stats_from_active_files(tx, stats.table_id)?;
    }
    Ok(())
}

#[cfg(any(feature = "postgres", feature = "quack"))]
async fn apply_append_table_stats_async<O, Tx>(
    ops: &O,
    tx: &Tx,
    enabled: bool,
    stats: PlannedAppendTableStatsUpdate,
) -> Result<()>
where
    O: AsyncTableDataCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    if !enabled {
        return Ok(());
    }
    let updated = ops
        .update_table_stats_for_append(
            tx,
            stats.table_id,
            stats.record_count,
            stats.allocated_row_id_count,
            stats.file_size_bytes,
        )
        .await?;
    if updated == 0 {
        ops.insert_table_stats_from_active_files(tx, stats.table_id)
            .await?;
    }
    Ok(())
}

#[derive(Debug, Clone, Copy)]
struct PlannedInlinedAppendTableStatsUpdate {
    table_id: TableId,
    rows_appended: i64,
    next_row_id: i64,
}

fn planned_inlined_append_table_stats_update(
    table_id: TableId,
    row_id_start: i64,
    rows_appended: i64,
) -> Result<PlannedInlinedAppendTableStatsUpdate> {
    Ok(PlannedInlinedAppendTableStatsUpdate {
        table_id,
        rows_appended,
        next_row_id: inlined_append_next_row_id(row_id_start, rows_appended)?,
    })
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
fn apply_inlined_append_table_stats_sync<'tx, O>(
    ops: &O,
    tx: &O::Tx<'tx>,
    enabled: bool,
    stats: PlannedInlinedAppendTableStatsUpdate,
) -> Result<()>
where
    O: SyncTableDataCommitOps,
{
    if !enabled {
        return Ok(());
    }
    let updated =
        ops.update_table_stats_for_inlined_append(tx, stats.table_id, stats.rows_appended)?;
    if updated == 0 {
        ops.insert_table_stats_for_inlined_append(
            tx,
            stats.table_id,
            stats.rows_appended,
            stats.next_row_id,
        )?;
    }
    Ok(())
}

#[cfg(any(feature = "postgres", feature = "quack"))]
async fn apply_inlined_append_table_stats_async<O, Tx>(
    ops: &O,
    tx: &Tx,
    enabled: bool,
    stats: PlannedInlinedAppendTableStatsUpdate,
) -> Result<()>
where
    O: AsyncTableDataCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    if !enabled {
        return Ok(());
    }
    let updated = ops
        .update_table_stats_for_inlined_append(tx, stats.table_id, stats.rows_appended)
        .await?;
    if updated == 0 {
        ops.insert_table_stats_for_inlined_append(
            tx,
            stats.table_id,
            stats.rows_appended,
            stats.next_row_id,
        )
        .await?;
    }
    Ok(())
}

fn delete_files_snapshot_commit(
    ids: CommitSnapshotIds,
    table_id: TableId,
    plan: &DeleteFilesSnapshotPlan,
) -> PlannedSnapshotCommit<'static> {
    planned_snapshot_commit(
        plan.new_snapshot.0,
        ids.schema_version,
        ids.next_catalog_id,
        plan.new_next_file_id,
        format!("deleted_from_table:{}", table_id.0),
        "datafusion delete",
    )
}

#[derive(Debug, Clone, Copy)]
struct PlannedDeleteTableStatsUpdate {
    table_id: TableId,
    total_delete_count: i64,
}

fn planned_delete_table_stats_update(
    table_id: TableId,
    plan: &DeleteFilesSnapshotPlan,
) -> PlannedDeleteTableStatsUpdate {
    PlannedDeleteTableStatsUpdate {
        table_id,
        total_delete_count: plan.total_delete_count,
    }
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
fn apply_delete_table_stats_sync<'tx, O>(
    ops: &O,
    tx: &O::Tx<'tx>,
    stats: PlannedDeleteTableStatsUpdate,
) -> Result<()>
where
    O: SyncTableDataCommitOps,
{
    if ops.table_exists(tx, "ducklake_table_stats")? {
        ops.update_table_stats_for_delete(tx, stats.table_id, stats.total_delete_count)?;
    }
    Ok(())
}

#[cfg(any(feature = "postgres", feature = "quack"))]
async fn apply_delete_table_stats_async<O, Tx>(
    ops: &O,
    tx: &Tx,
    stats: PlannedDeleteTableStatsUpdate,
) -> Result<()>
where
    O: AsyncTableDataCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    if ops.table_exists(tx, "ducklake_table_stats").await? {
        ops.update_table_stats_for_delete(tx, stats.table_id, stats.total_delete_count)
            .await?;
    }
    Ok(())
}

fn table_change_snapshot_commit<'a>(
    ids: CommitSnapshotIds,
    changes: &'a TableChangeCommit,
    plan: &TableChangeSnapshotPlan,
) -> PlannedSnapshotCommit<'a> {
    planned_snapshot_commit(
        plan.new_snapshot.0,
        ids.schema_version,
        ids.next_catalog_id,
        plan.new_next_file_id,
        changes.changes_made.as_str(),
        changes.commit_message.as_str(),
    )
}

fn add_table_change_delete_count(total: i64, added: i64) -> Result<i64> {
    total
        .checked_add(added)
        .ok_or_else(|| StorageError::Backend("DuckLake table-change delete count overflow".into()))
}

#[derive(Debug, Clone, Copy)]
struct PlannedTableChangeInlinedMutations<'a> {
    table_id: TableId,
    new_snapshot: i64,
    previous_snapshot: i64,
    initial_delete_count: i64,
    inlined_file_deletions: &'a [InlinedFileDeletionCommit],
    inlined_data_deletions: &'a [InlinedDataDeletionCommit],
    flushed_inlined_data: &'a [FlushedInlinedDataCommit],
    flushed_inlined_file_deletions: &'a [FlushedInlinedFileDeletionCommit],
}

fn planned_table_change_inlined_mutations<'a>(
    changes: &'a TableChangeCommit,
    ids: CommitSnapshotIds,
    plan: &TableChangeSnapshotPlan,
) -> PlannedTableChangeInlinedMutations<'a> {
    PlannedTableChangeInlinedMutations {
        table_id: changes.table_id,
        new_snapshot: plan.new_snapshot.0,
        previous_snapshot: ids.snapshot_id,
        initial_delete_count: plan.total_declared_deleted_rows,
        inlined_file_deletions: &changes.inlined_file_deletions,
        inlined_data_deletions: &changes.inlined_data_deletions,
        flushed_inlined_data: &changes.flushed_inlined_data,
        flushed_inlined_file_deletions: &changes.flushed_inlined_file_deletions,
    }
}

fn planned_table_change_total_delete_count(
    mutations: &PlannedTableChangeInlinedMutations<'_>,
    inlined_data_deleted: i64,
    flushed_inlined_data_deleted: i64,
) -> Result<i64> {
    let total =
        add_table_change_delete_count(mutations.initial_delete_count, inlined_data_deleted)?;
    add_table_change_delete_count(total, flushed_inlined_data_deleted)
}

#[derive(Debug, Clone, Copy)]
struct PlannedTableChangeStatsUpdate {
    table_id: TableId,
    appended_record_count: i64,
    total_delete_count: i64,
    appended_file_size_bytes: i64,
    allocated_row_id_count: i64,
}

fn planned_table_change_stats_update(
    table_id: TableId,
    total_delete_count: i64,
    plan: &TableChangeSnapshotPlan,
) -> PlannedTableChangeStatsUpdate {
    PlannedTableChangeStatsUpdate {
        table_id,
        appended_record_count: plan.total_appended_record_count,
        total_delete_count,
        appended_file_size_bytes: plan.total_appended_file_size_bytes,
        allocated_row_id_count: plan.total_allocated_row_id_count,
    }
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
fn apply_table_change_table_stats_sync<'tx, O>(
    ops: &O,
    tx: &O::Tx<'tx>,
    enabled: bool,
    stats: PlannedTableChangeStatsUpdate,
) -> Result<()>
where
    O: SyncTableDataCommitOps,
{
    if !enabled {
        return Ok(());
    }
    let updated = ops.update_table_stats_for_table_change(
        tx,
        stats.table_id,
        stats.appended_record_count,
        stats.total_delete_count,
        stats.appended_file_size_bytes,
        stats.allocated_row_id_count,
    )?;
    if updated == 0 {
        ops.insert_table_stats_from_active_files(tx, stats.table_id)?;
    }
    Ok(())
}

#[cfg(any(feature = "postgres", feature = "quack"))]
async fn apply_table_change_table_stats_async<O, Tx>(
    ops: &O,
    tx: &Tx,
    enabled: bool,
    stats: PlannedTableChangeStatsUpdate,
) -> Result<()>
where
    O: AsyncTableDataCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    if !enabled {
        return Ok(());
    }
    let updated = ops
        .update_table_stats_for_table_change(
            tx,
            stats.table_id,
            stats.appended_record_count,
            stats.total_delete_count,
            stats.appended_file_size_bytes,
            stats.allocated_row_id_count,
        )
        .await?;
    if updated == 0 {
        ops.insert_table_stats_from_active_files(tx, stats.table_id)
            .await?;
    }
    Ok(())
}

#[derive(Debug, Clone, Copy)]
struct PlannedCreateTableStatsInsert {
    table_id: TableId,
    record_count: i64,
    file_size_bytes: i64,
}

fn planned_create_table_stats_insert(
    table_id: TableId,
    plan: &crate::CreateTableWithFilesCommitPlan,
) -> PlannedCreateTableStatsInsert {
    PlannedCreateTableStatsInsert {
        table_id,
        record_count: plan.total_record_count,
        file_size_bytes: plan.total_file_size_bytes,
    }
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
fn apply_create_table_stats_sync<'tx, O>(
    ops: &O,
    tx: &O::Tx<'tx>,
    enabled: bool,
    stats: PlannedCreateTableStatsInsert,
) -> Result<()>
where
    O: SyncCreateTableWithFilesCommitOps,
{
    if enabled {
        ops.insert_table_stats_for_create_table(
            tx,
            stats.table_id,
            stats.record_count,
            stats.file_size_bytes,
        )?;
    }
    Ok(())
}

#[cfg(any(feature = "postgres", feature = "quack"))]
async fn apply_create_table_stats_async<O, Tx>(
    ops: &O,
    tx: &Tx,
    enabled: bool,
    stats: PlannedCreateTableStatsInsert,
) -> Result<()>
where
    O: AsyncCreateTableWithFilesCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    if enabled {
        ops.insert_table_stats_for_create_table(
            tx,
            stats.table_id,
            stats.record_count,
            stats.file_size_bytes,
        )
        .await?;
    }
    Ok(())
}

fn truncate_snapshot_commit<'a>(
    ids: CommitSnapshotIds,
    truncate: &'a TruncateTableCommit,
    plan: &TruncateTableSnapshotPlan,
) -> PlannedSnapshotCommit<'a> {
    planned_snapshot_commit(
        plan.new_snapshot.0,
        ids.schema_version,
        ids.next_catalog_id,
        plan.next_file_id,
        truncate.changes_made.as_str(),
        truncate.commit_message.as_str(),
    )
}

#[derive(Debug, Clone, Copy)]
struct PlannedTruncateTableEffects<'a> {
    table_id: TableId,
    new_snapshot: i64,
    inlined_data_deletions: &'a [InlinedDataDeletionCommit],
}

fn planned_truncate_table_effects<'a>(
    truncate: &'a TruncateTableCommit,
    plan: &TruncateTableSnapshotPlan,
) -> PlannedTruncateTableEffects<'a> {
    PlannedTruncateTableEffects {
        table_id: truncate.table_id,
        new_snapshot: plan.new_snapshot.0,
        inlined_data_deletions: &truncate.inlined_data_deletions,
    }
}

#[derive(Debug, Clone, Copy)]
struct PlannedTruncateTableStatsReset {
    table_id: TableId,
}

fn planned_truncate_table_stats_reset(
    effects: &PlannedTruncateTableEffects<'_>,
) -> PlannedTruncateTableStatsReset {
    PlannedTruncateTableStatsReset {
        table_id: effects.table_id,
    }
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
fn apply_truncate_table_stats_sync<'tx, O>(
    ops: &O,
    tx: &O::Tx<'tx>,
    stats_reset: PlannedTruncateTableStatsReset,
) -> Result<()>
where
    O: SyncMaintenanceCommitOps,
{
    if ops.table_exists(tx, "ducklake_table_stats")? {
        ops.reset_table_stats_for_truncate(tx, stats_reset.table_id)?;
    }
    Ok(())
}

#[cfg(any(feature = "postgres", feature = "quack"))]
async fn apply_truncate_table_stats_async<O, Tx>(
    ops: &O,
    tx: &Tx,
    stats_reset: PlannedTruncateTableStatsReset,
) -> Result<()>
where
    O: AsyncMaintenanceCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    if ops.table_exists(tx, "ducklake_table_stats").await? {
        ops.reset_table_stats_for_truncate(tx, stats_reset.table_id)
            .await?;
    }
    Ok(())
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
fn apply_truncate_table_effects_sync<'tx, O>(
    ops: &O,
    tx: &O::Tx<'tx>,
    plan: &crate::TruncateTableCommitPlan,
    effects: &PlannedTruncateTableEffects<'_>,
) -> Result<()>
where
    O: SyncMaintenanceCommitOps,
{
    ops.end_active_table_data_files(tx, effects.table_id, effects.new_snapshot)?;
    ops.end_active_table_delete_files(tx, effects.table_id, effects.new_snapshot)?;
    let inlined_rows_deleted =
        ops.apply_inlined_data_deletions(tx, effects.inlined_data_deletions, effects.new_snapshot)?;
    plan.validate_inlined_rows_deleted(inlined_rows_deleted)
}

#[cfg(any(feature = "postgres", feature = "quack"))]
async fn apply_truncate_table_effects_async<O, Tx>(
    ops: &O,
    tx: &Tx,
    plan: &crate::TruncateTableCommitPlan,
    effects: &PlannedTruncateTableEffects<'_>,
) -> Result<()>
where
    O: AsyncMaintenanceCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    ops.end_active_table_data_files(tx, effects.table_id, effects.new_snapshot)
        .await?;
    ops.end_active_table_delete_files(tx, effects.table_id, effects.new_snapshot)
        .await?;
    let inlined_rows_deleted = ops
        .apply_inlined_data_deletions(tx, effects.inlined_data_deletions, effects.new_snapshot)
        .await?;
    plan.validate_inlined_rows_deleted(inlined_rows_deleted)
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
fn finish_snapshot_commit_sync<'tx, O>(
    ops: &O,
    tx: &O::Tx<'tx>,
    commit: PlannedSnapshotCommit<'_>,
) -> Result<SnapshotId>
where
    O: SyncCommitOps,
{
    ops.insert_snapshot(
        tx,
        commit.new_snapshot,
        commit.new_schema_version,
        commit.next_catalog_id,
        commit.next_file_id,
    )?;
    if ops.table_exists(tx, "ducklake_snapshot_changes")? {
        ops.insert_snapshot_change(
            tx,
            commit.new_snapshot,
            commit.changes_made.as_ref(),
            commit.commit_message.as_ref(),
        )?;
    }
    Ok(commit.snapshot_id())
}

#[cfg(any(feature = "postgres", feature = "quack"))]
async fn finish_snapshot_commit_async<O, Tx>(
    ops: &O,
    tx: &Tx,
    commit: PlannedSnapshotCommit<'_>,
) -> Result<SnapshotId>
where
    O: AsyncCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    ops.insert_snapshot(
        tx,
        commit.new_snapshot,
        commit.new_schema_version,
        commit.next_catalog_id,
        commit.next_file_id,
    )
    .await?;
    if ops.table_exists(tx, "ducklake_snapshot_changes").await? {
        ops.insert_snapshot_change(
            tx,
            commit.new_snapshot,
            commit.changes_made.as_ref(),
            commit.commit_message.as_ref(),
        )
        .await?;
    }
    Ok(commit.snapshot_id())
}

fn first_i64_or_zero(values: Vec<i64>) -> i64 {
    values.into_iter().next().unwrap_or(0)
}

fn any_count(values: Vec<i64>) -> bool {
    first_i64_or_zero(values) > 0
}

fn first_string(values: Vec<String>) -> Option<String> {
    values.into_iter().next()
}

fn close_table_contents_queries(
    table_id: TableId,
    new_snapshot: i64,
) -> [crate::catalog_queries::CatalogQuery; 3] {
    [
        crate::catalog_queries::close_table_columns(table_id, new_snapshot),
        crate::catalog_queries::end_active_table_data_files(table_id, new_snapshot),
        crate::catalog_queries::end_active_table_delete_files(table_id, new_snapshot),
    ]
}

fn active_partition_spec_from_rows(
    partition: ducklake_catalog::PartitionInfo,
    columns: Vec<ducklake_catalog::PartitionColumn>,
) -> ActivePartitionSpec {
    ActivePartitionSpec {
        partition_id: partition.partition_id,
        columns: columns
            .into_iter()
            .map(|column| PartitionColumnCommit {
                column_id: column.column_id,
                transform: column.transform,
            })
            .collect(),
    }
}

fn active_sort_spec_from_rows(
    sort: ducklake_catalog::SortInfo,
    expressions: Vec<ducklake_catalog::SortExpression>,
) -> ActiveSortSpec {
    ActiveSortSpec {
        sort_id: sort.sort_id,
        expressions: expressions
            .into_iter()
            .map(sort_expression_commit)
            .collect(),
    }
}

#[derive(Debug, Clone, Copy)]
pub(crate) struct DataFileMetadataTables {
    file_column_stats: bool,
    file_variant_stats: bool,
}

impl DataFileMetadataTables {
    fn new(file_column_stats: bool, file_variant_stats: bool) -> Self {
        Self {
            file_column_stats,
            file_variant_stats,
        }
    }
}

#[derive(Debug, Clone, Copy)]
pub(crate) struct TableDataMetadataTables {
    data_file: DataFileMetadataTables,
    table_stats: bool,
}

impl TableDataMetadataTables {
    fn new(file_column_stats: bool, file_variant_stats: bool, table_stats: bool) -> Self {
        Self {
            data_file: DataFileMetadataTables::new(file_column_stats, file_variant_stats),
            table_stats,
        }
    }
}

#[derive(Debug, Clone, Copy)]
struct PlannedTableColumnStatsAppendUpdate<'a> {
    table_id: TableId,
    snapshot: SnapshotId,
    files: &'a [AppendFile],
}

fn planned_table_column_stats_append_update<'a>(
    table_id: TableId,
    snapshot: SnapshotId,
    files: &'a [AppendFile],
) -> PlannedTableColumnStatsAppendUpdate<'a> {
    PlannedTableColumnStatsAppendUpdate {
        table_id,
        snapshot,
        files,
    }
}

fn inlined_column_stats_append_file(
    table_id: TableId,
    expected_snapshot: SnapshotId,
    rows_appended: i64,
    column_stats: &[ducklake_catalog::FileColumnStats],
) -> Option<AppendFile> {
    if column_stats.is_empty() {
        return None;
    }
    Some(AppendFile {
        table_id,
        expected_snapshot,
        data_file_id: ducklake_catalog::DataFileId(0),
        begin_snapshot: None,
        path: String::new(),
        path_is_relative: true,
        record_count: rows_appended,
        file_size_bytes: 0,
        footer_size: None,
        row_id_start: None,
        explicit_row_ids: false,
        column_stats: column_stats.to_vec(),
        variant_stats: Vec::new(),
        partition_id: None,
        partition_values: Vec::new(),
        encryption_key: None,
        mapping_id: None,
        partial_max: None,
    })
}

fn inlined_append_column_stats_files(
    table_id: TableId,
    expected_snapshot: SnapshotId,
    appends: impl IntoIterator<Item = (i64, Vec<ducklake_catalog::FileColumnStats>)>,
) -> Vec<AppendFile> {
    appends
        .into_iter()
        .filter_map(|(rows_appended, column_stats)| {
            inlined_column_stats_append_file(
                table_id,
                expected_snapshot,
                rows_appended,
                &column_stats,
            )
        })
        .collect()
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
fn apply_table_column_stats_append_sync<'tx, O>(
    ops: &O,
    tx: &O::Tx<'tx>,
    stats: PlannedTableColumnStatsAppendUpdate<'_>,
) -> Result<()>
where
    O: SyncTableDataCommitOps,
{
    ops.update_table_column_stats_for_append(tx, stats.table_id, stats.snapshot, stats.files)
}

#[cfg(any(feature = "postgres", feature = "quack"))]
async fn apply_table_column_stats_append_async<O, Tx>(
    ops: &O,
    tx: &Tx,
    stats: PlannedTableColumnStatsAppendUpdate<'_>,
) -> Result<()>
where
    O: AsyncTableDataCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    ops.update_table_column_stats_for_append(tx, stats.table_id, stats.snapshot, stats.files)
        .await
}

fn insert_data_file_commands(
    file: &AppendFile,
    row_id_start: Option<i64>,
    begin_snapshot: i64,
    metadata_tables: DataFileMetadataTables,
) -> Vec<crate::catalog_queries::CatalogQuery> {
    let stats_capacity = if metadata_tables.file_column_stats {
        file.column_stats.len()
    } else {
        0
    } + if metadata_tables.file_variant_stats {
        file.variant_stats.len()
    } else {
        0
    };
    let mut commands = Vec::with_capacity(1 + file.partition_values.len() + stats_capacity);
    commands.push(crate::catalog_queries::insert_data_file(
        file,
        row_id_start,
        begin_snapshot,
    ));
    commands.extend(
        file.partition_values
            .iter()
            .map(crate::catalog_queries::insert_file_partition_value),
    );
    if metadata_tables.file_column_stats {
        commands.extend(
            file.column_stats
                .iter()
                .map(crate::catalog_queries::insert_file_column_stats),
        );
    }
    if metadata_tables.file_variant_stats {
        commands.extend(
            file.variant_stats
                .iter()
                .map(crate::catalog_queries::insert_file_variant_stats),
        );
    }
    commands
}

#[derive(Debug, Clone, Copy)]
struct PlannedDataFileInsert<'a> {
    file: &'a AppendFile,
    row_id_start: Option<i64>,
    begin_snapshot: i64,
    metadata_tables: DataFileMetadataTables,
}

trait PlannedDataFileRow {
    fn row_id_start(&self) -> Option<i64>;

    fn begin_snapshot(&self) -> i64;
}

impl PlannedDataFileRow for crate::AppendFilePlan {
    fn row_id_start(&self) -> Option<i64> {
        self.row_id_start
    }

    fn begin_snapshot(&self) -> i64 {
        self.begin_snapshot
    }
}

impl PlannedDataFileRow for crate::PlannedCreateTableWithFilesFile {
    fn row_id_start(&self) -> Option<i64> {
        self.row_id_start
    }

    fn begin_snapshot(&self) -> i64 {
        self.begin_snapshot
    }
}

fn planned_data_file_inserts<'a>(
    files: &'a [AppendFile],
    plans: &'a [impl PlannedDataFileRow],
    metadata_tables: DataFileMetadataTables,
    operation: &'static str,
) -> Result<Vec<PlannedDataFileInsert<'a>>> {
    if files.len() != plans.len() {
        return Err(StorageError::Backend(format!(
            "DuckLake {operation} planned {} data-file rows for {} input files",
            plans.len(),
            files.len()
        )));
    }
    Ok(files
        .iter()
        .zip(plans.iter())
        .map(|(file, plan)| PlannedDataFileInsert {
            file,
            row_id_start: plan.row_id_start(),
            begin_snapshot: plan.begin_snapshot(),
            metadata_tables,
        })
        .collect())
}

fn insert_delete_file_command(
    file: &DeleteFileCommit,
    begin_snapshot: i64,
    format: &str,
) -> crate::catalog_queries::CatalogQuery {
    crate::catalog_queries::insert_delete_file(file, begin_snapshot, format)
}

#[derive(Debug, Clone, Copy)]
struct PlannedDeleteFileInsert {
    begin_snapshot: i64,
    format: &'static str,
}

#[derive(Debug, Clone, Copy)]
struct PlannedDeleteFileWrite<'a> {
    file: &'a DeleteFileCommit,
    new_snapshot: i64,
    insert: PlannedDeleteFileInsert,
}

fn begin_snapshot_or(snapshot: Option<SnapshotId>, new_snapshot: i64) -> i64 {
    snapshot.map(|snapshot| snapshot.0).unwrap_or(new_snapshot)
}

fn planned_delete_file_insert(
    file: &DeleteFileCommit,
    new_snapshot: i64,
) -> Result<PlannedDeleteFileInsert> {
    Ok(PlannedDeleteFileInsert {
        begin_snapshot: begin_snapshot_or(file.begin_snapshot, new_snapshot),
        format: delete_format_label(&file.format)?,
    })
}

fn planned_delete_file_writes(
    files: &[DeleteFileCommit],
    new_snapshot: i64,
) -> Result<Vec<PlannedDeleteFileWrite<'_>>> {
    files
        .iter()
        .map(|file| {
            Ok(PlannedDeleteFileWrite {
                file,
                new_snapshot,
                insert: planned_delete_file_insert(file, new_snapshot)?,
            })
        })
        .collect()
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
fn insert_planned_data_files_sync<'tx, O, P>(
    ops: &O,
    tx: &O::Tx<'tx>,
    files: &[AppendFile],
    plans: &[P],
    metadata_tables: DataFileMetadataTables,
    operation: &'static str,
) -> Result<()>
where
    O: SyncTableDataCommitOps,
    P: PlannedDataFileRow,
{
    for planned_file in planned_data_file_inserts(files, plans, metadata_tables, operation)? {
        ops.insert_data_file(
            tx,
            planned_file.file,
            planned_file.row_id_start,
            planned_file.begin_snapshot,
            planned_file.metadata_tables,
        )?;
    }
    Ok(())
}

#[cfg(any(feature = "postgres", feature = "quack"))]
async fn insert_planned_data_files_async<O, Tx, P>(
    ops: &O,
    tx: &Tx,
    files: &[AppendFile],
    plans: &[P],
    metadata_tables: DataFileMetadataTables,
    operation: &'static str,
) -> Result<()>
where
    O: AsyncTableDataCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
    P: PlannedDataFileRow + Sync,
{
    for planned_file in planned_data_file_inserts(files, plans, metadata_tables, operation)? {
        ops.insert_data_file(
            tx,
            planned_file.file,
            planned_file.row_id_start,
            planned_file.begin_snapshot,
            planned_file.metadata_tables,
        )
        .await?;
    }
    Ok(())
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
fn insert_planned_delete_files_sync<'tx, O>(
    ops: &O,
    tx: &O::Tx<'tx>,
    files: &[DeleteFileCommit],
    new_snapshot: i64,
) -> Result<()>
where
    O: SyncTableDataCommitOps,
{
    for planned_file in planned_delete_file_writes(files, new_snapshot)? {
        ops.replace_overwritten_delete_file(tx, planned_file.file, planned_file.new_snapshot)?;
        ops.insert_delete_file(
            tx,
            planned_file.file,
            planned_file.insert.begin_snapshot,
            planned_file.insert.format,
        )?;
    }
    Ok(())
}

#[cfg(any(feature = "postgres", feature = "quack"))]
async fn insert_planned_delete_files_async<O, Tx>(
    ops: &O,
    tx: &Tx,
    files: &[DeleteFileCommit],
    new_snapshot: i64,
) -> Result<()>
where
    O: AsyncTableDataCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    for planned_file in planned_delete_file_writes(files, new_snapshot)? {
        ops.replace_overwritten_delete_file(tx, planned_file.file, planned_file.new_snapshot)
            .await?;
        ops.insert_delete_file(
            tx,
            planned_file.file,
            planned_file.insert.begin_snapshot,
            planned_file.insert.format,
        )
        .await?;
    }
    Ok(())
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
fn insert_planned_inlined_batches_sync<'tx, O>(
    ops: &O,
    tx: &O::Tx<'tx>,
    batch_inserts: &[PlannedInlinedBatchInsert<'_>],
) -> Result<()>
where
    O: SyncTableDataCommitOps,
{
    for batch_insert in batch_inserts {
        let inlined_table_name = ops.ensure_inlined_data_table(
            tx,
            batch_insert.table_id,
            batch_insert.schema_version,
            batch_insert.schema,
        )?;
        ops.insert_inlined_batches(
            tx,
            &inlined_table_name,
            batch_insert.schema,
            batch_insert.batches,
            batch_insert.preserved_row_ids,
            batch_insert.row_id_start,
            batch_insert.new_snapshot,
        )?;
    }
    Ok(())
}

#[cfg(any(feature = "postgres", feature = "quack"))]
async fn insert_planned_inlined_batches_async<O, Tx>(
    ops: &O,
    tx: &Tx,
    batch_inserts: &[PlannedInlinedBatchInsert<'_>],
) -> Result<()>
where
    O: AsyncTableDataCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    for batch_insert in batch_inserts {
        let inlined_table_name = ops
            .ensure_inlined_data_table(
                tx,
                batch_insert.table_id,
                batch_insert.schema_version,
                batch_insert.schema,
            )
            .await?;
        ops.insert_inlined_batches(
            tx,
            &inlined_table_name,
            batch_insert.schema,
            batch_insert.batches,
            batch_insert.preserved_row_ids,
            batch_insert.row_id_start,
            batch_insert.new_snapshot,
        )
        .await?;
    }
    Ok(())
}

#[derive(Debug, Clone)]
pub(crate) struct PlannedSchemaRow {
    schema_id: i64,
    schema_uuid: String,
    new_snapshot: i64,
    schema_name: String,
}

fn planned_schema_row(
    schema_id: i64,
    new_snapshot: i64,
    create: &CreateSchemaCommit,
) -> PlannedSchemaRow {
    PlannedSchemaRow {
        schema_id,
        schema_uuid: crate::deterministic_catalog_uuid(schema_id),
        new_snapshot,
        schema_name: create.schema_name.clone(),
    }
}

#[derive(Debug, Clone)]
pub(crate) struct PlannedViewRow<'a> {
    view_id: i64,
    view_uuid: String,
    new_snapshot: i64,
    schema_id: SchemaId,
    create: &'a CreateViewCommit,
}

fn planned_view_row<'a>(
    view_id: i64,
    new_snapshot: i64,
    schema_id: SchemaId,
    create: &'a CreateViewCommit,
) -> PlannedViewRow<'a> {
    PlannedViewRow {
        view_id,
        view_uuid: crate::deterministic_catalog_uuid(view_id),
        new_snapshot,
        schema_id,
        create,
    }
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
fn apply_create_schema_sync<'tx, O>(
    ops: &O,
    tx: &O::Tx<'tx>,
    schema: &PlannedSchemaRow,
) -> Result<()>
where
    O: SyncCatalogObjectCommitOps,
{
    ops.insert_schema(tx, schema)
}

#[cfg(any(feature = "postgres", feature = "quack"))]
async fn apply_create_schema_async<O, Tx>(ops: &O, tx: &Tx, schema: &PlannedSchemaRow) -> Result<()>
where
    O: AsyncCatalogObjectCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    ops.insert_schema(tx, schema).await
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
fn apply_create_view_sync<'tx, O>(ops: &O, tx: &O::Tx<'tx>, view: &PlannedViewRow<'_>) -> Result<()>
where
    O: SyncCatalogObjectCommitOps,
{
    ops.insert_view(tx, view)
}

#[cfg(any(feature = "postgres", feature = "quack"))]
async fn apply_create_view_async<O, Tx>(ops: &O, tx: &Tx, view: &PlannedViewRow<'_>) -> Result<()>
where
    O: AsyncCatalogObjectCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    ops.insert_view(tx, view).await
}

#[derive(Debug, Clone)]
struct PlannedTableRow {
    table_id: i64,
    table_uuid: String,
    new_snapshot: i64,
    schema_id: SchemaId,
    table_name: String,
    table_path: Option<String>,
    table_path_is_relative: bool,
}

#[derive(Debug, Clone)]
struct PlannedTableInsert<'a> {
    row: PlannedTableRow,
    columns: &'a [CreateTableColumn],
}

fn planned_create_table_row(
    table_id: i64,
    new_snapshot: i64,
    schema_id: SchemaId,
    create: &CreateTableCommit,
) -> PlannedTableRow {
    PlannedTableRow {
        table_id,
        table_uuid: crate::deterministic_catalog_uuid(table_id),
        new_snapshot,
        schema_id,
        table_name: create.table_name.clone(),
        table_path: Some(format!("{}/", create.table_name)),
        table_path_is_relative: true,
    }
}

fn planned_table_insert<'a>(
    table_id: i64,
    new_snapshot: i64,
    schema_id: SchemaId,
    create: &'a CreateTableCommit,
) -> PlannedTableInsert<'a> {
    PlannedTableInsert {
        row: planned_create_table_row(table_id, new_snapshot, schema_id, create),
        columns: &create.columns,
    }
}

fn planned_renamed_table_row(
    data_path: &str,
    source_schema: &SchemaEntry,
    target_schema: &SchemaEntry,
    table: &TableEntry,
    rename: &RenameTableCommit,
    new_snapshot: i64,
) -> PlannedTableRow {
    let (table_path, table_path_is_relative) =
        renamed_table_path_for_target_schema(data_path, source_schema, target_schema, table);
    PlannedTableRow {
        table_id: table.table_id.0,
        table_uuid: table.table_uuid.clone(),
        new_snapshot,
        schema_id: target_schema.schema_id,
        table_name: rename.new_table_name.clone(),
        table_path,
        table_path_is_relative,
    }
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
fn insert_planned_table_row_sync<'tx, O>(
    ops: &O,
    tx: &O::Tx<'tx>,
    table: &PlannedTableRow,
) -> Result<()>
where
    O: SyncCatalogObjectCommitOps,
{
    ops.insert_table(
        tx,
        table.table_id,
        &table.table_uuid,
        table.new_snapshot,
        table.schema_id,
        &table.table_name,
        table.table_path.as_deref(),
        table.table_path_is_relative,
    )
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
fn insert_planned_table_sync<'tx, O>(
    ops: &O,
    tx: &O::Tx<'tx>,
    table: &PlannedTableInsert<'_>,
) -> Result<()>
where
    O: SyncCatalogObjectCommitOps,
{
    insert_planned_table_row_sync(ops, tx, &table.row)?;
    ops.insert_table_columns(
        tx,
        table.row.table_id,
        table.row.new_snapshot,
        table.columns,
    )
}

#[cfg(any(feature = "postgres", feature = "quack"))]
async fn insert_planned_table_row_async<O, Tx>(
    ops: &O,
    tx: &Tx,
    table: &PlannedTableRow,
) -> Result<()>
where
    O: AsyncCatalogObjectCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    ops.insert_table(
        tx,
        table.table_id,
        &table.table_uuid,
        table.new_snapshot,
        table.schema_id,
        &table.table_name,
        table.table_path.as_deref(),
        table.table_path_is_relative,
    )
    .await
}

#[cfg(any(feature = "postgres", feature = "quack"))]
async fn insert_planned_table_async<O, Tx>(
    ops: &O,
    tx: &Tx,
    table: &PlannedTableInsert<'_>,
) -> Result<()>
where
    O: AsyncCatalogObjectCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    insert_planned_table_row_async(ops, tx, &table.row).await?;
    ops.insert_table_columns(
        tx,
        table.row.table_id,
        table.row.new_snapshot,
        table.columns,
    )
    .await
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
fn apply_create_table_sync<'tx, O>(
    ops: &O,
    tx: &O::Tx<'tx>,
    table: &PlannedTableInsert<'_>,
) -> Result<()>
where
    O: SyncCatalogObjectCommitOps,
{
    insert_planned_table_sync(ops, tx, table)
}

#[cfg(any(feature = "postgres", feature = "quack"))]
async fn apply_create_table_async<O, Tx>(
    ops: &O,
    tx: &Tx,
    table: &PlannedTableInsert<'_>,
) -> Result<()>
where
    O: AsyncCatalogObjectCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    insert_planned_table_async(ops, tx, table).await
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
fn apply_drop_table_sync<'tx, O>(
    ops: &O,
    tx: &O::Tx<'tx>,
    table_id: TableId,
    new_snapshot: i64,
) -> Result<()>
where
    O: SyncCatalogObjectCommitOps,
{
    ops.close_table(tx, table_id, new_snapshot)?;
    ops.close_table_contents(tx, table_id, new_snapshot)
}

#[cfg(any(feature = "postgres", feature = "quack"))]
async fn apply_drop_table_async<O, Tx>(
    ops: &O,
    tx: &Tx,
    table_id: TableId,
    new_snapshot: i64,
) -> Result<()>
where
    O: AsyncCatalogObjectCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    ops.close_table(tx, table_id, new_snapshot).await?;
    ops.close_table_contents(tx, table_id, new_snapshot).await
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
fn apply_rename_table_sync<'tx, O>(
    ops: &O,
    tx: &O::Tx<'tx>,
    table_id: TableId,
    new_snapshot: i64,
    renamed_table: &PlannedTableRow,
) -> Result<()>
where
    O: SyncCatalogObjectCommitOps,
{
    let closed = ops.close_table(tx, table_id, new_snapshot)?;
    ensure_closed_table_for_rename(table_id, closed)?;
    insert_planned_table_row_sync(ops, tx, renamed_table)
}

#[cfg(any(feature = "postgres", feature = "quack"))]
async fn apply_rename_table_async<O, Tx>(
    ops: &O,
    tx: &Tx,
    table_id: TableId,
    new_snapshot: i64,
    renamed_table: &PlannedTableRow,
) -> Result<()>
where
    O: AsyncCatalogObjectCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    let closed = ops.close_table(tx, table_id, new_snapshot).await?;
    ensure_closed_table_for_rename(table_id, closed)?;
    insert_planned_table_row_async(ops, tx, renamed_table).await
}

fn ensure_closed_table_for_rename(table_id: TableId, closed: usize) -> Result<()> {
    if closed != 1 {
        return Err(StorageError::Backend(format!(
            "DuckLake RENAME TABLE expected one active table row for table {}, updated {}",
            table_id.0, closed
        )));
    }
    Ok(())
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
fn apply_drop_view_sync<'tx, O>(
    ops: &O,
    tx: &O::Tx<'tx>,
    view_id: ViewId,
    new_snapshot: i64,
) -> Result<()>
where
    O: SyncCatalogObjectCommitOps,
{
    ops.close_view(tx, view_id, new_snapshot)
}

#[cfg(any(feature = "postgres", feature = "quack"))]
async fn apply_drop_view_async<O, Tx>(
    ops: &O,
    tx: &Tx,
    view_id: ViewId,
    new_snapshot: i64,
) -> Result<()>
where
    O: AsyncCatalogObjectCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    ops.close_view(tx, view_id, new_snapshot).await
}

#[derive(Debug, Clone)]
struct PlannedDropSchema {
    plan: crate::CatalogObjectMutationPlan,
    schema_id: SchemaId,
    snapshot: SnapshotId,
    table_ids: Vec<i64>,
    close_views: bool,
}

fn planned_drop_schema(
    drop: &DropSchemaCommit,
    ids: CommitSnapshotIds,
    schema_id: SchemaId,
    snapshot: SnapshotId,
    table_ids: Vec<i64>,
    view_ids: &[i64],
    close_views: bool,
) -> Result<PlannedDropSchema> {
    let plan =
        plan_drop_schema_catalog_mutation(drop, catalog_ids(ids), schema_id, &table_ids, view_ids)?;
    Ok(PlannedDropSchema {
        plan,
        schema_id,
        snapshot,
        table_ids,
        close_views,
    })
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
fn apply_planned_drop_schema_sync<'tx, O>(
    ops: &O,
    tx: &O::Tx<'tx>,
    planned: &PlannedDropSchema,
) -> Result<()>
where
    O: SyncCatalogObjectCommitOps,
{
    ops.close_schema(tx, planned.schema_id, planned.plan.new_snapshot)?;
    ops.close_tables_for_schema(
        tx,
        planned.schema_id,
        planned.snapshot,
        planned.plan.new_snapshot,
    )?;
    if planned.close_views {
        ops.close_views_for_schema(
            tx,
            planned.schema_id,
            planned.snapshot,
            planned.plan.new_snapshot,
        )?;
    }
    for table_id in &planned.table_ids {
        ops.close_table_contents(tx, TableId(*table_id), planned.plan.new_snapshot)?;
    }
    Ok(())
}

#[cfg(any(feature = "postgres", feature = "quack"))]
async fn apply_planned_drop_schema_async<O, Tx>(
    ops: &O,
    tx: &Tx,
    planned: &PlannedDropSchema,
) -> Result<()>
where
    O: AsyncCatalogObjectCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    ops.close_schema(tx, planned.schema_id, planned.plan.new_snapshot)
        .await?;
    ops.close_tables_for_schema(
        tx,
        planned.schema_id,
        planned.snapshot,
        planned.plan.new_snapshot,
    )
    .await?;
    if planned.close_views {
        ops.close_views_for_schema(
            tx,
            planned.schema_id,
            planned.snapshot,
            planned.plan.new_snapshot,
        )
        .await?;
    }
    for table_id in &planned.table_ids {
        ops.close_table_contents(tx, TableId(*table_id), planned.plan.new_snapshot)
            .await?;
    }
    Ok(())
}

fn sort_expressions_reference_column(expressions: Vec<String>, column_name: &str) -> Result<bool> {
    for expression in expressions {
        if sort_expression_references_unqualified_column(&expression, column_name)? {
            return Ok(true);
        }
    }
    Ok(false)
}

fn validate_set_not_null_stats_from_rows(
    column: &ColumnEntry,
    table_stats_rows: Vec<i64>,
    contains_null_rows: Option<Vec<Option<bool>>>,
) -> Result<()> {
    let contains_null = contains_null_rows
        .and_then(|rows| rows.into_iter().next())
        .flatten();
    crate::catalog_queries::validate_set_not_null_stats_result(
        column,
        !table_stats_rows.is_empty(),
        contains_null,
    )
}

fn validate_set_not_null_without_table_stats(column: &ColumnEntry) -> Result<()> {
    crate::catalog_queries::validate_set_not_null_stats_result(column, false, None)
}

fn set_not_null_table_stats_query(column: &ColumnEntry) -> crate::catalog_queries::CatalogQuery {
    crate::catalog_queries::table_stats_present(column.table_id)
}

fn set_not_null_contains_null_query(column: &ColumnEntry) -> crate::catalog_queries::CatalogQuery {
    crate::catalog_queries::table_column_contains_null(column.table_id, column.column_id)
}

fn table_column_stats_count_query(
    stats: &TableColumnStats,
) -> crate::catalog_queries::CatalogQuery {
    crate::catalog_queries::table_column_stats_count(stats.table_id, stats.column_id)
}

fn table_column_stats_upsert_query(
    stats: &TableColumnStats,
    existing_rows: Vec<i64>,
) -> crate::catalog_queries::CatalogQuery {
    if first_i64_or_zero(existing_rows) > 0 {
        crate::catalog_queries::update_table_column_stats(stats)
    } else {
        crate::catalog_queries::insert_table_column_stats(stats)
    }
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) fn replace_overwritten_delete_file_sync<
    CleanupRow,
    CloseActive,
    LoadCleanup,
    DeleteOverwritten,
    ScheduleCleanup,
>(
    file: &DeleteFileCommit,
    new_snapshot: i64,
    mut close_active: CloseActive,
    mut load_cleanup: LoadCleanup,
    mut delete_overwritten: DeleteOverwritten,
    mut schedule_cleanup: ScheduleCleanup,
) -> Result<()>
where
    CloseActive: FnMut(&DeleteFileCommit, i64) -> Result<usize>,
    LoadCleanup: FnMut(DeleteFileId, TableId) -> Result<CleanupRow>,
    DeleteOverwritten: FnMut(&DeleteFileCommit) -> Result<usize>,
    ScheduleCleanup: FnMut(CleanupRow) -> Result<()>,
{
    let Some(overwritten) = file.overwritten_delete_file_id else {
        return Ok(());
    };
    if file.begin_snapshot.is_none() {
        let updated = close_active(file, new_snapshot)?;
        if updated != 1 {
            return Err(StorageError::Backend(format!(
                "DuckLake overwrite expected one active delete file {}, updated {} rows",
                overwritten.0, updated
            )));
        }
        return Ok(());
    }

    let cleanup_row = load_cleanup(overwritten, file.table_id)?;
    let updated = delete_overwritten(file)?;
    if updated != 1 {
        return Err(StorageError::Backend(format!(
            "DuckLake overwrite expected one delete file {}, deleted {} rows",
            overwritten.0, updated
        )));
    }
    schedule_cleanup(cleanup_row)
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) trait SyncCommitOps {
    type Tx<'tx>;

    fn latest_snapshot_ids(&self, tx: &Self::Tx<'_>) -> Result<CommitSnapshotIds>;

    fn table_exists(&self, tx: &Self::Tx<'_>, table_name: &str) -> Result<bool>;

    fn insert_snapshot(
        &self,
        tx: &Self::Tx<'_>,
        snapshot_id: i64,
        schema_version: i64,
        next_catalog_id: i64,
        next_file_id: i64,
    ) -> Result<()>;

    fn insert_snapshot_change(
        &self,
        tx: &Self::Tx<'_>,
        snapshot_id: i64,
        changes_made: &str,
        commit_message: &str,
    ) -> Result<()>;
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) trait SyncTableDataCommitOps: SyncCommitOps {
    fn table_data_execute(
        &self,
        tx: &Self::Tx<'_>,
        query: &crate::catalog_queries::CatalogQuery,
    ) -> Result<usize>;

    fn next_row_id_for_append(
        &self,
        tx: &Self::Tx<'_>,
        table_id: TableId,
        has_table_stats: bool,
    ) -> Result<Option<i64>>;

    fn insert_data_file(
        &self,
        tx: &Self::Tx<'_>,
        file: &AppendFile,
        row_id_start: Option<i64>,
        begin_snapshot: i64,
        metadata_tables: DataFileMetadataTables,
    ) -> Result<()> {
        for command in
            insert_data_file_commands(file, row_id_start, begin_snapshot, metadata_tables)
        {
            self.table_data_execute(tx, &command)?;
        }

        Ok(())
    }

    fn update_table_column_stats_for_append(
        &self,
        tx: &Self::Tx<'_>,
        table_id: TableId,
        latest_snapshot: SnapshotId,
        files: &[AppendFile],
    ) -> Result<()>;

    fn update_table_stats_for_append(
        &self,
        tx: &Self::Tx<'_>,
        table_id: TableId,
        record_count: i64,
        allocated_row_id_count: i64,
        file_size_bytes: i64,
    ) -> Result<usize> {
        self.table_data_execute(
            tx,
            &crate::catalog_queries::update_table_stats_for_append(
                table_id,
                record_count,
                allocated_row_id_count,
                file_size_bytes,
            ),
        )
    }

    fn insert_table_stats_from_active_files(
        &self,
        _tx: &Self::Tx<'_>,
        _table_id: TableId,
    ) -> Result<()> {
        Ok(())
    }

    fn validate_inlined_append(
        &self,
        rows_appended: i64,
        schema: &TableSchema,
        batches: &[RecordBatch],
        preserved_row_ids: Option<&[i64]>,
        context: &str,
    ) -> Result<()>;

    fn ensure_inlined_data_table(
        &self,
        tx: &Self::Tx<'_>,
        table_id: TableId,
        schema_version: i64,
        schema: &TableSchema,
    ) -> Result<String>;

    fn insert_inlined_batches(
        &self,
        tx: &Self::Tx<'_>,
        table_name: &str,
        schema: &TableSchema,
        batches: &[RecordBatch],
        preserved_row_ids: Option<&[i64]>,
        row_id_start: i64,
        begin_snapshot: i64,
    ) -> Result<()>;

    fn update_table_stats_for_inlined_append(
        &self,
        tx: &Self::Tx<'_>,
        table_id: TableId,
        rows_appended: i64,
    ) -> Result<usize> {
        self.table_data_execute(
            tx,
            &crate::catalog_queries::update_table_stats_for_inlined_append(table_id, rows_appended),
        )
    }

    fn insert_table_stats_for_inlined_append(
        &self,
        tx: &Self::Tx<'_>,
        table_id: TableId,
        rows_appended: i64,
        next_row_id: i64,
    ) -> Result<()> {
        self.table_data_execute(
            tx,
            &crate::catalog_queries::insert_table_stats_for_inlined_append(
                table_id,
                rows_appended,
                next_row_id,
            ),
        )?;
        Ok(())
    }

    fn replace_overwritten_delete_file(
        &self,
        tx: &Self::Tx<'_>,
        file: &DeleteFileCommit,
        new_snapshot: i64,
    ) -> Result<()>;

    fn insert_delete_file(
        &self,
        tx: &Self::Tx<'_>,
        file: &DeleteFileCommit,
        begin_snapshot: i64,
        format: &str,
    ) -> Result<()> {
        self.table_data_execute(
            tx,
            &insert_delete_file_command(file, begin_snapshot, format),
        )?;
        Ok(())
    }

    fn update_table_stats_for_delete(
        &self,
        tx: &Self::Tx<'_>,
        table_id: TableId,
        deleted_rows: i64,
    ) -> Result<usize> {
        self.table_data_execute(
            tx,
            &crate::catalog_queries::update_table_stats_for_delete(table_id, deleted_rows),
        )
    }

    fn write_inlined_file_deletions(
        &self,
        tx: &Self::Tx<'_>,
        table_id: TableId,
        deletions: &[InlinedFileDeletionCommit],
        new_snapshot: i64,
    ) -> Result<()>;

    fn apply_inlined_data_deletions(
        &self,
        tx: &Self::Tx<'_>,
        deletions: &[InlinedDataDeletionCommit],
        new_snapshot: i64,
    ) -> Result<i64>;

    fn delete_flushed_inlined_data(
        &self,
        tx: &Self::Tx<'_>,
        flushed_tables: &[FlushedInlinedDataCommit],
        flush_snapshot: i64,
    ) -> Result<i64>;

    fn delete_flushed_inlined_file_deletions(
        &self,
        tx: &Self::Tx<'_>,
        table_id: TableId,
        flushed_deletions: &[FlushedInlinedFileDeletionCommit],
    ) -> Result<()>;

    fn update_table_stats_for_table_change(
        &self,
        tx: &Self::Tx<'_>,
        table_id: TableId,
        appended_rows: i64,
        deleted_rows: i64,
        appended_file_size_bytes: i64,
        allocated_row_id_count: i64,
    ) -> Result<usize> {
        self.table_data_execute(
            tx,
            &crate::catalog_queries::update_table_stats_for_table_change(
                table_id,
                appended_rows,
                deleted_rows,
                appended_file_size_bytes,
                allocated_row_id_count,
            ),
        )
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct MaintenanceSourceFile {
    pub(crate) id: i64,
    pub(crate) cleanup_path: String,
    pub(crate) cleanup_path_is_relative: bool,
    pub(crate) file_size_bytes: i64,
}

impl MaintenanceSourceFile {
    fn cleanup_file(&self) -> CleanupMetadataFile {
        CleanupMetadataFile {
            id: self.id,
            path: self.cleanup_path.clone(),
            path_is_relative: self.cleanup_path_is_relative,
        }
    }
}

#[derive(Debug, Clone)]
struct PlannedMaintenanceCleanup {
    table_id: TableId,
    source_ids: Vec<i64>,
    source_cleanup_rows: Vec<CleanupMetadataFile>,
}

fn planned_maintenance_cleanup(
    table_id: TableId,
    source_ids: &[i64],
    source_files: &[MaintenanceSourceFile],
) -> PlannedMaintenanceCleanup {
    PlannedMaintenanceCleanup {
        table_id,
        source_ids: source_ids.to_vec(),
        source_cleanup_rows: source_files
            .iter()
            .map(MaintenanceSourceFile::cleanup_file)
            .collect(),
    }
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
fn apply_maintenance_cleanup_sync<'tx, O>(
    ops: &O,
    tx: &O::Tx<'tx>,
    snapshot_id: i64,
    cleanup: &PlannedMaintenanceCleanup,
) -> Result<()>
where
    O: SyncMaintenanceCommitOps,
{
    let delete_cleanup_rows = ops.delete_file_cleanup_rows(tx, snapshot_id, &cleanup.source_ids)?;
    ops.schedule_cleanup_files(tx, &cleanup.source_cleanup_rows)?;
    ops.schedule_cleanup_files(tx, &delete_cleanup_rows)?;
    ops.delete_data_file_metadata(tx, &cleanup.source_ids)?;
    ops.delete_delete_files_by_data_file_id(tx, &cleanup.source_ids)?;
    ops.delete_inlined_file_deletions_for_files(tx, cleanup.table_id, &cleanup.source_ids)?;
    Ok(())
}

#[cfg(any(feature = "postgres", feature = "quack"))]
async fn apply_maintenance_cleanup_async<O, Tx>(
    ops: &O,
    tx: &Tx,
    snapshot_id: i64,
    cleanup: &PlannedMaintenanceCleanup,
) -> Result<()>
where
    O: AsyncMaintenanceCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    let delete_cleanup_rows = ops
        .delete_file_cleanup_rows(tx, snapshot_id, &cleanup.source_ids)
        .await?;
    ops.schedule_cleanup_files(tx, &cleanup.source_cleanup_rows)
        .await?;
    ops.schedule_cleanup_files(tx, &delete_cleanup_rows).await?;
    ops.delete_data_file_metadata(tx, &cleanup.source_ids)
        .await?;
    ops.delete_delete_files_by_data_file_id(tx, &cleanup.source_ids)
        .await?;
    ops.delete_inlined_file_deletions_for_files(tx, cleanup.table_id, &cleanup.source_ids)
        .await?;
    Ok(())
}

#[derive(Debug, Clone, Copy)]
struct PlannedMaintenanceOutputFile<'a> {
    file: &'a AppendFile,
    new_snapshot: i64,
}

#[derive(Debug, Clone, Copy)]
struct PlannedMaintenanceTableStatsUpdate {
    table_id: TableId,
    source_file_size_bytes: i64,
    output_file_size_bytes: i64,
}

#[derive(Debug, Clone)]
struct PlannedMaintenanceFinish<'a> {
    table_id: TableId,
    new_snapshot: i64,
    output_files: Vec<PlannedMaintenanceOutputFile<'a>>,
    table_stats: PlannedMaintenanceTableStatsUpdate,
}

fn planned_maintenance_finish<'a>(
    table_id: TableId,
    output_files: &'a [AppendFile],
    plan: &MaintenanceCommitPlan,
) -> PlannedMaintenanceFinish<'a> {
    PlannedMaintenanceFinish {
        table_id,
        new_snapshot: plan.new_snapshot,
        output_files: output_files
            .iter()
            .map(|file| PlannedMaintenanceOutputFile {
                file,
                new_snapshot: plan.new_snapshot,
            })
            .collect(),
        table_stats: PlannedMaintenanceTableStatsUpdate {
            table_id,
            source_file_size_bytes: plan.source_file_size_bytes,
            output_file_size_bytes: plan.output_file_size_bytes,
        },
    }
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
fn apply_maintenance_output_files_sync<'tx, O>(
    ops: &O,
    tx: &O::Tx<'tx>,
    finish: &PlannedMaintenanceFinish<'_>,
) -> Result<()>
where
    O: SyncMaintenanceCommitOps,
{
    for output in &finish.output_files {
        ops.insert_maintenance_data_file(tx, output.file, output.new_snapshot)?;
    }
    ops.recompute_table_column_stats_from_active_files(
        tx,
        finish.table_id,
        SnapshotId(finish.new_snapshot),
    )
}

#[cfg(any(feature = "postgres", feature = "quack"))]
async fn apply_maintenance_output_files_async<O, Tx>(
    ops: &O,
    tx: &Tx,
    finish: &PlannedMaintenanceFinish<'_>,
) -> Result<()>
where
    O: AsyncMaintenanceCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    for output in &finish.output_files {
        ops.insert_maintenance_data_file(tx, output.file, output.new_snapshot)
            .await?;
    }
    ops.recompute_table_column_stats_from_active_files(
        tx,
        finish.table_id,
        SnapshotId(finish.new_snapshot),
    )
    .await
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
fn apply_maintenance_table_stats_sync<'tx, O>(
    ops: &O,
    tx: &O::Tx<'tx>,
    stats: PlannedMaintenanceTableStatsUpdate,
) -> Result<()>
where
    O: SyncMaintenanceCommitOps,
{
    if ops.table_exists(tx, "ducklake_table_stats")? {
        ops.update_table_stats_for_maintenance(
            tx,
            stats.table_id,
            stats.source_file_size_bytes,
            stats.output_file_size_bytes,
        )?;
    }
    Ok(())
}

#[cfg(any(feature = "postgres", feature = "quack"))]
async fn apply_maintenance_table_stats_async<O, Tx>(
    ops: &O,
    tx: &Tx,
    stats: PlannedMaintenanceTableStatsUpdate,
) -> Result<()>
where
    O: AsyncMaintenanceCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    if ops.table_exists(tx, "ducklake_table_stats").await? {
        ops.update_table_stats_for_maintenance(
            tx,
            stats.table_id,
            stats.source_file_size_bytes,
            stats.output_file_size_bytes,
        )
        .await?;
    }
    Ok(())
}

#[derive(Debug, Clone)]
struct PlannedRewriteClose {
    table_id: TableId,
    source_ids: Vec<i64>,
    new_snapshot: i64,
    close_delete_files: bool,
}

fn planned_rewrite_close(
    table_id: TableId,
    source_ids: &[i64],
    new_snapshot: i64,
    close_delete_files: bool,
) -> PlannedRewriteClose {
    PlannedRewriteClose {
        table_id,
        source_ids: source_ids.to_vec(),
        new_snapshot,
        close_delete_files,
    }
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
fn apply_rewrite_close_sync<'tx, O>(
    ops: &O,
    tx: &O::Tx<'tx>,
    close: &PlannedRewriteClose,
) -> Result<()>
where
    O: SyncMaintenanceCommitOps,
{
    ops.end_active_data_files_by_id(tx, close.table_id, &close.source_ids, close.new_snapshot)?;
    if close.close_delete_files {
        ops.end_active_delete_files_by_data_file_id(
            tx,
            close.table_id,
            &close.source_ids,
            close.new_snapshot,
        )?;
    }
    Ok(())
}

#[cfg(any(feature = "postgres", feature = "quack"))]
async fn apply_rewrite_close_async<O, Tx>(
    ops: &O,
    tx: &Tx,
    close: &PlannedRewriteClose,
) -> Result<()>
where
    O: AsyncMaintenanceCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    ops.end_active_data_files_by_id(tx, close.table_id, &close.source_ids, close.new_snapshot)
        .await?;
    if close.close_delete_files {
        ops.end_active_delete_files_by_data_file_id(
            tx,
            close.table_id,
            &close.source_ids,
            close.new_snapshot,
        )
        .await?;
    }
    Ok(())
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) trait SyncMaintenanceCommitOps: SyncTableDataCommitOps {
    fn maintenance_execute(
        &self,
        tx: &Self::Tx<'_>,
        query: &crate::catalog_queries::CatalogQuery,
    ) -> Result<usize>;

    fn source_data_files(
        &self,
        tx: &Self::Tx<'_>,
        table_id: TableId,
        snapshot: i64,
        source_ids: &[i64],
    ) -> Result<Vec<MaintenanceSourceFile>>;

    fn delete_file_cleanup_rows(
        &self,
        tx: &Self::Tx<'_>,
        latest_snapshot: i64,
        source_ids: &[i64],
    ) -> Result<Vec<CleanupMetadataFile>>;

    fn schedule_cleanup_files(
        &self,
        tx: &Self::Tx<'_>,
        files: &[CleanupMetadataFile],
    ) -> Result<()>;

    fn delete_data_file_metadata(&self, tx: &Self::Tx<'_>, source_ids: &[i64]) -> Result<()>;

    fn delete_delete_files_by_data_file_id(
        &self,
        tx: &Self::Tx<'_>,
        source_ids: &[i64],
    ) -> Result<()>;

    fn delete_inlined_file_deletions_for_files(
        &self,
        tx: &Self::Tx<'_>,
        table_id: TableId,
        source_ids: &[i64],
    ) -> Result<()>;

    fn insert_maintenance_data_file(
        &self,
        tx: &Self::Tx<'_>,
        file: &AppendFile,
        new_snapshot: i64,
    ) -> Result<()> {
        let begin_snapshot = begin_snapshot_or(file.begin_snapshot, new_snapshot);
        let metadata_tables = DataFileMetadataTables::new(
            self.table_exists(tx, "ducklake_file_column_stats")?,
            self.table_exists(tx, "ducklake_file_variant_stats")?,
        );
        self.insert_data_file(tx, file, file.row_id_start, begin_snapshot, metadata_tables)
    }

    fn recompute_table_column_stats_from_active_files(
        &self,
        _tx: &Self::Tx<'_>,
        _table_id: TableId,
        _snapshot: SnapshotId,
    ) -> Result<()> {
        Ok(())
    }

    fn end_active_table_data_files(
        &self,
        tx: &Self::Tx<'_>,
        table_id: TableId,
        new_snapshot: i64,
    ) -> Result<()> {
        self.maintenance_execute(
            tx,
            &crate::catalog_queries::end_active_table_data_files(table_id, new_snapshot),
        )?;
        Ok(())
    }

    fn end_active_table_delete_files(
        &self,
        tx: &Self::Tx<'_>,
        table_id: TableId,
        new_snapshot: i64,
    ) -> Result<()> {
        self.maintenance_execute(
            tx,
            &crate::catalog_queries::end_active_table_delete_files(table_id, new_snapshot),
        )?;
        Ok(())
    }

    fn end_active_data_files_by_id(
        &self,
        tx: &Self::Tx<'_>,
        table_id: TableId,
        source_ids: &[i64],
        new_snapshot: i64,
    ) -> Result<()> {
        self.maintenance_execute(
            tx,
            &crate::catalog_queries::end_active_data_files_by_id(
                table_id,
                source_ids,
                new_snapshot,
            ),
        )?;
        Ok(())
    }

    fn end_active_delete_files_by_data_file_id(
        &self,
        tx: &Self::Tx<'_>,
        table_id: TableId,
        source_ids: &[i64],
        new_snapshot: i64,
    ) -> Result<()> {
        self.maintenance_execute(
            tx,
            &crate::catalog_queries::end_active_delete_files_by_data_file_id(
                table_id,
                source_ids,
                new_snapshot,
            ),
        )?;
        Ok(())
    }

    fn reset_table_stats_for_truncate(&self, tx: &Self::Tx<'_>, table_id: TableId) -> Result<()> {
        self.maintenance_execute(
            tx,
            &crate::catalog_queries::reset_table_stats_for_truncate(table_id),
        )?;
        Ok(())
    }

    fn update_table_stats_for_maintenance(
        &self,
        tx: &Self::Tx<'_>,
        table_id: TableId,
        source_file_size_bytes: i64,
        output_file_size_bytes: i64,
    ) -> Result<()> {
        self.maintenance_execute(
            tx,
            &crate::catalog_queries::update_table_stats_for_maintenance(
                table_id,
                source_file_size_bytes,
                output_file_size_bytes,
            ),
        )?;
        Ok(())
    }
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) trait SyncCatalogObjectCommitOps: SyncCommitOps {
    fn catalog_object_uuid_style(&self) -> crate::catalog_queries::UuidParamStyle;

    fn catalog_object_execute(
        &self,
        tx: &Self::Tx<'_>,
        query: &crate::catalog_queries::CatalogQuery,
    ) -> Result<usize>;

    fn catalog_object_close_inlined_data_rows_for_table(
        &self,
        _tx: &Self::Tx<'_>,
        _table_id: TableId,
        _new_snapshot: i64,
    ) -> Result<()> {
        Ok(())
    }

    fn catalog_object_query_schemas(
        &self,
        tx: &Self::Tx<'_>,
        query: &crate::catalog_queries::CatalogQuery,
    ) -> Result<Vec<SchemaEntry>>;

    fn catalog_object_query_tables(
        &self,
        tx: &Self::Tx<'_>,
        query: &crate::catalog_queries::CatalogQuery,
    ) -> Result<Vec<TableEntry>>;

    fn catalog_object_query_views(
        &self,
        tx: &Self::Tx<'_>,
        query: &crate::catalog_queries::CatalogQuery,
    ) -> Result<Vec<ViewEntry>>;

    fn catalog_object_query_i64s(
        &self,
        tx: &Self::Tx<'_>,
        query: &crate::catalog_queries::CatalogQuery,
    ) -> Result<Vec<i64>>;

    fn visible_schema(
        &self,
        tx: &Self::Tx<'_>,
        schema_name: &str,
        snapshot: SnapshotId,
    ) -> Result<Option<SchemaEntry>> {
        let exact = self.catalog_object_query_schemas(
            tx,
            &crate::catalog_queries::visible_schema_exact(schema_name, snapshot),
        )?;
        if !exact.is_empty() {
            return crate::catalog_queries::resolve_visible_schema(schema_name, exact, Vec::new());
        }
        let case_matches = self.catalog_object_query_schemas(
            tx,
            &crate::catalog_queries::visible_schema_case(schema_name, snapshot),
        )?;
        crate::catalog_queries::resolve_visible_schema(schema_name, Vec::new(), case_matches)
    }

    fn visible_table(
        &self,
        tx: &Self::Tx<'_>,
        schema_id: SchemaId,
        table_name: &str,
        snapshot: SnapshotId,
    ) -> Result<Option<TableEntry>> {
        let exact = self.catalog_object_query_tables(
            tx,
            &crate::catalog_queries::visible_table_exact(schema_id, table_name, snapshot),
        )?;
        if !exact.is_empty() {
            return crate::catalog_queries::resolve_visible_table(table_name, exact, Vec::new());
        }
        let case_matches =
            self.visible_tables_case_insensitive(tx, schema_id, table_name, snapshot)?;
        crate::catalog_queries::resolve_visible_table(table_name, Vec::new(), case_matches)
    }

    fn visible_view(
        &self,
        tx: &Self::Tx<'_>,
        schema_id: SchemaId,
        view_name: &str,
        snapshot: SnapshotId,
    ) -> Result<Option<ViewEntry>> {
        let exact = self.catalog_object_query_views(
            tx,
            &crate::catalog_queries::visible_view_exact(schema_id, view_name, snapshot),
        )?;
        if !exact.is_empty() {
            return crate::catalog_queries::resolve_visible_view(view_name, exact, Vec::new());
        }
        let case_matches =
            self.visible_views_case_insensitive(tx, schema_id, view_name, snapshot)?;
        crate::catalog_queries::resolve_visible_view(view_name, Vec::new(), case_matches)
    }

    fn visible_tables_case_insensitive(
        &self,
        tx: &Self::Tx<'_>,
        schema_id: SchemaId,
        table_name: &str,
        snapshot: SnapshotId,
    ) -> Result<Vec<TableEntry>> {
        self.catalog_object_query_tables(
            tx,
            &crate::catalog_queries::visible_tables_case(schema_id, table_name, snapshot),
        )
    }

    fn visible_views_case_insensitive(
        &self,
        tx: &Self::Tx<'_>,
        schema_id: SchemaId,
        view_name: &str,
        snapshot: SnapshotId,
    ) -> Result<Vec<ViewEntry>> {
        self.catalog_object_query_views(
            tx,
            &crate::catalog_queries::visible_views_case(schema_id, view_name, snapshot),
        )
    }

    fn visible_table_ids_for_schema(
        &self,
        tx: &Self::Tx<'_>,
        schema_id: SchemaId,
        snapshot: SnapshotId,
    ) -> Result<Vec<i64>> {
        self.catalog_object_query_i64s(
            tx,
            &crate::catalog_queries::visible_table_ids_for_schema(schema_id, snapshot),
        )
    }

    fn visible_view_ids_for_schema(
        &self,
        tx: &Self::Tx<'_>,
        schema_id: SchemaId,
        snapshot: SnapshotId,
    ) -> Result<Vec<i64>> {
        if !self.table_exists(tx, "ducklake_view")? {
            return Ok(Vec::new());
        }
        self.catalog_object_query_i64s(
            tx,
            &crate::catalog_queries::visible_view_ids_for_schema(schema_id, snapshot),
        )
    }

    fn data_path(&self, tx: &Self::Tx<'_>) -> Result<String>;

    fn insert_schema(&self, tx: &Self::Tx<'_>, schema: &PlannedSchemaRow) -> Result<()> {
        self.catalog_object_execute(
            tx,
            &crate::catalog_queries::insert_schema(
                schema.schema_id,
                schema.schema_uuid.clone(),
                schema.new_snapshot,
                &schema.schema_name,
                self.catalog_object_uuid_style(),
            ),
        )?;
        Ok(())
    }

    fn insert_table(
        &self,
        tx: &Self::Tx<'_>,
        table_id: i64,
        table_uuid: &str,
        new_snapshot: i64,
        schema_id: SchemaId,
        table_name: &str,
        path: Option<&str>,
        path_is_relative: bool,
    ) -> Result<()> {
        self.catalog_object_execute(
            tx,
            &crate::catalog_queries::insert_table(
                table_id,
                table_uuid,
                new_snapshot,
                schema_id,
                table_name,
                path,
                path_is_relative,
                self.catalog_object_uuid_style(),
            ),
        )?;
        Ok(())
    }

    fn insert_table_columns(
        &self,
        tx: &Self::Tx<'_>,
        table_id: i64,
        new_snapshot: i64,
        columns: &[CreateTableColumn],
    ) -> Result<()> {
        for column in columns {
            self.catalog_object_execute(
                tx,
                &crate::catalog_queries::insert_create_table_column(table_id, column, new_snapshot),
            )?;
        }
        Ok(())
    }

    fn insert_view(&self, tx: &Self::Tx<'_>, view: &PlannedViewRow<'_>) -> Result<()> {
        self.catalog_object_execute(
            tx,
            &crate::catalog_queries::insert_view(
                view.view_id,
                view.view_uuid.clone(),
                view.new_snapshot,
                view.schema_id,
                view.create,
                self.catalog_object_uuid_style(),
            ),
        )?;
        Ok(())
    }

    fn close_schema(
        &self,
        tx: &Self::Tx<'_>,
        schema_id: SchemaId,
        new_snapshot: i64,
    ) -> Result<()> {
        self.catalog_object_execute(
            tx,
            &crate::catalog_queries::close_schema(schema_id, new_snapshot),
        )?;
        Ok(())
    }

    fn close_tables_for_schema(
        &self,
        tx: &Self::Tx<'_>,
        schema_id: SchemaId,
        snapshot: SnapshotId,
        new_snapshot: i64,
    ) -> Result<()> {
        self.catalog_object_execute(
            tx,
            &crate::catalog_queries::close_tables_for_schema(schema_id, snapshot, new_snapshot),
        )?;
        Ok(())
    }

    fn close_views_for_schema(
        &self,
        tx: &Self::Tx<'_>,
        schema_id: SchemaId,
        snapshot: SnapshotId,
        new_snapshot: i64,
    ) -> Result<()> {
        self.catalog_object_execute(
            tx,
            &crate::catalog_queries::close_views_for_schema(schema_id, snapshot, new_snapshot),
        )?;
        Ok(())
    }

    fn close_table(
        &self,
        tx: &Self::Tx<'_>,
        table_id: TableId,
        new_snapshot: i64,
    ) -> Result<usize> {
        self.catalog_object_execute(
            tx,
            &crate::catalog_queries::close_table(table_id, new_snapshot),
        )
    }

    fn close_view(&self, tx: &Self::Tx<'_>, view_id: ViewId, new_snapshot: i64) -> Result<()> {
        self.catalog_object_execute(
            tx,
            &crate::catalog_queries::close_view(view_id, new_snapshot),
        )?;
        Ok(())
    }

    fn close_table_contents(
        &self,
        tx: &Self::Tx<'_>,
        table_id: TableId,
        new_snapshot: i64,
    ) -> Result<()> {
        for query in close_table_contents_queries(table_id, new_snapshot) {
            self.catalog_object_execute(tx, &query)?;
        }
        if self.table_exists(tx, "ducklake_inlined_data_tables")? {
            self.catalog_object_close_inlined_data_rows_for_table(tx, table_id, new_snapshot)?;
        }
        Ok(())
    }
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) trait SyncCreateTableWithFilesCommitOps:
    SyncCatalogObjectCommitOps + SyncTableDataCommitOps
{
    fn insert_table_stats_for_create_table(
        &self,
        tx: &Self::Tx<'_>,
        table_id: TableId,
        record_count: i64,
        file_size_bytes: i64,
    ) -> Result<()> {
        self.catalog_object_execute(
            tx,
            &crate::catalog_queries::insert_table_stats_for_create_table(
                table_id,
                record_count,
                file_size_bytes,
            ),
        )?;
        Ok(())
    }
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) trait SyncPartitioningCommitOps: SyncCommitOps {
    fn partitioning_execute(
        &self,
        tx: &Self::Tx<'_>,
        query: &crate::catalog_queries::CatalogQuery,
    ) -> Result<usize>;

    fn partitioning_query_column_parent_refs(
        &self,
        tx: &Self::Tx<'_>,
        query: &crate::catalog_queries::CatalogQuery,
    ) -> Result<Vec<(ColumnId, Option<i64>)>>;

    fn partitioning_query_infos(
        &self,
        tx: &Self::Tx<'_>,
        query: &crate::catalog_queries::CatalogQuery,
    ) -> Result<Vec<ducklake_catalog::PartitionInfo>>;

    fn partitioning_query_columns(
        &self,
        tx: &Self::Tx<'_>,
        query: &crate::catalog_queries::CatalogQuery,
    ) -> Result<Vec<ducklake_catalog::PartitionColumn>>;

    fn validate_partitioning_columns(
        &self,
        tx: &Self::Tx<'_>,
        table_id: TableId,
        snapshot: SnapshotId,
        columns: &[PartitionColumnCommit],
    ) -> Result<()> {
        let visible = self
            .partitioning_query_column_parent_refs(
                tx,
                &crate::catalog_queries::visible_column_parent_refs(table_id, snapshot),
            )?
            .into_iter()
            .map(|(column_id, _)| column_id);
        crate::catalog_queries::validate_partitioning_columns(table_id, columns, visible)
    }

    fn active_partition_specs(
        &self,
        tx: &Self::Tx<'_>,
        table_id: TableId,
        snapshot: SnapshotId,
    ) -> Result<Vec<ActivePartitionSpec>> {
        let partitions = self.partitioning_query_infos(
            tx,
            &crate::catalog_queries::active_partition_infos(table_id, snapshot),
        )?;
        let mut specs = Vec::with_capacity(partitions.len());
        for partition in partitions {
            let columns = self.partitioning_query_columns(
                tx,
                &crate::catalog_queries::partition_columns(partition.partition_id),
            )?;
            specs.push(active_partition_spec_from_rows(partition, columns));
        }
        Ok(specs)
    }

    fn close_partition(
        &self,
        tx: &Self::Tx<'_>,
        partition_id: ducklake_catalog::PartitionId,
        new_snapshot: i64,
    ) -> Result<usize> {
        self.partitioning_execute(
            tx,
            &crate::catalog_queries::close_partition(partition_id, new_snapshot),
        )
    }

    fn insert_partition_info(
        &self,
        tx: &Self::Tx<'_>,
        partition_id: ducklake_catalog::PartitionId,
        table_id: TableId,
        new_snapshot: i64,
    ) -> Result<()> {
        self.partitioning_execute(
            tx,
            &crate::catalog_queries::insert_partition_info(partition_id, table_id, new_snapshot),
        )?;
        Ok(())
    }

    fn insert_partition_column(
        &self,
        tx: &Self::Tx<'_>,
        partition_id: ducklake_catalog::PartitionId,
        table_id: TableId,
        partition_key_index: i64,
        column: &PartitionColumnCommit,
    ) -> Result<()> {
        self.partitioning_execute(
            tx,
            &crate::catalog_queries::insert_partition_column(
                partition_id,
                table_id,
                partition_key_index,
                column,
            ),
        )?;
        Ok(())
    }
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) trait SyncSortedByCommitOps: SyncCommitOps {
    fn sorted_by_execute(
        &self,
        tx: &Self::Tx<'_>,
        query: &crate::catalog_queries::CatalogQuery,
    ) -> Result<usize>;

    fn sorted_by_query_column_names(
        &self,
        tx: &Self::Tx<'_>,
        query: &crate::catalog_queries::CatalogQuery,
    ) -> Result<Vec<String>>;

    fn sorted_by_query_infos(
        &self,
        tx: &Self::Tx<'_>,
        query: &crate::catalog_queries::CatalogQuery,
    ) -> Result<Vec<ducklake_catalog::SortInfo>>;

    fn sorted_by_query_expressions(
        &self,
        tx: &Self::Tx<'_>,
        query: &crate::catalog_queries::CatalogQuery,
    ) -> Result<Vec<ducklake_catalog::SortExpression>>;

    fn validate_sort_columns(
        &self,
        tx: &Self::Tx<'_>,
        table_id: TableId,
        snapshot: SnapshotId,
        expressions: &[SortExpressionCommit],
    ) -> Result<()> {
        if expressions.is_empty() {
            return Ok(());
        }
        let column_names = self.sorted_by_query_column_names(
            tx,
            &crate::catalog_queries::visible_top_level_column_names(table_id, snapshot),
        )?;
        validate_sort_expression_columns(table_id, expressions, &column_names)
    }

    fn active_sort_specs(
        &self,
        tx: &Self::Tx<'_>,
        table_id: TableId,
        snapshot: SnapshotId,
    ) -> Result<Vec<ActiveSortSpec>> {
        let sorts = self.sorted_by_query_infos(
            tx,
            &crate::catalog_queries::active_sort_infos(table_id, snapshot),
        )?;
        let mut specs = Vec::with_capacity(sorts.len());
        for sort in sorts {
            let expressions = self.sorted_by_query_expressions(
                tx,
                &crate::catalog_queries::sort_expressions(sort.sort_id, table_id),
            )?;
            specs.push(active_sort_spec_from_rows(sort, expressions));
        }
        Ok(specs)
    }

    fn close_sort(
        &self,
        tx: &Self::Tx<'_>,
        sort_id: ducklake_catalog::SortId,
        table_id: TableId,
        new_snapshot: i64,
    ) -> Result<usize> {
        self.sorted_by_execute(
            tx,
            &crate::catalog_queries::close_sort(sort_id, table_id, new_snapshot),
        )
    }

    fn insert_sort_info(
        &self,
        tx: &Self::Tx<'_>,
        sort_id: ducklake_catalog::SortId,
        table_id: TableId,
        new_snapshot: i64,
    ) -> Result<()> {
        self.sorted_by_execute(
            tx,
            &crate::catalog_queries::insert_sort_info(sort_id, table_id, new_snapshot),
        )?;
        Ok(())
    }

    fn insert_sort_expression(
        &self,
        tx: &Self::Tx<'_>,
        sort_id: ducklake_catalog::SortId,
        table_id: TableId,
        sort_key_index: i64,
        expression: &SortExpressionCommit,
    ) -> Result<()> {
        self.sorted_by_execute(
            tx,
            &crate::catalog_queries::insert_sort_expression(
                sort_id,
                table_id,
                sort_key_index,
                expression,
            ),
        )?;
        Ok(())
    }
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) trait SyncColumnEvolutionCommitOps: SyncCommitOps {
    fn column_execute(
        &self,
        tx: &Self::Tx<'_>,
        query: &crate::catalog_queries::CatalogQuery,
    ) -> Result<usize>;

    fn column_query_i64s(
        &self,
        tx: &Self::Tx<'_>,
        query: &crate::catalog_queries::CatalogQuery,
    ) -> Result<Vec<i64>>;

    fn column_query_strings(
        &self,
        tx: &Self::Tx<'_>,
        query: &crate::catalog_queries::CatalogQuery,
    ) -> Result<Vec<String>>;

    fn column_query_opt_bools(
        &self,
        tx: &Self::Tx<'_>,
        query: &crate::catalog_queries::CatalogQuery,
    ) -> Result<Vec<Option<bool>>>;

    fn column_query_active_refs(
        &self,
        tx: &Self::Tx<'_>,
        query: &crate::catalog_queries::CatalogQuery,
    ) -> Result<Vec<(ColumnId, String, Option<ColumnId>)>>;

    fn visible_column(
        &self,
        tx: &Self::Tx<'_>,
        table_id: TableId,
        column_id: ColumnId,
        snapshot: SnapshotId,
    ) -> Result<Option<ColumnEntry>>;

    fn visible_columns(
        &self,
        tx: &Self::Tx<'_>,
        table_id: TableId,
        snapshot: SnapshotId,
    ) -> Result<Vec<ColumnEntry>>;

    fn table_id_exists(
        &self,
        tx: &Self::Tx<'_>,
        table_id: TableId,
        snapshot: SnapshotId,
    ) -> Result<bool> {
        Ok(any_count(self.column_query_i64s(
            tx,
            &crate::catalog_queries::table_id_exists(table_id, snapshot),
        )?))
    }

    fn active_column_refs(
        &self,
        tx: &Self::Tx<'_>,
        table_id: TableId,
        snapshot: SnapshotId,
    ) -> Result<Vec<(ColumnId, String, Option<ColumnId>)>> {
        self.column_query_active_refs(
            tx,
            &crate::catalog_queries::active_column_refs(table_id, snapshot),
        )
    }

    fn conflicting_sibling_column_name(
        &self,
        tx: &Self::Tx<'_>,
        table_id: TableId,
        column_id: ColumnId,
        parent_column: Option<ColumnId>,
        new_column_name: &str,
        snapshot: SnapshotId,
    ) -> Result<Option<String>> {
        Ok(first_string(self.column_query_strings(
            tx,
            &crate::catalog_queries::conflicting_sibling_column_name(
                table_id,
                column_id,
                parent_column,
                new_column_name,
                snapshot,
            ),
        )?))
    }

    fn rewrite_sort_expressions_for_rename(
        &self,
        tx: &Self::Tx<'_>,
        table_id: TableId,
        latest_snapshot: SnapshotId,
        new_snapshot: i64,
        next_catalog_id: i64,
        old_column_name: &str,
        new_column_name: &str,
    ) -> Result<i64>;

    fn top_level_column_count(
        &self,
        tx: &Self::Tx<'_>,
        table_id: TableId,
        snapshot: SnapshotId,
    ) -> Result<i64> {
        Ok(first_i64_or_zero(self.column_query_i64s(
            tx,
            &crate::catalog_queries::top_level_column_count(table_id, snapshot),
        )?))
    }

    fn drop_column_referenced_by_partition(
        &self,
        tx: &Self::Tx<'_>,
        table_id: TableId,
        column_id: ColumnId,
        snapshot: SnapshotId,
    ) -> Result<bool> {
        if !self.table_exists(tx, "ducklake_partition_info")?
            || !self.table_exists(tx, "ducklake_partition_column")?
        {
            return Ok(false);
        }
        Ok(any_count(self.column_query_i64s(
            tx,
            &crate::catalog_queries::active_partition_reference_count(
                table_id, column_id, snapshot,
            ),
        )?))
    }

    fn drop_column_referenced_by_sort(
        &self,
        tx: &Self::Tx<'_>,
        table_id: TableId,
        column_name: &str,
        snapshot: SnapshotId,
    ) -> Result<bool> {
        if !self.table_exists(tx, "ducklake_sort_info")?
            || !self.table_exists(tx, "ducklake_sort_expression")?
        {
            return Ok(false);
        }
        sort_expressions_reference_column(
            self.column_query_strings(
                tx,
                &crate::catalog_queries::active_sort_expressions(table_id, snapshot),
            )?,
            column_name,
        )
    }

    fn validate_set_not_null_stats(&self, tx: &Self::Tx<'_>, column: &ColumnEntry) -> Result<()> {
        if !self.table_exists(tx, "ducklake_table_stats")? {
            return validate_set_not_null_without_table_stats(column);
        }
        let table_stats_rows =
            self.column_query_i64s(tx, &set_not_null_table_stats_query(column))?;
        if !self.table_exists(tx, "ducklake_table_column_stats")? {
            return validate_set_not_null_stats_from_rows(column, table_stats_rows, None);
        }
        validate_set_not_null_stats_from_rows(
            column,
            table_stats_rows,
            Some(self.column_query_opt_bools(tx, &set_not_null_contains_null_query(column))?),
        )
    }

    fn close_column(
        &self,
        tx: &Self::Tx<'_>,
        table_id: TableId,
        column_id: ColumnId,
        new_snapshot: i64,
    ) -> Result<usize> {
        self.column_execute(
            tx,
            &crate::catalog_queries::close_column(table_id, column_id, new_snapshot),
        )
    }

    fn close_column_subtree(
        &self,
        tx: &Self::Tx<'_>,
        table_id: TableId,
        column_id: ColumnId,
        latest_snapshot: SnapshotId,
        new_snapshot: i64,
    ) -> Result<usize> {
        self.column_execute(
            tx,
            &crate::catalog_queries::close_column_subtree(
                table_id,
                column_id,
                latest_snapshot,
                new_snapshot,
            ),
        )
    }

    fn insert_column(
        &self,
        tx: &Self::Tx<'_>,
        column: &ColumnEntry,
        new_snapshot: i64,
    ) -> Result<()> {
        self.column_execute(
            tx,
            &crate::catalog_queries::insert_column(column, new_snapshot),
        )?;
        Ok(())
    }

    fn insert_type_tree_column(
        &self,
        tx: &Self::Tx<'_>,
        table_id: TableId,
        column: &PlannedColumnTypeTreeRow,
        new_snapshot: i64,
    ) -> Result<()> {
        self.column_execute(
            tx,
            &crate::catalog_queries::insert_type_tree_column(table_id, column, new_snapshot),
        )?;
        Ok(())
    }

    fn insert_add_column(
        &self,
        tx: &Self::Tx<'_>,
        table_id: TableId,
        column: &AddColumn,
        new_snapshot: i64,
    ) -> Result<()> {
        self.column_execute(
            tx,
            &crate::catalog_queries::insert_add_column(table_id, column, new_snapshot),
        )?;
        Ok(())
    }

    fn upsert_table_column_stats(&self, tx: &Self::Tx<'_>, stats: &TableColumnStats) -> Result<()> {
        let existing = self.column_query_i64s(tx, &table_column_stats_count_query(stats))?;
        let query = table_column_stats_upsert_query(stats, existing);
        self.column_execute(tx, &query)?;
        Ok(())
    }
}

#[cfg(any(feature = "postgres", feature = "quack"))]
#[async_trait]
pub(crate) trait AsyncCommitOps<Tx: Sync + ?Sized> {
    async fn prepare_commit(&self, _tx: &Tx) -> Result<()> {
        Ok(())
    }

    async fn latest_snapshot_ids(&self, tx: &Tx) -> Result<CommitSnapshotIds>;

    async fn table_exists(&self, tx: &Tx, table_name: &str) -> Result<bool>;

    async fn insert_snapshot(
        &self,
        tx: &Tx,
        snapshot_id: i64,
        schema_version: i64,
        next_catalog_id: i64,
        next_file_id: i64,
    ) -> Result<()>;

    async fn insert_snapshot_change(
        &self,
        tx: &Tx,
        snapshot_id: i64,
        changes_made: &str,
        commit_message: &str,
    ) -> Result<()>;
}

#[cfg(any(feature = "postgres", feature = "quack"))]
#[async_trait]
pub(crate) trait AsyncTableDataCommitOps<Tx: Sync + ?Sized>: AsyncCommitOps<Tx> {
    async fn table_data_execute(
        &self,
        tx: &Tx,
        query: &crate::catalog_queries::CatalogQuery,
    ) -> Result<usize>;

    async fn next_row_id_for_append(
        &self,
        tx: &Tx,
        table_id: TableId,
        has_table_stats: bool,
    ) -> Result<Option<i64>>;

    async fn insert_data_file(
        &self,
        tx: &Tx,
        file: &AppendFile,
        row_id_start: Option<i64>,
        begin_snapshot: i64,
        metadata_tables: DataFileMetadataTables,
    ) -> Result<()> {
        for command in
            insert_data_file_commands(file, row_id_start, begin_snapshot, metadata_tables)
        {
            self.table_data_execute(tx, &command).await?;
        }

        Ok(())
    }

    async fn update_table_column_stats_for_append(
        &self,
        tx: &Tx,
        table_id: TableId,
        latest_snapshot: SnapshotId,
        files: &[AppendFile],
    ) -> Result<()>;

    async fn update_table_stats_for_append(
        &self,
        tx: &Tx,
        table_id: TableId,
        record_count: i64,
        allocated_row_id_count: i64,
        file_size_bytes: i64,
    ) -> Result<usize> {
        self.table_data_execute(
            tx,
            &crate::catalog_queries::update_table_stats_for_append(
                table_id,
                record_count,
                allocated_row_id_count,
                file_size_bytes,
            ),
        )
        .await
    }

    async fn insert_table_stats_from_active_files(
        &self,
        _tx: &Tx,
        _table_id: TableId,
    ) -> Result<()> {
        Ok(())
    }

    fn validate_inlined_append(
        &self,
        rows_appended: i64,
        schema: &TableSchema,
        batches: &[RecordBatch],
        preserved_row_ids: Option<&[i64]>,
        context: &str,
    ) -> Result<()>;

    async fn ensure_inlined_data_table(
        &self,
        tx: &Tx,
        table_id: TableId,
        schema_version: i64,
        schema: &TableSchema,
    ) -> Result<String>;

    async fn insert_inlined_batches(
        &self,
        tx: &Tx,
        table_name: &str,
        schema: &TableSchema,
        batches: &[RecordBatch],
        preserved_row_ids: Option<&[i64]>,
        row_id_start: i64,
        begin_snapshot: i64,
    ) -> Result<()>;

    async fn update_table_stats_for_inlined_append(
        &self,
        tx: &Tx,
        table_id: TableId,
        rows_appended: i64,
    ) -> Result<usize> {
        self.table_data_execute(
            tx,
            &crate::catalog_queries::update_table_stats_for_inlined_append(table_id, rows_appended),
        )
        .await
    }

    async fn insert_table_stats_for_inlined_append(
        &self,
        tx: &Tx,
        table_id: TableId,
        rows_appended: i64,
        next_row_id: i64,
    ) -> Result<()> {
        self.table_data_execute(
            tx,
            &crate::catalog_queries::insert_table_stats_for_inlined_append(
                table_id,
                rows_appended,
                next_row_id,
            ),
        )
        .await?;
        Ok(())
    }

    async fn replace_overwritten_delete_file(
        &self,
        tx: &Tx,
        file: &DeleteFileCommit,
        new_snapshot: i64,
    ) -> Result<()>;

    async fn insert_delete_file(
        &self,
        tx: &Tx,
        file: &DeleteFileCommit,
        begin_snapshot: i64,
        format: &str,
    ) -> Result<()> {
        self.table_data_execute(
            tx,
            &insert_delete_file_command(file, begin_snapshot, format),
        )
        .await?;
        Ok(())
    }

    async fn update_table_stats_for_delete(
        &self,
        tx: &Tx,
        table_id: TableId,
        deleted_rows: i64,
    ) -> Result<usize> {
        self.table_data_execute(
            tx,
            &crate::catalog_queries::update_table_stats_for_delete(table_id, deleted_rows),
        )
        .await
    }

    async fn write_inlined_file_deletions(
        &self,
        tx: &Tx,
        table_id: TableId,
        deletions: &[InlinedFileDeletionCommit],
        new_snapshot: i64,
    ) -> Result<()>;

    async fn apply_inlined_data_deletions(
        &self,
        tx: &Tx,
        deletions: &[InlinedDataDeletionCommit],
        new_snapshot: i64,
    ) -> Result<i64>;

    async fn delete_flushed_inlined_data(
        &self,
        tx: &Tx,
        flushed_tables: &[FlushedInlinedDataCommit],
        flush_snapshot: i64,
    ) -> Result<i64>;

    async fn delete_flushed_inlined_file_deletions(
        &self,
        tx: &Tx,
        table_id: TableId,
        flushed_deletions: &[FlushedInlinedFileDeletionCommit],
    ) -> Result<()>;

    async fn update_table_stats_for_table_change(
        &self,
        tx: &Tx,
        table_id: TableId,
        appended_rows: i64,
        deleted_rows: i64,
        appended_file_size_bytes: i64,
        allocated_row_id_count: i64,
    ) -> Result<usize> {
        self.table_data_execute(
            tx,
            &crate::catalog_queries::update_table_stats_for_table_change(
                table_id,
                appended_rows,
                deleted_rows,
                appended_file_size_bytes,
                allocated_row_id_count,
            ),
        )
        .await
    }
}

#[cfg(any(feature = "postgres", feature = "quack"))]
#[async_trait]
pub(crate) trait AsyncMaintenanceCommitOps<Tx: Sync + ?Sized>:
    AsyncTableDataCommitOps<Tx>
{
    async fn maintenance_execute(
        &self,
        tx: &Tx,
        query: &crate::catalog_queries::CatalogQuery,
    ) -> Result<usize>;

    async fn source_data_files(
        &self,
        tx: &Tx,
        table_id: TableId,
        snapshot: i64,
        source_ids: &[i64],
    ) -> Result<Vec<MaintenanceSourceFile>>;

    async fn delete_file_cleanup_rows(
        &self,
        tx: &Tx,
        latest_snapshot: i64,
        source_ids: &[i64],
    ) -> Result<Vec<CleanupMetadataFile>>;

    async fn schedule_cleanup_files(&self, tx: &Tx, files: &[CleanupMetadataFile]) -> Result<()>;

    async fn delete_data_file_metadata(&self, tx: &Tx, source_ids: &[i64]) -> Result<()>;

    async fn delete_delete_files_by_data_file_id(&self, tx: &Tx, source_ids: &[i64]) -> Result<()>;

    async fn delete_inlined_file_deletions_for_files(
        &self,
        tx: &Tx,
        table_id: TableId,
        source_ids: &[i64],
    ) -> Result<()>;

    async fn insert_maintenance_data_file(
        &self,
        tx: &Tx,
        file: &AppendFile,
        new_snapshot: i64,
    ) -> Result<()> {
        let begin_snapshot = begin_snapshot_or(file.begin_snapshot, new_snapshot);
        let metadata_tables = DataFileMetadataTables::new(
            self.table_exists(tx, "ducklake_file_column_stats").await?,
            self.table_exists(tx, "ducklake_file_variant_stats").await?,
        );
        self.insert_data_file(tx, file, file.row_id_start, begin_snapshot, metadata_tables)
            .await
    }

    async fn recompute_table_column_stats_from_active_files(
        &self,
        _tx: &Tx,
        _table_id: TableId,
        _snapshot: SnapshotId,
    ) -> Result<()> {
        Ok(())
    }

    async fn end_active_table_data_files(
        &self,
        tx: &Tx,
        table_id: TableId,
        new_snapshot: i64,
    ) -> Result<()> {
        self.maintenance_execute(
            tx,
            &crate::catalog_queries::end_active_table_data_files(table_id, new_snapshot),
        )
        .await?;
        Ok(())
    }

    async fn end_active_table_delete_files(
        &self,
        tx: &Tx,
        table_id: TableId,
        new_snapshot: i64,
    ) -> Result<()> {
        self.maintenance_execute(
            tx,
            &crate::catalog_queries::end_active_table_delete_files(table_id, new_snapshot),
        )
        .await?;
        Ok(())
    }

    async fn end_active_data_files_by_id(
        &self,
        tx: &Tx,
        table_id: TableId,
        source_ids: &[i64],
        new_snapshot: i64,
    ) -> Result<()> {
        self.maintenance_execute(
            tx,
            &crate::catalog_queries::end_active_data_files_by_id(
                table_id,
                source_ids,
                new_snapshot,
            ),
        )
        .await?;
        Ok(())
    }

    async fn end_active_delete_files_by_data_file_id(
        &self,
        tx: &Tx,
        table_id: TableId,
        source_ids: &[i64],
        new_snapshot: i64,
    ) -> Result<()> {
        self.maintenance_execute(
            tx,
            &crate::catalog_queries::end_active_delete_files_by_data_file_id(
                table_id,
                source_ids,
                new_snapshot,
            ),
        )
        .await?;
        Ok(())
    }

    async fn reset_table_stats_for_truncate(&self, tx: &Tx, table_id: TableId) -> Result<()> {
        self.maintenance_execute(
            tx,
            &crate::catalog_queries::reset_table_stats_for_truncate(table_id),
        )
        .await?;
        Ok(())
    }

    async fn update_table_stats_for_maintenance(
        &self,
        tx: &Tx,
        table_id: TableId,
        source_file_size_bytes: i64,
        output_file_size_bytes: i64,
    ) -> Result<()> {
        self.maintenance_execute(
            tx,
            &crate::catalog_queries::update_table_stats_for_maintenance(
                table_id,
                source_file_size_bytes,
                output_file_size_bytes,
            ),
        )
        .await?;
        Ok(())
    }
}

#[cfg(any(feature = "postgres", feature = "quack"))]
#[async_trait]
pub(crate) trait AsyncCatalogObjectCommitOps<Tx: Sync + ?Sized>: AsyncCommitOps<Tx> {
    fn catalog_object_uuid_style(&self) -> crate::catalog_queries::UuidParamStyle;

    async fn catalog_object_execute(
        &self,
        tx: &Tx,
        query: &crate::catalog_queries::CatalogQuery,
    ) -> Result<usize>;

    async fn catalog_object_close_inlined_data_rows_for_table(
        &self,
        _tx: &Tx,
        _table_id: TableId,
        _new_snapshot: i64,
    ) -> Result<()> {
        Ok(())
    }

    async fn catalog_object_query_schemas(
        &self,
        tx: &Tx,
        query: &crate::catalog_queries::CatalogQuery,
    ) -> Result<Vec<SchemaEntry>>;

    async fn catalog_object_query_tables(
        &self,
        tx: &Tx,
        query: &crate::catalog_queries::CatalogQuery,
    ) -> Result<Vec<TableEntry>>;

    async fn catalog_object_query_views(
        &self,
        tx: &Tx,
        query: &crate::catalog_queries::CatalogQuery,
    ) -> Result<Vec<ViewEntry>>;

    async fn catalog_object_query_i64s(
        &self,
        tx: &Tx,
        query: &crate::catalog_queries::CatalogQuery,
    ) -> Result<Vec<i64>>;

    async fn visible_schema(
        &self,
        tx: &Tx,
        schema_name: &str,
        snapshot: SnapshotId,
    ) -> Result<Option<SchemaEntry>> {
        let exact = self
            .catalog_object_query_schemas(
                tx,
                &crate::catalog_queries::visible_schema_exact(schema_name, snapshot),
            )
            .await?;
        if !exact.is_empty() {
            return crate::catalog_queries::resolve_visible_schema(schema_name, exact, Vec::new());
        }
        let case_matches = self
            .catalog_object_query_schemas(
                tx,
                &crate::catalog_queries::visible_schema_case(schema_name, snapshot),
            )
            .await?;
        crate::catalog_queries::resolve_visible_schema(schema_name, Vec::new(), case_matches)
    }

    async fn visible_table(
        &self,
        tx: &Tx,
        schema_id: SchemaId,
        table_name: &str,
        snapshot: SnapshotId,
    ) -> Result<Option<TableEntry>> {
        let exact = self
            .catalog_object_query_tables(
                tx,
                &crate::catalog_queries::visible_table_exact(schema_id, table_name, snapshot),
            )
            .await?;
        if !exact.is_empty() {
            return crate::catalog_queries::resolve_visible_table(table_name, exact, Vec::new());
        }
        let case_matches = self
            .visible_tables_case_insensitive(tx, schema_id, table_name, snapshot)
            .await?;
        crate::catalog_queries::resolve_visible_table(table_name, Vec::new(), case_matches)
    }

    async fn visible_view(
        &self,
        tx: &Tx,
        schema_id: SchemaId,
        view_name: &str,
        snapshot: SnapshotId,
    ) -> Result<Option<ViewEntry>> {
        let exact = self
            .catalog_object_query_views(
                tx,
                &crate::catalog_queries::visible_view_exact(schema_id, view_name, snapshot),
            )
            .await?;
        if !exact.is_empty() {
            return crate::catalog_queries::resolve_visible_view(view_name, exact, Vec::new());
        }
        let case_matches = self
            .visible_views_case_insensitive(tx, schema_id, view_name, snapshot)
            .await?;
        crate::catalog_queries::resolve_visible_view(view_name, Vec::new(), case_matches)
    }

    async fn visible_tables_case_insensitive(
        &self,
        tx: &Tx,
        schema_id: SchemaId,
        table_name: &str,
        snapshot: SnapshotId,
    ) -> Result<Vec<TableEntry>> {
        self.catalog_object_query_tables(
            tx,
            &crate::catalog_queries::visible_tables_case(schema_id, table_name, snapshot),
        )
        .await
    }

    async fn visible_views_case_insensitive(
        &self,
        tx: &Tx,
        schema_id: SchemaId,
        view_name: &str,
        snapshot: SnapshotId,
    ) -> Result<Vec<ViewEntry>> {
        self.catalog_object_query_views(
            tx,
            &crate::catalog_queries::visible_views_case(schema_id, view_name, snapshot),
        )
        .await
    }

    async fn visible_table_ids_for_schema(
        &self,
        tx: &Tx,
        schema_id: SchemaId,
        snapshot: SnapshotId,
    ) -> Result<Vec<i64>> {
        self.catalog_object_query_i64s(
            tx,
            &crate::catalog_queries::visible_table_ids_for_schema(schema_id, snapshot),
        )
        .await
    }

    async fn visible_view_ids_for_schema(
        &self,
        tx: &Tx,
        schema_id: SchemaId,
        snapshot: SnapshotId,
    ) -> Result<Vec<i64>> {
        if !self.table_exists(tx, "ducklake_view").await? {
            return Ok(Vec::new());
        }
        self.catalog_object_query_i64s(
            tx,
            &crate::catalog_queries::visible_view_ids_for_schema(schema_id, snapshot),
        )
        .await
    }

    async fn data_path(&self, tx: &Tx) -> Result<String>;

    async fn insert_schema(&self, tx: &Tx, schema: &PlannedSchemaRow) -> Result<()> {
        self.catalog_object_execute(
            tx,
            &crate::catalog_queries::insert_schema(
                schema.schema_id,
                schema.schema_uuid.clone(),
                schema.new_snapshot,
                &schema.schema_name,
                self.catalog_object_uuid_style(),
            ),
        )
        .await?;
        Ok(())
    }

    async fn insert_table(
        &self,
        tx: &Tx,
        table_id: i64,
        table_uuid: &str,
        new_snapshot: i64,
        schema_id: SchemaId,
        table_name: &str,
        path: Option<&str>,
        path_is_relative: bool,
    ) -> Result<()> {
        self.catalog_object_execute(
            tx,
            &crate::catalog_queries::insert_table(
                table_id,
                table_uuid,
                new_snapshot,
                schema_id,
                table_name,
                path,
                path_is_relative,
                self.catalog_object_uuid_style(),
            ),
        )
        .await?;
        Ok(())
    }

    async fn insert_table_columns(
        &self,
        tx: &Tx,
        table_id: i64,
        new_snapshot: i64,
        columns: &[CreateTableColumn],
    ) -> Result<()> {
        for column in columns {
            self.catalog_object_execute(
                tx,
                &crate::catalog_queries::insert_create_table_column(table_id, column, new_snapshot),
            )
            .await?;
        }
        Ok(())
    }

    async fn insert_view(&self, tx: &Tx, view: &PlannedViewRow<'_>) -> Result<()> {
        self.catalog_object_execute(
            tx,
            &crate::catalog_queries::insert_view(
                view.view_id,
                view.view_uuid.clone(),
                view.new_snapshot,
                view.schema_id,
                view.create,
                self.catalog_object_uuid_style(),
            ),
        )
        .await?;
        Ok(())
    }

    async fn close_schema(&self, tx: &Tx, schema_id: SchemaId, new_snapshot: i64) -> Result<()> {
        self.catalog_object_execute(
            tx,
            &crate::catalog_queries::close_schema(schema_id, new_snapshot),
        )
        .await?;
        Ok(())
    }

    async fn close_tables_for_schema(
        &self,
        tx: &Tx,
        schema_id: SchemaId,
        snapshot: SnapshotId,
        new_snapshot: i64,
    ) -> Result<()> {
        self.catalog_object_execute(
            tx,
            &crate::catalog_queries::close_tables_for_schema(schema_id, snapshot, new_snapshot),
        )
        .await?;
        Ok(())
    }

    async fn close_views_for_schema(
        &self,
        tx: &Tx,
        schema_id: SchemaId,
        snapshot: SnapshotId,
        new_snapshot: i64,
    ) -> Result<()> {
        self.catalog_object_execute(
            tx,
            &crate::catalog_queries::close_views_for_schema(schema_id, snapshot, new_snapshot),
        )
        .await?;
        Ok(())
    }

    async fn close_table(&self, tx: &Tx, table_id: TableId, new_snapshot: i64) -> Result<usize> {
        self.catalog_object_execute(
            tx,
            &crate::catalog_queries::close_table(table_id, new_snapshot),
        )
        .await
    }

    async fn close_view(&self, tx: &Tx, view_id: ViewId, new_snapshot: i64) -> Result<()> {
        self.catalog_object_execute(
            tx,
            &crate::catalog_queries::close_view(view_id, new_snapshot),
        )
        .await?;
        Ok(())
    }

    async fn close_table_contents(
        &self,
        tx: &Tx,
        table_id: TableId,
        new_snapshot: i64,
    ) -> Result<()> {
        for query in close_table_contents_queries(table_id, new_snapshot) {
            self.catalog_object_execute(tx, &query).await?;
        }
        if self
            .table_exists(tx, "ducklake_inlined_data_tables")
            .await?
        {
            self.catalog_object_close_inlined_data_rows_for_table(tx, table_id, new_snapshot)
                .await?;
        }
        Ok(())
    }
}

#[cfg(any(feature = "postgres", feature = "quack"))]
#[async_trait]
pub(crate) trait AsyncCreateTableWithFilesCommitOps<Tx: Sync + ?Sized>:
    AsyncCatalogObjectCommitOps<Tx> + AsyncTableDataCommitOps<Tx>
{
    async fn insert_table_stats_for_create_table(
        &self,
        tx: &Tx,
        table_id: TableId,
        record_count: i64,
        file_size_bytes: i64,
    ) -> Result<()> {
        self.catalog_object_execute(
            tx,
            &crate::catalog_queries::insert_table_stats_for_create_table(
                table_id,
                record_count,
                file_size_bytes,
            ),
        )
        .await?;
        Ok(())
    }
}

#[cfg(any(feature = "postgres", feature = "quack"))]
#[async_trait]
pub(crate) trait AsyncPartitioningCommitOps<Tx: Sync + ?Sized>: AsyncCommitOps<Tx> {
    async fn partitioning_execute(
        &self,
        tx: &Tx,
        query: &crate::catalog_queries::CatalogQuery,
    ) -> Result<usize>;

    async fn partitioning_query_column_parent_refs(
        &self,
        tx: &Tx,
        query: &crate::catalog_queries::CatalogQuery,
    ) -> Result<Vec<(ColumnId, Option<i64>)>>;

    async fn partitioning_query_infos(
        &self,
        tx: &Tx,
        query: &crate::catalog_queries::CatalogQuery,
    ) -> Result<Vec<ducklake_catalog::PartitionInfo>>;

    async fn partitioning_query_columns(
        &self,
        tx: &Tx,
        query: &crate::catalog_queries::CatalogQuery,
    ) -> Result<Vec<ducklake_catalog::PartitionColumn>>;

    async fn validate_partitioning_columns(
        &self,
        tx: &Tx,
        table_id: TableId,
        snapshot: SnapshotId,
        columns: &[PartitionColumnCommit],
    ) -> Result<()> {
        let visible = self
            .partitioning_query_column_parent_refs(
                tx,
                &crate::catalog_queries::visible_column_parent_refs(table_id, snapshot),
            )
            .await?
            .into_iter()
            .map(|(column_id, _)| column_id);
        crate::catalog_queries::validate_partitioning_columns(table_id, columns, visible)
    }

    async fn active_partition_specs(
        &self,
        tx: &Tx,
        table_id: TableId,
        snapshot: SnapshotId,
    ) -> Result<Vec<ActivePartitionSpec>> {
        let partitions = self
            .partitioning_query_infos(
                tx,
                &crate::catalog_queries::active_partition_infos(table_id, snapshot),
            )
            .await?;
        let mut specs = Vec::with_capacity(partitions.len());
        for partition in partitions {
            let columns = self
                .partitioning_query_columns(
                    tx,
                    &crate::catalog_queries::partition_columns(partition.partition_id),
                )
                .await?;
            specs.push(active_partition_spec_from_rows(partition, columns));
        }
        Ok(specs)
    }

    async fn close_partition(
        &self,
        tx: &Tx,
        partition_id: ducklake_catalog::PartitionId,
        new_snapshot: i64,
    ) -> Result<usize> {
        self.partitioning_execute(
            tx,
            &crate::catalog_queries::close_partition(partition_id, new_snapshot),
        )
        .await
    }

    async fn insert_partition_info(
        &self,
        tx: &Tx,
        partition_id: ducklake_catalog::PartitionId,
        table_id: TableId,
        new_snapshot: i64,
    ) -> Result<()> {
        self.partitioning_execute(
            tx,
            &crate::catalog_queries::insert_partition_info(partition_id, table_id, new_snapshot),
        )
        .await?;
        Ok(())
    }

    async fn insert_partition_column(
        &self,
        tx: &Tx,
        partition_id: ducklake_catalog::PartitionId,
        table_id: TableId,
        partition_key_index: i64,
        column: &PartitionColumnCommit,
    ) -> Result<()> {
        self.partitioning_execute(
            tx,
            &crate::catalog_queries::insert_partition_column(
                partition_id,
                table_id,
                partition_key_index,
                column,
            ),
        )
        .await?;
        Ok(())
    }
}

#[cfg(any(feature = "postgres", feature = "quack"))]
#[async_trait]
pub(crate) trait AsyncSortedByCommitOps<Tx: Sync + ?Sized>: AsyncCommitOps<Tx> {
    async fn sorted_by_execute(
        &self,
        tx: &Tx,
        query: &crate::catalog_queries::CatalogQuery,
    ) -> Result<usize>;

    async fn sorted_by_query_column_names(
        &self,
        tx: &Tx,
        query: &crate::catalog_queries::CatalogQuery,
    ) -> Result<Vec<String>>;

    async fn sorted_by_query_infos(
        &self,
        tx: &Tx,
        query: &crate::catalog_queries::CatalogQuery,
    ) -> Result<Vec<ducklake_catalog::SortInfo>>;

    async fn sorted_by_query_expressions(
        &self,
        tx: &Tx,
        query: &crate::catalog_queries::CatalogQuery,
    ) -> Result<Vec<ducklake_catalog::SortExpression>>;

    async fn validate_sort_columns(
        &self,
        tx: &Tx,
        table_id: TableId,
        snapshot: SnapshotId,
        expressions: &[SortExpressionCommit],
    ) -> Result<()> {
        if expressions.is_empty() {
            return Ok(());
        }
        let column_names = self
            .sorted_by_query_column_names(
                tx,
                &crate::catalog_queries::visible_top_level_column_names(table_id, snapshot),
            )
            .await?;
        validate_sort_expression_columns(table_id, expressions, &column_names)
    }

    async fn active_sort_specs(
        &self,
        tx: &Tx,
        table_id: TableId,
        snapshot: SnapshotId,
    ) -> Result<Vec<ActiveSortSpec>> {
        let sorts = self
            .sorted_by_query_infos(
                tx,
                &crate::catalog_queries::active_sort_infos(table_id, snapshot),
            )
            .await?;
        let mut specs = Vec::with_capacity(sorts.len());
        for sort in sorts {
            let expressions = self
                .sorted_by_query_expressions(
                    tx,
                    &crate::catalog_queries::sort_expressions(sort.sort_id, table_id),
                )
                .await?;
            specs.push(active_sort_spec_from_rows(sort, expressions));
        }
        Ok(specs)
    }

    async fn close_sort(
        &self,
        tx: &Tx,
        sort_id: ducklake_catalog::SortId,
        table_id: TableId,
        new_snapshot: i64,
    ) -> Result<usize> {
        self.sorted_by_execute(
            tx,
            &crate::catalog_queries::close_sort(sort_id, table_id, new_snapshot),
        )
        .await
    }

    async fn insert_sort_info(
        &self,
        tx: &Tx,
        sort_id: ducklake_catalog::SortId,
        table_id: TableId,
        new_snapshot: i64,
    ) -> Result<()> {
        self.sorted_by_execute(
            tx,
            &crate::catalog_queries::insert_sort_info(sort_id, table_id, new_snapshot),
        )
        .await?;
        Ok(())
    }

    async fn insert_sort_expression(
        &self,
        tx: &Tx,
        sort_id: ducklake_catalog::SortId,
        table_id: TableId,
        sort_key_index: i64,
        expression: &SortExpressionCommit,
    ) -> Result<()> {
        self.sorted_by_execute(
            tx,
            &crate::catalog_queries::insert_sort_expression(
                sort_id,
                table_id,
                sort_key_index,
                expression,
            ),
        )
        .await?;
        Ok(())
    }
}

#[cfg(any(feature = "postgres", feature = "quack"))]
#[async_trait]
pub(crate) trait AsyncColumnEvolutionCommitOps<Tx: Sync + ?Sized>:
    AsyncCommitOps<Tx>
{
    async fn column_execute(
        &self,
        tx: &Tx,
        query: &crate::catalog_queries::CatalogQuery,
    ) -> Result<usize>;

    async fn column_query_i64s(
        &self,
        tx: &Tx,
        query: &crate::catalog_queries::CatalogQuery,
    ) -> Result<Vec<i64>>;

    async fn column_query_strings(
        &self,
        tx: &Tx,
        query: &crate::catalog_queries::CatalogQuery,
    ) -> Result<Vec<String>>;

    async fn column_query_opt_bools(
        &self,
        tx: &Tx,
        query: &crate::catalog_queries::CatalogQuery,
    ) -> Result<Vec<Option<bool>>>;

    async fn column_query_active_refs(
        &self,
        tx: &Tx,
        query: &crate::catalog_queries::CatalogQuery,
    ) -> Result<Vec<(ColumnId, String, Option<ColumnId>)>>;

    async fn visible_column(
        &self,
        tx: &Tx,
        table_id: TableId,
        column_id: ColumnId,
        snapshot: SnapshotId,
    ) -> Result<Option<ColumnEntry>>;

    async fn visible_columns(
        &self,
        tx: &Tx,
        table_id: TableId,
        snapshot: SnapshotId,
    ) -> Result<Vec<ColumnEntry>>;

    async fn table_id_exists(
        &self,
        tx: &Tx,
        table_id: TableId,
        snapshot: SnapshotId,
    ) -> Result<bool> {
        Ok(any_count(
            self.column_query_i64s(
                tx,
                &crate::catalog_queries::table_id_exists(table_id, snapshot),
            )
            .await?,
        ))
    }

    async fn active_column_refs(
        &self,
        tx: &Tx,
        table_id: TableId,
        snapshot: SnapshotId,
    ) -> Result<Vec<(ColumnId, String, Option<ColumnId>)>> {
        self.column_query_active_refs(
            tx,
            &crate::catalog_queries::active_column_refs(table_id, snapshot),
        )
        .await
    }

    async fn conflicting_sibling_column_name(
        &self,
        tx: &Tx,
        table_id: TableId,
        column_id: ColumnId,
        parent_column: Option<ColumnId>,
        new_column_name: &str,
        snapshot: SnapshotId,
    ) -> Result<Option<String>> {
        Ok(first_string(
            self.column_query_strings(
                tx,
                &crate::catalog_queries::conflicting_sibling_column_name(
                    table_id,
                    column_id,
                    parent_column,
                    new_column_name,
                    snapshot,
                ),
            )
            .await?,
        ))
    }

    async fn rewrite_sort_expressions_for_rename(
        &self,
        tx: &Tx,
        table_id: TableId,
        latest_snapshot: SnapshotId,
        new_snapshot: i64,
        next_catalog_id: i64,
        old_column_name: &str,
        new_column_name: &str,
    ) -> Result<i64>;

    async fn top_level_column_count(
        &self,
        tx: &Tx,
        table_id: TableId,
        snapshot: SnapshotId,
    ) -> Result<i64> {
        Ok(first_i64_or_zero(
            self.column_query_i64s(
                tx,
                &crate::catalog_queries::top_level_column_count(table_id, snapshot),
            )
            .await?,
        ))
    }

    async fn drop_column_referenced_by_partition(
        &self,
        tx: &Tx,
        table_id: TableId,
        column_id: ColumnId,
        snapshot: SnapshotId,
    ) -> Result<bool> {
        if !self.table_exists(tx, "ducklake_partition_info").await?
            || !self.table_exists(tx, "ducklake_partition_column").await?
        {
            return Ok(false);
        }
        Ok(any_count(
            self.column_query_i64s(
                tx,
                &crate::catalog_queries::active_partition_reference_count(
                    table_id, column_id, snapshot,
                ),
            )
            .await?,
        ))
    }

    async fn drop_column_referenced_by_sort(
        &self,
        tx: &Tx,
        table_id: TableId,
        column_name: &str,
        snapshot: SnapshotId,
    ) -> Result<bool> {
        if !self.table_exists(tx, "ducklake_sort_info").await?
            || !self.table_exists(tx, "ducklake_sort_expression").await?
        {
            return Ok(false);
        }
        sort_expressions_reference_column(
            self.column_query_strings(
                tx,
                &crate::catalog_queries::active_sort_expressions(table_id, snapshot),
            )
            .await?,
            column_name,
        )
    }

    async fn validate_set_not_null_stats(&self, tx: &Tx, column: &ColumnEntry) -> Result<()> {
        if !self.table_exists(tx, "ducklake_table_stats").await? {
            return validate_set_not_null_without_table_stats(column);
        }
        let table_stats_rows = self
            .column_query_i64s(tx, &set_not_null_table_stats_query(column))
            .await?;
        if !self.table_exists(tx, "ducklake_table_column_stats").await? {
            return validate_set_not_null_stats_from_rows(column, table_stats_rows, None);
        }
        validate_set_not_null_stats_from_rows(
            column,
            table_stats_rows,
            Some(
                self.column_query_opt_bools(tx, &set_not_null_contains_null_query(column))
                    .await?,
            ),
        )
    }

    async fn close_column(
        &self,
        tx: &Tx,
        table_id: TableId,
        column_id: ColumnId,
        new_snapshot: i64,
    ) -> Result<usize> {
        self.column_execute(
            tx,
            &crate::catalog_queries::close_column(table_id, column_id, new_snapshot),
        )
        .await
    }

    async fn close_column_subtree(
        &self,
        tx: &Tx,
        table_id: TableId,
        column_id: ColumnId,
        latest_snapshot: SnapshotId,
        new_snapshot: i64,
    ) -> Result<usize> {
        self.column_execute(
            tx,
            &crate::catalog_queries::close_column_subtree(
                table_id,
                column_id,
                latest_snapshot,
                new_snapshot,
            ),
        )
        .await
    }

    async fn insert_column(&self, tx: &Tx, column: &ColumnEntry, new_snapshot: i64) -> Result<()> {
        self.column_execute(
            tx,
            &crate::catalog_queries::insert_column(column, new_snapshot),
        )
        .await?;
        Ok(())
    }

    async fn insert_type_tree_column(
        &self,
        tx: &Tx,
        table_id: TableId,
        column: &PlannedColumnTypeTreeRow,
        new_snapshot: i64,
    ) -> Result<()> {
        self.column_execute(
            tx,
            &crate::catalog_queries::insert_type_tree_column(table_id, column, new_snapshot),
        )
        .await?;
        Ok(())
    }

    async fn insert_add_column(
        &self,
        tx: &Tx,
        table_id: TableId,
        column: &AddColumn,
        new_snapshot: i64,
    ) -> Result<()> {
        self.column_execute(
            tx,
            &crate::catalog_queries::insert_add_column(table_id, column, new_snapshot),
        )
        .await?;
        Ok(())
    }

    async fn upsert_table_column_stats(&self, tx: &Tx, stats: &TableColumnStats) -> Result<()> {
        let existing = self
            .column_query_i64s(tx, &table_column_stats_count_query(stats))
            .await?;
        let query = table_column_stats_upsert_query(stats, existing);
        self.column_execute(tx, &query).await?;
        Ok(())
    }
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) fn commit_append_files_sync<'tx, O>(
    ops: &O,
    tx: &O::Tx<'tx>,
    files: &[AppendFile],
) -> Result<SnapshotId>
where
    O: SyncTableDataCommitOps,
{
    let append_plan = plan_append_files_commit(files)?;
    let table_id = append_plan.table_id;
    let ids = ops.latest_snapshot_ids(tx)?;

    let metadata_tables = table_data_metadata_tables_sync(
        ops,
        tx,
        "append",
        append_files_have_partition_values(files),
    )?;
    let next_row_id = ops.next_row_id_for_append(tx, table_id, metadata_tables.table_stats)?;
    let snapshot_plan =
        append_plan.plan_snapshot(ids.snapshot_id, ids.next_file_id, next_row_id, files)?;

    insert_planned_data_files_sync(
        ops,
        tx,
        files,
        &snapshot_plan.files,
        metadata_tables.data_file,
        "append",
    )?;
    let column_stats =
        planned_table_column_stats_append_update(table_id, SnapshotId(ids.snapshot_id), files);
    apply_table_column_stats_append_sync(ops, tx, column_stats)?;
    let snapshot_id = finish_snapshot_commit_sync(
        ops,
        tx,
        append_files_snapshot_commit(ids, table_id, &snapshot_plan),
    )?;

    let stats = planned_append_table_stats_update(table_id, &snapshot_plan);
    apply_append_table_stats_sync(ops, tx, metadata_tables.table_stats, stats)?;

    Ok(snapshot_id)
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) fn commit_append_inlined_data_sync<'tx, O>(
    ops: &O,
    tx: &O::Tx<'tx>,
    append: &InlinedDataAppendCommit,
) -> Result<SnapshotId>
where
    O: SyncTableDataCommitOps,
{
    ops.validate_inlined_append(
        append.rows_appended,
        &append.schema,
        &append.batches,
        None,
        "DuckLake inlined append",
    )?;
    let ids = ops.latest_snapshot_ids(tx)?;
    let has_table_stats = ops.table_exists(tx, "ducklake_table_stats")?;
    let plan = planned_inlined_append_commit(
        ids,
        append,
        ops.table_exists(tx, "ducklake_inlined_data_tables")?,
        has_table_stats,
        ops.next_row_id_for_append(tx, append.table_id, has_table_stats)?,
    )?;
    let batch_insert = planned_inlined_append_batch_insert(append, &plan);
    insert_planned_inlined_batches_sync(ops, tx, &[batch_insert])?;
    let column_stats_files = inlined_append_column_stats_files(
        append.table_id,
        SnapshotId(ids.snapshot_id),
        [(append.rows_appended, append.column_stats.clone())],
    );
    let column_stats = planned_table_column_stats_append_update(
        append.table_id,
        SnapshotId(ids.snapshot_id),
        &column_stats_files,
    );
    apply_table_column_stats_append_sync(ops, tx, column_stats)?;

    let snapshot_id = finish_snapshot_commit_sync(
        ops,
        tx,
        inlined_append_snapshot_commit(ids, append, plan.new_snapshot, plan.next_file_id),
    )?;
    let stats = planned_inlined_append_table_stats_update(
        append.table_id,
        plan.row_id_start,
        append.rows_appended,
    )?;
    apply_inlined_append_table_stats_sync(ops, tx, plan.has_table_stats, stats)?;

    Ok(snapshot_id)
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) fn commit_delete_files_sync<'tx, O>(
    ops: &O,
    tx: &O::Tx<'tx>,
    files: &[DeleteFileCommit],
) -> Result<SnapshotId>
where
    O: SyncTableDataCommitOps,
{
    let delete_plan = plan_delete_files_commit(files)?;
    let table_id = delete_plan.table_id;
    let ids = ops.latest_snapshot_ids(tx)?;
    let snapshot_plan = delete_plan.plan_snapshot(ids.snapshot_id, ids.next_file_id, files)?;
    let new_snapshot = snapshot_plan.new_snapshot.0;

    insert_planned_delete_files_sync(ops, tx, files, new_snapshot)?;
    let snapshot_id = finish_snapshot_commit_sync(
        ops,
        tx,
        delete_files_snapshot_commit(ids, table_id, &snapshot_plan),
    )?;
    let stats = planned_delete_table_stats_update(table_id, &snapshot_plan);
    apply_delete_table_stats_sync(ops, tx, stats)?;

    Ok(snapshot_id)
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) fn commit_table_changes_sync<'tx, O>(
    ops: &O,
    tx: &O::Tx<'tx>,
    changes: &TableChangeCommit,
) -> Result<SnapshotId>
where
    O: SyncTableDataCommitOps,
{
    let table_change_plan = plan_table_change_commit(changes)?;
    for append in &changes.inlined_data_appends {
        validate_table_change_inlined_append(ops, append)?;
    }

    let ids = ops.latest_snapshot_ids(tx)?;
    let metadata_tables = table_data_metadata_tables_sync(
        ops,
        tx,
        "table-change",
        table_change_has_partition_values(changes),
    )?;
    let next_row_id =
        ops.next_row_id_for_append(tx, changes.table_id, metadata_tables.table_stats)?;
    let snapshot_plan =
        table_change_plan.plan_snapshot(ids.snapshot_id, ids.next_file_id, next_row_id, changes)?;
    let new_snapshot = snapshot_plan.new_snapshot.0;

    insert_planned_data_files_sync(
        ops,
        tx,
        &changes.append_files,
        &snapshot_plan.append_files,
        metadata_tables.data_file,
        "table-change",
    )?;
    let column_stats = planned_table_column_stats_append_update(
        changes.table_id,
        SnapshotId(ids.snapshot_id),
        &changes.append_files,
    );
    apply_table_column_stats_append_sync(ops, tx, column_stats)?;

    let batch_inserts = planned_table_change_inlined_batch_inserts(
        changes,
        &snapshot_plan.inlined_data_appends,
        new_snapshot,
    )?;
    insert_planned_inlined_batches_sync(ops, tx, &batch_inserts)?;
    let inlined_column_stats_files = inlined_append_column_stats_files(
        changes.table_id,
        SnapshotId(ids.snapshot_id),
        changes
            .inlined_data_appends
            .iter()
            .map(|append| (append.rows_appended, append.column_stats.clone())),
    );
    let inlined_column_stats = planned_table_column_stats_append_update(
        changes.table_id,
        SnapshotId(ids.snapshot_id),
        &inlined_column_stats_files,
    );
    apply_table_column_stats_append_sync(ops, tx, inlined_column_stats)?;

    insert_planned_delete_files_sync(ops, tx, &changes.delete_files, new_snapshot)?;
    let inlined_mutations = planned_table_change_inlined_mutations(changes, ids, &snapshot_plan);
    ops.write_inlined_file_deletions(
        tx,
        inlined_mutations.table_id,
        inlined_mutations.inlined_file_deletions,
        inlined_mutations.new_snapshot,
    )?;
    let inlined_data_deleted = ops.apply_inlined_data_deletions(
        tx,
        inlined_mutations.inlined_data_deletions,
        inlined_mutations.new_snapshot,
    )?;
    let flushed_inlined_data_deleted = ops.delete_flushed_inlined_data(
        tx,
        inlined_mutations.flushed_inlined_data,
        inlined_mutations.previous_snapshot,
    )?;
    ops.delete_flushed_inlined_file_deletions(
        tx,
        inlined_mutations.table_id,
        inlined_mutations.flushed_inlined_file_deletions,
    )?;
    let total_delete_count = planned_table_change_total_delete_count(
        &inlined_mutations,
        inlined_data_deleted,
        flushed_inlined_data_deleted,
    )?;

    let snapshot_id = finish_snapshot_commit_sync(
        ops,
        tx,
        table_change_snapshot_commit(ids, changes, &snapshot_plan),
    )?;
    let stats =
        planned_table_change_stats_update(changes.table_id, total_delete_count, &snapshot_plan);
    apply_table_change_table_stats_sync(ops, tx, metadata_tables.table_stats, stats)?;

    Ok(snapshot_id)
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) fn commit_truncate_table_sync<'tx, O>(
    ops: &O,
    tx: &O::Tx<'tx>,
    truncate: &TruncateTableCommit,
) -> Result<SnapshotId>
where
    O: SyncMaintenanceCommitOps,
{
    let truncate_plan = plan_truncate_table_commit(truncate)?;
    let ids = ops.latest_snapshot_ids(tx)?;
    let snapshot_plan = truncate_plan.plan_snapshot(ids.snapshot_id, ids.next_file_id)?;
    let effects = planned_truncate_table_effects(truncate, &snapshot_plan);

    apply_truncate_table_effects_sync(ops, tx, &truncate_plan, &effects)?;

    let snapshot_id = finish_snapshot_commit_sync(
        ops,
        tx,
        truncate_snapshot_commit(ids, truncate, &snapshot_plan),
    )?;
    let stats_reset = planned_truncate_table_stats_reset(&effects);
    apply_truncate_table_stats_sync(ops, tx, stats_reset)?;

    Ok(snapshot_id)
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) fn commit_merge_adjacent_compaction_sync<'tx, O>(
    ops: &O,
    tx: &O::Tx<'tx>,
    compaction: &MergeAdjacentCompactionCommit,
) -> Result<SnapshotId>
where
    O: SyncMaintenanceCommitOps,
{
    let source_ids = plan_merge_adjacent_compaction_sources(compaction)?;
    let ids = ops.latest_snapshot_ids(tx)?;
    let source_files =
        ops.source_data_files(tx, compaction.table_id, ids.snapshot_id, &source_ids)?;
    require_operation_table(
        "merge-adjacent compaction",
        "ducklake_files_scheduled_for_deletion",
        ops.table_exists(tx, "ducklake_files_scheduled_for_deletion")?,
    )?;
    require_file_partition_value_table(
        "merge-adjacent compaction",
        compaction.has_partition_values(),
        ops.table_exists(tx, "ducklake_file_partition_value")?,
    )?;

    let plan = plan_merge_adjacent_compaction_commit(
        compaction,
        catalog_ids(ids),
        &source_ids,
        source_files.len(),
        source_files.iter().map(|file| file.file_size_bytes),
    )?;
    let cleanup = planned_maintenance_cleanup(compaction.table_id, &source_ids, &source_files);
    apply_maintenance_cleanup_sync(ops, tx, ids.snapshot_id, &cleanup)?;

    finish_maintenance_commit_sync(
        ops,
        tx,
        compaction.table_id,
        &compaction.compacted_files,
        plan,
    )
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) fn commit_rewrite_data_files_sync<'tx, O>(
    ops: &O,
    tx: &O::Tx<'tx>,
    rewrite: &RewriteDataFilesCommit,
) -> Result<SnapshotId>
where
    O: SyncMaintenanceCommitOps,
{
    let source_ids = plan_rewrite_data_files_sources(rewrite)?;
    let ids = ops.latest_snapshot_ids(tx)?;
    let source_files = ops.source_data_files(tx, rewrite.table_id, ids.snapshot_id, &source_ids)?;
    require_file_partition_value_table(
        "rewrite-data-files",
        rewrite.has_partition_values(),
        ops.table_exists(tx, "ducklake_file_partition_value")?,
    )?;

    let plan = plan_rewrite_data_files_commit(
        rewrite,
        catalog_ids(ids),
        &source_ids,
        source_files.len(),
        source_files.iter().map(|file| file.file_size_bytes),
    )?;
    let close = planned_rewrite_close(
        rewrite.table_id,
        &source_ids,
        plan.new_snapshot,
        ops.table_exists(tx, "ducklake_delete_file")?,
    );
    apply_rewrite_close_sync(ops, tx, &close)?;

    finish_maintenance_commit_sync(ops, tx, rewrite.table_id, &rewrite.rewritten_files, plan)
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
fn finish_maintenance_commit_sync<'tx, O>(
    ops: &O,
    tx: &O::Tx<'tx>,
    table_id: TableId,
    output_files: &[AppendFile],
    plan: MaintenanceCommitPlan,
) -> Result<SnapshotId>
where
    O: SyncMaintenanceCommitOps,
{
    let finish = planned_maintenance_finish(table_id, output_files, &plan);
    apply_maintenance_output_files_sync(ops, tx, &finish)?;
    let snapshot_id = finish_snapshot_commit_sync(ops, tx, maintenance_snapshot_commit(&plan))?;
    apply_maintenance_table_stats_sync(ops, tx, finish.table_stats)?;

    Ok(snapshot_id)
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) fn commit_create_schema_sync<'tx, O>(
    ops: &O,
    tx: &O::Tx<'tx>,
    create: &CreateSchemaCommit,
) -> Result<SnapshotId>
where
    O: SyncCatalogObjectCommitOps,
{
    require_catalog_table_exists("ducklake_schema", ops.table_exists(tx, "ducklake_schema")?)?;
    let ids = ops.latest_snapshot_ids(tx)?;
    let snapshot = SnapshotId(ids.snapshot_id);
    ensure_schema_absent(
        &create.schema_name,
        ops.visible_schema(tx, &create.schema_name, snapshot)?,
    )?;
    let plan = plan_create_schema_catalog_mutation(create, catalog_ids(ids))?;
    let schema_id = require_allocated_catalog_id(&plan, "CREATE SCHEMA", "schema")?;
    let schema = planned_schema_row(schema_id, plan.new_snapshot, create);
    apply_create_schema_sync(ops, tx, &schema)?;
    finish_catalog_object_mutation_sync(ops, tx, &plan)
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) fn commit_create_table_sync<'tx, O>(
    ops: &O,
    tx: &O::Tx<'tx>,
    create: &CreateTableCommit,
) -> Result<SnapshotId>
where
    O: SyncCatalogObjectCommitOps,
{
    let ids = ops.latest_snapshot_ids(tx)?;
    let snapshot = SnapshotId(ids.snapshot_id);
    let schema = create_object_target_schema_sync(
        ops,
        tx,
        &create.schema_name,
        &create.table_name,
        snapshot,
    )?;
    let plan = plan_create_table_catalog_mutation(create, catalog_ids(ids))?;
    let table_id = require_allocated_catalog_id(&plan, "CREATE TABLE", "table")?;
    let table = planned_table_insert(table_id, plan.new_snapshot, schema.schema_id, create);
    apply_create_table_sync(ops, tx, &table)?;
    finish_catalog_object_mutation_sync(ops, tx, &plan)
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) fn commit_create_table_with_files_sync<'tx, O>(
    ops: &O,
    tx: &O::Tx<'tx>,
    create: &CreateTableWithFilesCommit,
) -> Result<SnapshotId>
where
    O: SyncCreateTableWithFilesCommitOps,
{
    let ids = ops.latest_snapshot_ids(tx)?;
    let snapshot = SnapshotId(ids.snapshot_id);
    let schema = create_object_target_schema_sync(
        ops,
        tx,
        &create.create.schema_name,
        &create.create.table_name,
        snapshot,
    )?;
    let metadata_tables = table_data_metadata_tables_sync(
        ops,
        tx,
        "CREATE TABLE AS",
        append_files_have_partition_values(&create.files),
    )?;
    let plan =
        plan_create_table_with_files_commit(create, catalog_ids(ids), metadata_tables.table_stats)?;
    let table = planned_table_insert(
        create.table_id.0,
        plan.catalog.new_snapshot,
        schema.schema_id,
        &create.create,
    );
    apply_create_table_sync(ops, tx, &table)?;
    insert_planned_data_files_sync(
        ops,
        tx,
        &create.files,
        &plan.files,
        metadata_tables.data_file,
        "CREATE TABLE AS",
    )?;
    let column_stats = planned_table_column_stats_append_update(
        create.table_id,
        SnapshotId(plan.catalog.new_snapshot),
        &create.files,
    );
    apply_table_column_stats_append_sync(ops, tx, column_stats)?;
    let snapshot_id = finish_snapshot_commit_sync(
        ops,
        tx,
        catalog_object_snapshot_commit_with_next_file_id(&plan.catalog, plan.new_next_file_id),
    )?;
    let stats = planned_create_table_stats_insert(create.table_id, &plan);
    apply_create_table_stats_sync(ops, tx, metadata_tables.table_stats, stats)?;
    Ok(snapshot_id)
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) fn commit_create_table_with_inlined_data_sync<'tx, O>(
    ops: &O,
    tx: &O::Tx<'tx>,
    create: &CreateTableWithInlinedDataCommit,
) -> Result<SnapshotId>
where
    O: SyncCreateTableWithFilesCommitOps,
{
    let ids = ops.latest_snapshot_ids(tx)?;
    let snapshot = SnapshotId(ids.snapshot_id);
    let schema_entry = create_object_target_schema_sync(
        ops,
        tx,
        &create.create.schema_name,
        &create.create.table_name,
        snapshot,
    )?;
    let plan = plan_create_table_with_inlined_data_commit(
        create,
        catalog_ids(ids),
        ops.table_exists(tx, "ducklake_inlined_data_tables")?,
    )?;
    let table_schema = create_table_schema_from_commit(
        create.table_id,
        plan.catalog.new_snapshot,
        &create.create.columns,
    )?;
    ops.validate_inlined_append(
        create.rows_inserted,
        &table_schema,
        &create.batches,
        None,
        "DuckLake CREATE TABLE AS inlined commit",
    )?;
    let table = planned_table_insert(
        create.table_id.0,
        plan.catalog.new_snapshot,
        schema_entry.schema_id,
        &create.create,
    );
    apply_create_table_sync(ops, tx, &table)?;
    let batch_insert = planned_create_table_inlined_batch_insert(
        create,
        &table_schema,
        plan.catalog.new_schema_version,
        plan.catalog.new_snapshot,
    );
    insert_planned_inlined_batches_sync(ops, tx, &[batch_insert])?;
    let column_stats_files = inlined_append_column_stats_files(
        create.table_id,
        SnapshotId(plan.catalog.new_snapshot),
        [(plan.total_record_count, create.column_stats.clone())],
    );
    let column_stats = planned_table_column_stats_append_update(
        create.table_id,
        SnapshotId(plan.catalog.new_snapshot),
        &column_stats_files,
    );
    apply_table_column_stats_append_sync(ops, tx, column_stats)?;
    let snapshot_id =
        finish_snapshot_commit_sync(ops, tx, catalog_object_snapshot_commit(&plan.catalog))?;
    let stats =
        planned_inlined_append_table_stats_update(create.table_id, 0, plan.total_record_count)?;
    apply_inlined_append_table_stats_sync(
        ops,
        tx,
        ops.table_exists(tx, "ducklake_table_stats")?,
        stats,
    )?;
    Ok(snapshot_id)
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) fn commit_create_view_sync<'tx, O>(
    ops: &O,
    tx: &O::Tx<'tx>,
    create: &CreateViewCommit,
) -> Result<SnapshotId>
where
    O: SyncCatalogObjectCommitOps,
{
    let ids = ops.latest_snapshot_ids(tx)?;
    let snapshot = SnapshotId(ids.snapshot_id);
    let schema = create_object_target_schema_sync(
        ops,
        tx,
        &create.schema_name,
        &create.view_name,
        snapshot,
    )?;
    let plan = plan_create_view_catalog_mutation(create, catalog_ids(ids))?;
    let view_id = require_allocated_catalog_id(&plan, "CREATE VIEW", "view")?;
    let view = planned_view_row(view_id, plan.new_snapshot, schema.schema_id, create);
    apply_create_view_sync(ops, tx, &view)?;
    finish_catalog_object_mutation_sync(ops, tx, &plan)
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) fn commit_drop_table_sync<'tx, O>(
    ops: &O,
    tx: &O::Tx<'tx>,
    drop: &DropTableCommit,
) -> Result<SnapshotId>
where
    O: SyncCatalogObjectCommitOps,
{
    let ids = ops.latest_snapshot_ids(tx)?;
    let snapshot = SnapshotId(ids.snapshot_id);
    let table = drop_table_target_sync(ops, tx, drop, snapshot)?;
    let plan = plan_drop_table_catalog_mutation(drop, catalog_ids(ids), table.table_id)?;
    apply_drop_table_sync(ops, tx, table.table_id, plan.new_snapshot)?;
    finish_catalog_object_mutation_sync(ops, tx, &plan)
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) fn commit_rename_table_sync<'tx, O>(
    ops: &O,
    tx: &O::Tx<'tx>,
    rename: &RenameTableCommit,
) -> Result<SnapshotId>
where
    O: SyncCatalogObjectCommitOps,
{
    let ids = ops.latest_snapshot_ids(tx)?;
    let snapshot = SnapshotId(ids.snapshot_id);
    let facts = rename_table_facts_sync(ops, tx, rename, snapshot)?;
    let plan =
        plan_rename_table_catalog_mutation(rename, catalog_ids(ids), &facts.target_schema_name)?;
    let renamed_table = planned_renamed_table_row(
        &facts.data_path,
        &facts.source_schema,
        &facts.target_schema,
        &facts.table,
        rename,
        plan.new_snapshot,
    );
    apply_rename_table_sync(
        ops,
        tx,
        facts.table.table_id,
        plan.new_snapshot,
        &renamed_table,
    )?;
    finish_catalog_object_mutation_sync(ops, tx, &plan)
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) fn commit_drop_view_sync<'tx, O>(
    ops: &O,
    tx: &O::Tx<'tx>,
    drop: &DropViewCommit,
) -> Result<SnapshotId>
where
    O: SyncCatalogObjectCommitOps,
{
    let ids = ops.latest_snapshot_ids(tx)?;
    let snapshot = SnapshotId(ids.snapshot_id);
    let view = drop_view_target_sync(ops, tx, drop, snapshot)?;
    let plan = plan_drop_view_catalog_mutation(drop, catalog_ids(ids), view.view_id)?;
    apply_drop_view_sync(ops, tx, view.view_id, plan.new_snapshot)?;
    finish_catalog_object_mutation_sync(ops, tx, &plan)
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) fn commit_drop_schema_sync<'tx, O>(
    ops: &O,
    tx: &O::Tx<'tx>,
    drop: &DropSchemaCommit,
) -> Result<SnapshotId>
where
    O: SyncCatalogObjectCommitOps,
{
    let ids = ops.latest_snapshot_ids(tx)?;
    let snapshot = SnapshotId(ids.snapshot_id);
    let schema = require_schema(
        ops.visible_schema(tx, &drop.schema_name, snapshot)?,
        &drop.schema_name,
    )?;
    let table_ids = ops.visible_table_ids_for_schema(tx, schema.schema_id, snapshot)?;
    let view_ids = ops.visible_view_ids_for_schema(tx, schema.schema_id, snapshot)?;
    let close_views = ops.table_exists(tx, "ducklake_view")?;
    let planned = planned_drop_schema(
        drop,
        ids,
        schema.schema_id,
        snapshot,
        table_ids,
        &view_ids,
        close_views,
    )?;
    apply_planned_drop_schema_sync(ops, tx, &planned)?;
    finish_catalog_object_mutation_sync(ops, tx, &planned.plan)
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
fn finish_catalog_object_mutation_sync<'tx, O>(
    ops: &O,
    tx: &O::Tx<'tx>,
    plan: &crate::CatalogObjectMutationPlan,
) -> Result<SnapshotId>
where
    O: SyncCatalogObjectCommitOps,
{
    finish_snapshot_commit_sync(ops, tx, catalog_object_snapshot_commit(plan))
}

fn catalog_ids(ids: CommitSnapshotIds) -> CatalogSnapshotIds {
    CatalogSnapshotIds {
        snapshot_id: ids.snapshot_id,
        schema_version: ids.schema_version,
        next_catalog_id: ids.next_catalog_id,
        next_file_id: ids.next_file_id,
    }
}

fn require_schema(schema: Option<SchemaEntry>, schema_name: &str) -> Result<SchemaEntry> {
    schema.ok_or_else(|| {
        StorageError::Catalog(ducklake_catalog::CatalogError::NotFound(format!(
            "schema {schema_name}"
        )))
    })
}

fn require_table(
    table: Option<TableEntry>,
    schema_name: &str,
    table_name: &str,
) -> Result<TableEntry> {
    table.ok_or_else(|| {
        StorageError::Catalog(ducklake_catalog::CatalogError::NotFound(format!(
            "table {schema_name}.{table_name}"
        )))
    })
}

fn require_view(view: Option<ViewEntry>, schema_name: &str, view_name: &str) -> Result<ViewEntry> {
    view.ok_or_else(|| {
        StorageError::Catalog(ducklake_catalog::CatalogError::NotFound(format!(
            "view {schema_name}.{view_name}"
        )))
    })
}

fn require_catalog_table_exists(table_name: &str, exists: bool) -> Result<()> {
    if !exists {
        return Err(StorageError::Backend(format!(
            "DuckLake catalog does not contain {table_name}"
        )));
    }
    Ok(())
}

fn ensure_schema_absent(schema_name: &str, schema: Option<SchemaEntry>) -> Result<()> {
    if schema.is_some() {
        return Err(StorageError::Backend(format!(
            "DuckLake schema {schema_name} already exists"
        )));
    }
    Ok(())
}

fn require_allocated_catalog_id(
    plan: &crate::CatalogObjectMutationPlan,
    operation: &str,
    object: &str,
) -> Result<i64> {
    plan.allocated_catalog_id.ok_or_else(|| {
        StorageError::Backend(format!("DuckLake {operation} did not allocate {object} id"))
    })
}

fn table_data_metadata_tables_for_operation(
    operation: &str,
    has_partition_values: bool,
    has_partition_value_table: bool,
    has_file_column_stats: bool,
    has_file_variant_stats: bool,
    has_table_stats: bool,
) -> Result<TableDataMetadataTables> {
    require_file_partition_value_table(operation, has_partition_values, has_partition_value_table)?;
    Ok(TableDataMetadataTables::new(
        has_file_column_stats,
        has_file_variant_stats,
        has_table_stats,
    ))
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
fn table_data_metadata_tables_sync<'tx, O>(
    ops: &O,
    tx: &O::Tx<'tx>,
    operation: &str,
    has_partition_values: bool,
) -> Result<TableDataMetadataTables>
where
    O: SyncTableDataCommitOps,
{
    table_data_metadata_tables_for_operation(
        operation,
        has_partition_values,
        ops.table_exists(tx, "ducklake_file_partition_value")?,
        ops.table_exists(tx, "ducklake_file_column_stats")?,
        ops.table_exists(tx, "ducklake_file_variant_stats")?,
        ops.table_exists(tx, "ducklake_table_stats")?,
    )
}

#[cfg(any(feature = "postgres", feature = "quack"))]
async fn table_data_metadata_tables_async<O, Tx>(
    ops: &O,
    tx: &Tx,
    operation: &str,
    has_partition_values: bool,
) -> Result<TableDataMetadataTables>
where
    O: AsyncTableDataCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    table_data_metadata_tables_for_operation(
        operation,
        has_partition_values,
        ops.table_exists(tx, "ducklake_file_partition_value")
            .await?,
        ops.table_exists(tx, "ducklake_file_column_stats").await?,
        ops.table_exists(tx, "ducklake_file_variant_stats").await?,
        ops.table_exists(tx, "ducklake_table_stats").await?,
    )
}

fn append_files_have_partition_values(files: &[AppendFile]) -> bool {
    files.iter().any(|file| !file.partition_values.is_empty())
}

fn table_change_has_partition_values(changes: &TableChangeCommit) -> bool {
    append_files_have_partition_values(&changes.append_files)
}

fn require_file_partition_value_table(
    operation: &str,
    has_partition_values: bool,
    exists: bool,
) -> Result<()> {
    if has_partition_values && !exists {
        return Err(StorageError::Backend(format!(
            "DuckLake {operation} requires ducklake_file_partition_value for partitioned tables"
        )));
    }
    Ok(())
}

fn require_operation_table(operation: &str, table_name: &str, exists: bool) -> Result<()> {
    if !exists {
        return Err(StorageError::Backend(format!(
            "DuckLake {operation} requires {table_name}"
        )));
    }
    Ok(())
}

fn require_partition_metadata_tables(
    partition_info_exists: bool,
    partition_column_exists: bool,
) -> Result<()> {
    if !partition_info_exists || !partition_column_exists {
        return Err(StorageError::Backend(
            "DuckLake catalog does not contain partition metadata tables".into(),
        ));
    }
    Ok(())
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
fn require_partition_metadata_tables_sync<'tx, O>(ops: &O, tx: &O::Tx<'tx>) -> Result<()>
where
    O: SyncPartitioningCommitOps,
{
    require_partition_metadata_tables(
        ops.table_exists(tx, "ducklake_partition_info")?,
        ops.table_exists(tx, "ducklake_partition_column")?,
    )
}

#[cfg(any(feature = "postgres", feature = "quack"))]
async fn require_partition_metadata_tables_async<O, Tx>(ops: &O, tx: &Tx) -> Result<()>
where
    O: AsyncPartitioningCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    require_partition_metadata_tables(
        ops.table_exists(tx, "ducklake_partition_info").await?,
        ops.table_exists(tx, "ducklake_partition_column").await?,
    )
}

fn require_sort_metadata_tables(
    sort_info_exists: bool,
    sort_expression_exists: bool,
) -> Result<()> {
    if !sort_info_exists || !sort_expression_exists {
        return Err(StorageError::Backend(
            "DuckLake catalog does not contain sort metadata tables".into(),
        ));
    }
    Ok(())
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
fn require_sort_metadata_tables_sync<'tx, O>(ops: &O, tx: &O::Tx<'tx>) -> Result<()>
where
    O: SyncSortedByCommitOps,
{
    require_sort_metadata_tables(
        ops.table_exists(tx, "ducklake_sort_info")?,
        ops.table_exists(tx, "ducklake_sort_expression")?,
    )
}

#[cfg(any(feature = "postgres", feature = "quack"))]
async fn require_sort_metadata_tables_async<O, Tx>(ops: &O, tx: &Tx) -> Result<()>
where
    O: AsyncSortedByCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    require_sort_metadata_tables(
        ops.table_exists(tx, "ducklake_sort_info").await?,
        ops.table_exists(tx, "ducklake_sort_expression").await?,
    )
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
fn partitioning_table_facts_sync<'tx, O>(
    ops: &O,
    tx: &O::Tx<'tx>,
    partitioning: &PartitioningBatchCommit,
    snapshot: SnapshotId,
) -> Result<Vec<PartitioningTableFacts>>
where
    O: SyncPartitioningCommitOps,
{
    for change in &partitioning.changes {
        ops.validate_partitioning_columns(tx, change.table_id, snapshot, &change.columns)?;
    }
    let final_changes = final_partitioning_changes(&partitioning.changes);
    let mut facts = Vec::with_capacity(final_changes.len());
    for change in final_changes {
        let active = ops.active_partition_specs(tx, change.table_id, snapshot)?;
        facts.push(PartitioningTableFacts {
            change: change.clone(),
            active,
        });
    }
    Ok(facts)
}

#[cfg(any(feature = "postgres", feature = "quack"))]
async fn partitioning_table_facts_async<O, Tx>(
    ops: &O,
    tx: &Tx,
    partitioning: &PartitioningBatchCommit,
    snapshot: SnapshotId,
) -> Result<Vec<PartitioningTableFacts>>
where
    O: AsyncPartitioningCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    for change in &partitioning.changes {
        ops.validate_partitioning_columns(tx, change.table_id, snapshot, &change.columns)
            .await?;
    }
    let final_changes = final_partitioning_changes(&partitioning.changes);
    let mut facts = Vec::with_capacity(final_changes.len());
    for change in final_changes {
        let active = ops
            .active_partition_specs(tx, change.table_id, snapshot)
            .await?;
        facts.push(PartitioningTableFacts {
            change: change.clone(),
            active,
        });
    }
    Ok(facts)
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
fn sorted_by_active_specs_sync<'tx, O>(
    ops: &O,
    tx: &O::Tx<'tx>,
    sorted_by: &SortedByCommit,
    snapshot: SnapshotId,
) -> Result<Vec<ActiveSortSpec>>
where
    O: SyncSortedByCommitOps,
{
    ops.validate_sort_columns(tx, sorted_by.table_id, snapshot, &sorted_by.expressions)?;
    ops.active_sort_specs(tx, sorted_by.table_id, snapshot)
}

#[cfg(any(feature = "postgres", feature = "quack"))]
async fn sorted_by_active_specs_async<O, Tx>(
    ops: &O,
    tx: &Tx,
    sorted_by: &SortedByCommit,
    snapshot: SnapshotId,
) -> Result<Vec<ActiveSortSpec>>
where
    O: AsyncSortedByCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    ops.validate_sort_columns(tx, sorted_by.table_id, snapshot, &sorted_by.expressions)
        .await?;
    ops.active_sort_specs(tx, sorted_by.table_id, snapshot)
        .await
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
fn drop_column_facts_sync<'tx, O>(
    ops: &O,
    tx: &O::Tx<'tx>,
    drop: &DropColumnCommit,
    snapshot: SnapshotId,
) -> Result<DropColumnFacts>
where
    O: SyncColumnEvolutionCommitOps,
{
    let column = ops.visible_column(tx, drop.table_id, drop.column_id, snapshot)?;
    let top_level_column_count = ops.top_level_column_count(tx, drop.table_id, snapshot)?;
    let referenced_by_partition =
        ops.drop_column_referenced_by_partition(tx, drop.table_id, drop.column_id, snapshot)?;
    let referenced_by_sort =
        if let Some(lookup) = drop_column_sort_reference_lookup(drop.table_id, column.as_ref()) {
            ops.drop_column_referenced_by_sort(tx, lookup.table_id, lookup.column_name, snapshot)?
        } else {
            false
        };
    Ok(DropColumnFacts {
        column,
        top_level_column_count,
        referenced_by_partition,
        referenced_by_sort,
    })
}

#[cfg(any(feature = "postgres", feature = "quack"))]
async fn drop_column_facts_async<O, Tx>(
    ops: &O,
    tx: &Tx,
    drop: &DropColumnCommit,
    snapshot: SnapshotId,
) -> Result<DropColumnFacts>
where
    O: AsyncColumnEvolutionCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    let column = ops
        .visible_column(tx, drop.table_id, drop.column_id, snapshot)
        .await?;
    let top_level_column_count = ops
        .top_level_column_count(tx, drop.table_id, snapshot)
        .await?;
    let referenced_by_partition = ops
        .drop_column_referenced_by_partition(tx, drop.table_id, drop.column_id, snapshot)
        .await?;
    let referenced_by_sort =
        if let Some(lookup) = drop_column_sort_reference_lookup(drop.table_id, column.as_ref()) {
            ops.drop_column_referenced_by_sort(tx, lookup.table_id, lookup.column_name, snapshot)
                .await?
        } else {
            false
        };
    Ok(DropColumnFacts {
        column,
        top_level_column_count,
        referenced_by_partition,
        referenced_by_sort,
    })
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
fn apply_drop_column_subtree_close_sync<'tx, O>(
    ops: &O,
    tx: &O::Tx<'tx>,
    plan: &MetadataMutationPlan,
    close: &ColumnSubtreeCloseRequest,
) -> Result<SnapshotId>
where
    O: SyncColumnEvolutionCommitOps,
{
    let closed = ops.close_column_subtree(
        tx,
        close.table_id,
        close.column_id,
        close.latest_snapshot,
        close.new_snapshot,
    )?;
    ensure_closed_subtree(close, closed)?;
    finish_column_metadata_mutation_sync(ops, tx, plan)
}

#[cfg(any(feature = "postgres", feature = "quack"))]
async fn apply_drop_column_subtree_close_async<O, Tx>(
    ops: &O,
    tx: &Tx,
    plan: &MetadataMutationPlan,
    close: &ColumnSubtreeCloseRequest,
) -> Result<SnapshotId>
where
    O: AsyncColumnEvolutionCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    let closed = ops
        .close_column_subtree(
            tx,
            close.table_id,
            close.column_id,
            close.latest_snapshot,
            close.new_snapshot,
        )
        .await?;
    ensure_closed_subtree(close, closed)?;
    finish_column_metadata_mutation_async(ops, tx, plan).await
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
fn add_columns_plan_sync<'tx, O>(
    ops: &O,
    tx: &O::Tx<'tx>,
    add: &AddColumnCommit,
    ids: CommitSnapshotIds,
    snapshot: SnapshotId,
) -> Result<MetadataMutationPlan>
where
    O: SyncColumnEvolutionCommitOps,
{
    let active_rows = ops.active_column_refs(tx, add.table_id, snapshot)?;
    plan_add_columns_commit(
        add,
        catalog_ids(ids),
        ops.table_id_exists(tx, add.table_id, snapshot)?,
        &active_rows,
        ops.table_exists(tx, "ducklake_table_column_stats")?,
    )
}

#[cfg(any(feature = "postgres", feature = "quack"))]
async fn add_columns_plan_async<O, Tx>(
    ops: &O,
    tx: &Tx,
    add: &AddColumnCommit,
    ids: CommitSnapshotIds,
    snapshot: SnapshotId,
) -> Result<MetadataMutationPlan>
where
    O: AsyncColumnEvolutionCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    let active_rows = ops.active_column_refs(tx, add.table_id, snapshot).await?;
    plan_add_columns_commit(
        add,
        catalog_ids(ids),
        ops.table_id_exists(tx, add.table_id, snapshot).await?,
        &active_rows,
        ops.table_exists(tx, "ducklake_table_column_stats").await?,
    )
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
fn rename_column_plan_sync<'tx, O>(
    ops: &O,
    tx: &O::Tx<'tx>,
    rename: &RenameColumnCommit,
    ids: CommitSnapshotIds,
    snapshot: SnapshotId,
) -> Result<PlannedColumnReplacement>
where
    O: SyncColumnEvolutionCommitOps,
{
    let column = ops.visible_column(tx, rename.table_id, rename.column_id, snapshot)?;
    let start = plan_rename_column_mutation_start(rename.clone(), catalog_ids(ids), column)?;
    let conflict_lookup = start.conflict_lookup();
    let conflict = ops.conflicting_sibling_column_name(
        tx,
        conflict_lookup.table_id,
        conflict_lookup.column_id,
        conflict_lookup.parent_column,
        conflict_lookup.new_column_name,
        snapshot,
    )?;
    let next_catalog_id =
        if let Some(rewrite) = start.sort_rewrite_request(snapshot, ids.next_catalog_id) {
            ops.rewrite_sort_expressions_for_rename(
                tx,
                rewrite.table_id,
                rewrite.latest_snapshot,
                rewrite.new_snapshot,
                rewrite.next_catalog_id,
                rewrite.old_column_name,
                rewrite.new_column_name,
            )?
        } else {
            ids.next_catalog_id
        };
    start.finish(conflict, next_catalog_id)
}

#[cfg(any(feature = "postgres", feature = "quack"))]
async fn rename_column_plan_async<O, Tx>(
    ops: &O,
    tx: &Tx,
    rename: &RenameColumnCommit,
    ids: CommitSnapshotIds,
    snapshot: SnapshotId,
) -> Result<PlannedColumnReplacement>
where
    O: AsyncColumnEvolutionCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    let column = ops
        .visible_column(tx, rename.table_id, rename.column_id, snapshot)
        .await?;
    let start = plan_rename_column_mutation_start(rename.clone(), catalog_ids(ids), column)?;
    let conflict_lookup = start.conflict_lookup();
    let conflict = ops
        .conflicting_sibling_column_name(
            tx,
            conflict_lookup.table_id,
            conflict_lookup.column_id,
            conflict_lookup.parent_column,
            conflict_lookup.new_column_name,
            snapshot,
        )
        .await?;
    let next_catalog_id =
        if let Some(rewrite) = start.sort_rewrite_request(snapshot, ids.next_catalog_id) {
            ops.rewrite_sort_expressions_for_rename(
                tx,
                rewrite.table_id,
                rewrite.latest_snapshot,
                rewrite.new_snapshot,
                rewrite.next_catalog_id,
                rewrite.old_column_name,
                rewrite.new_column_name,
            )
            .await?
        } else {
            ids.next_catalog_id
        };
    start.finish(conflict, next_catalog_id)
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
fn column_type_tree_change_sync<'tx, O>(
    ops: &O,
    tx: &O::Tx<'tx>,
    change: &ColumnTypeTreeCommit,
    ids: CommitSnapshotIds,
    snapshot: SnapshotId,
) -> Result<PlannedColumnTypeTreeChange>
where
    O: SyncColumnEvolutionCommitOps,
{
    let active_columns = ops.visible_columns(tx, change.table_id, snapshot)?;
    let schema = try_build_schema(&active_columns, snapshot).map_err(StorageError::Catalog)?;
    let root = schema.field(change.column_id).cloned();
    plan_column_type_tree_mutation(change.clone(), catalog_ids(ids), root, &active_columns)
}

#[cfg(any(feature = "postgres", feature = "quack"))]
async fn column_type_tree_change_async<O, Tx>(
    ops: &O,
    tx: &Tx,
    change: &ColumnTypeTreeCommit,
    ids: CommitSnapshotIds,
    snapshot: SnapshotId,
) -> Result<PlannedColumnTypeTreeChange>
where
    O: AsyncColumnEvolutionCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    let active_columns = ops.visible_columns(tx, change.table_id, snapshot).await?;
    let schema = try_build_schema(&active_columns, snapshot).map_err(StorageError::Catalog)?;
    let root = schema.field(change.column_id).cloned();
    plan_column_type_tree_mutation(change.clone(), catalog_ids(ids), root, &active_columns)
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
fn column_nullability_replacement_sync<'tx, O>(
    ops: &O,
    tx: &O::Tx<'tx>,
    change: &ColumnNullabilityCommit,
    ids: CommitSnapshotIds,
    snapshot: SnapshotId,
) -> Result<PlannedColumnReplacement>
where
    O: SyncColumnEvolutionCommitOps,
{
    let column = ops.visible_column(tx, change.table_id, change.column_id, snapshot)?;
    let plan = plan_column_nullability_mutation(change.clone(), catalog_ids(ids), column)?;
    if plan.requires_existing_value_validation {
        ops.validate_set_not_null_stats(tx, &plan.replacement.column)?;
    }
    Ok(plan.replacement)
}

#[cfg(any(feature = "postgres", feature = "quack"))]
async fn column_nullability_replacement_async<O, Tx>(
    ops: &O,
    tx: &Tx,
    change: &ColumnNullabilityCommit,
    ids: CommitSnapshotIds,
    snapshot: SnapshotId,
) -> Result<PlannedColumnReplacement>
where
    O: AsyncColumnEvolutionCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    let column = ops
        .visible_column(tx, change.table_id, change.column_id, snapshot)
        .await?;
    let plan = plan_column_nullability_mutation(change.clone(), catalog_ids(ids), column)?;
    if plan.requires_existing_value_validation {
        ops.validate_set_not_null_stats(tx, &plan.replacement.column)
            .await?;
    }
    Ok(plan.replacement)
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
fn column_default_replacement_sync<'tx, O>(
    ops: &O,
    tx: &O::Tx<'tx>,
    change: &ColumnDefaultCommit,
    ids: CommitSnapshotIds,
    snapshot: SnapshotId,
) -> Result<PlannedColumnReplacement>
where
    O: SyncColumnEvolutionCommitOps,
{
    let column = ops.visible_column(tx, change.table_id, change.column_id, snapshot)?;
    plan_column_default_mutation(change.clone(), catalog_ids(ids), column)
}

#[cfg(any(feature = "postgres", feature = "quack"))]
async fn column_default_replacement_async<O, Tx>(
    ops: &O,
    tx: &Tx,
    change: &ColumnDefaultCommit,
    ids: CommitSnapshotIds,
    snapshot: SnapshotId,
) -> Result<PlannedColumnReplacement>
where
    O: AsyncColumnEvolutionCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    let column = ops
        .visible_column(tx, change.table_id, change.column_id, snapshot)
        .await?;
    plan_column_default_mutation(change.clone(), catalog_ids(ids), column)
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
fn column_type_change_sync<'tx, O>(
    ops: &O,
    tx: &O::Tx<'tx>,
    change: &ColumnTypeCommit,
    ids: CommitSnapshotIds,
    snapshot: SnapshotId,
) -> Result<PlannedColumnTypeChange>
where
    O: SyncColumnEvolutionCommitOps,
{
    let column = ops.visible_column(tx, change.table_id, change.column_id, snapshot)?;
    plan_column_type_mutation(change.clone(), catalog_ids(ids), column)
}

#[cfg(any(feature = "postgres", feature = "quack"))]
async fn column_type_change_async<O, Tx>(
    ops: &O,
    tx: &Tx,
    change: &ColumnTypeCommit,
    ids: CommitSnapshotIds,
    snapshot: SnapshotId,
) -> Result<PlannedColumnTypeChange>
where
    O: AsyncColumnEvolutionCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    let column = ops
        .visible_column(tx, change.table_id, change.column_id, snapshot)
        .await?;
    plan_column_type_mutation(change.clone(), catalog_ids(ids), column)
}

fn ensure_table_or_view_absent_from_rows(
    tables: Vec<TableEntry>,
    views: Vec<ViewEntry>,
    schema_name: &str,
    object_name: &str,
) -> Result<()> {
    if !tables.is_empty() || !views.is_empty() {
        return Err(StorageError::Backend(format!(
            "DuckLake table or view {schema_name}.{object_name} already exists"
        )));
    }
    Ok(())
}

fn ensure_rename_target_available_from_rows(
    tables: Vec<TableEntry>,
    views: Vec<ViewEntry>,
    schema_name: &str,
    object_name: &str,
    source_table_id: TableId,
) -> Result<()> {
    for existing in tables {
        if existing.table_id != source_table_id {
            return Err(StorageError::Backend(format!(
                "DuckLake table or view {schema_name}.{object_name} already exists"
            )));
        }
    }
    if !views.is_empty() {
        return Err(StorageError::Backend(format!(
            "DuckLake table or view {schema_name}.{object_name} already exists"
        )));
    }
    Ok(())
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
fn ensure_table_or_view_absent_sync<'tx, O>(
    ops: &O,
    tx: &O::Tx<'tx>,
    schema_id: SchemaId,
    object_name: &str,
    snapshot: SnapshotId,
    schema_name: &str,
) -> Result<()>
where
    O: SyncCatalogObjectCommitOps,
{
    let tables = ops.visible_tables_case_insensitive(tx, schema_id, object_name, snapshot)?;
    let views = if ops.table_exists(tx, "ducklake_view")? {
        ops.visible_views_case_insensitive(tx, schema_id, object_name, snapshot)?
    } else {
        Vec::new()
    };
    ensure_table_or_view_absent_from_rows(tables, views, schema_name, object_name)
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
fn ensure_rename_target_available_sync<'tx, O>(
    ops: &O,
    tx: &O::Tx<'tx>,
    schema_id: SchemaId,
    object_name: &str,
    snapshot: SnapshotId,
    schema_name: &str,
    source_table_id: TableId,
) -> Result<()>
where
    O: SyncCatalogObjectCommitOps,
{
    let tables = ops.visible_tables_case_insensitive(tx, schema_id, object_name, snapshot)?;
    let views = ops.visible_views_case_insensitive(tx, schema_id, object_name, snapshot)?;
    ensure_rename_target_available_from_rows(
        tables,
        views,
        schema_name,
        object_name,
        source_table_id,
    )
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
fn create_object_target_schema_sync<'tx, O>(
    ops: &O,
    tx: &O::Tx<'tx>,
    schema_name: &str,
    object_name: &str,
    snapshot: SnapshotId,
) -> Result<SchemaEntry>
where
    O: SyncCatalogObjectCommitOps,
{
    let schema = require_schema(ops.visible_schema(tx, schema_name, snapshot)?, schema_name)?;
    ensure_table_or_view_absent_sync(
        ops,
        tx,
        schema.schema_id,
        object_name,
        snapshot,
        schema_name,
    )?;
    Ok(schema)
}

#[cfg(any(feature = "postgres", feature = "quack"))]
async fn create_object_target_schema_async<O, Tx>(
    ops: &O,
    tx: &Tx,
    schema_name: &str,
    object_name: &str,
    snapshot: SnapshotId,
) -> Result<SchemaEntry>
where
    O: AsyncCatalogObjectCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    let schema = require_schema(
        ops.visible_schema(tx, schema_name, snapshot).await?,
        schema_name,
    )?;
    ensure_table_or_view_absent_async(
        ops,
        tx,
        schema.schema_id,
        object_name,
        snapshot,
        schema_name,
    )
    .await?;
    Ok(schema)
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
fn drop_table_target_sync<'tx, O>(
    ops: &O,
    tx: &O::Tx<'tx>,
    drop: &DropTableCommit,
    snapshot: SnapshotId,
) -> Result<TableEntry>
where
    O: SyncCatalogObjectCommitOps,
{
    let schema = require_schema(
        ops.visible_schema(tx, &drop.schema_name, snapshot)?,
        &drop.schema_name,
    )?;
    require_table(
        ops.visible_table(tx, schema.schema_id, &drop.table_name, snapshot)?,
        &drop.schema_name,
        &drop.table_name,
    )
}

#[cfg(any(feature = "postgres", feature = "quack"))]
async fn drop_table_target_async<O, Tx>(
    ops: &O,
    tx: &Tx,
    drop: &DropTableCommit,
    snapshot: SnapshotId,
) -> Result<TableEntry>
where
    O: AsyncCatalogObjectCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    let schema = require_schema(
        ops.visible_schema(tx, &drop.schema_name, snapshot).await?,
        &drop.schema_name,
    )?;
    require_table(
        ops.visible_table(tx, schema.schema_id, &drop.table_name, snapshot)
            .await?,
        &drop.schema_name,
        &drop.table_name,
    )
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
fn drop_view_target_sync<'tx, O>(
    ops: &O,
    tx: &O::Tx<'tx>,
    drop: &DropViewCommit,
    snapshot: SnapshotId,
) -> Result<ViewEntry>
where
    O: SyncCatalogObjectCommitOps,
{
    let schema = require_schema(
        ops.visible_schema(tx, &drop.schema_name, snapshot)?,
        &drop.schema_name,
    )?;
    require_view(
        ops.visible_view(tx, schema.schema_id, &drop.view_name, snapshot)?,
        &drop.schema_name,
        &drop.view_name,
    )
}

#[cfg(any(feature = "postgres", feature = "quack"))]
async fn drop_view_target_async<O, Tx>(
    ops: &O,
    tx: &Tx,
    drop: &DropViewCommit,
    snapshot: SnapshotId,
) -> Result<ViewEntry>
where
    O: AsyncCatalogObjectCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    let schema = require_schema(
        ops.visible_schema(tx, &drop.schema_name, snapshot).await?,
        &drop.schema_name,
    )?;
    require_view(
        ops.visible_view(tx, schema.schema_id, &drop.view_name, snapshot)
            .await?,
        &drop.schema_name,
        &drop.view_name,
    )
}

#[derive(Debug, Clone)]
struct RenameTableFacts {
    source_schema: SchemaEntry,
    target_schema: SchemaEntry,
    table: TableEntry,
    data_path: String,
    target_schema_name: String,
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
fn rename_table_facts_sync<'tx, O>(
    ops: &O,
    tx: &O::Tx<'tx>,
    rename: &RenameTableCommit,
    snapshot: SnapshotId,
) -> Result<RenameTableFacts>
where
    O: SyncCatalogObjectCommitOps,
{
    let source_schema = require_schema(
        ops.visible_schema(tx, &rename.schema_name, snapshot)?,
        &rename.schema_name,
    )?;
    let target_schema_name = rename
        .new_schema_name
        .as_deref()
        .unwrap_or(&rename.schema_name)
        .to_string();
    let target_schema = require_schema(
        ops.visible_schema(tx, &target_schema_name, snapshot)?,
        &target_schema_name,
    )?;
    let table = require_table(
        ops.visible_table(
            tx,
            source_schema.schema_id,
            &rename.old_table_name,
            snapshot,
        )?,
        &rename.schema_name,
        &rename.old_table_name,
    )?;
    ensure_rename_target_available_sync(
        ops,
        tx,
        target_schema.schema_id,
        &rename.new_table_name,
        snapshot,
        &target_schema_name,
        table.table_id,
    )?;
    let data_path = ops.data_path(tx)?;
    Ok(RenameTableFacts {
        source_schema,
        target_schema,
        table,
        data_path,
        target_schema_name,
    })
}

#[cfg(any(feature = "postgres", feature = "quack"))]
async fn rename_table_facts_async<O, Tx>(
    ops: &O,
    tx: &Tx,
    rename: &RenameTableCommit,
    snapshot: SnapshotId,
) -> Result<RenameTableFacts>
where
    O: AsyncCatalogObjectCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    let source_schema = require_schema(
        ops.visible_schema(tx, &rename.schema_name, snapshot)
            .await?,
        &rename.schema_name,
    )?;
    let target_schema_name = rename
        .new_schema_name
        .as_deref()
        .unwrap_or(&rename.schema_name)
        .to_string();
    let target_schema = require_schema(
        ops.visible_schema(tx, &target_schema_name, snapshot)
            .await?,
        &target_schema_name,
    )?;
    let table = require_table(
        ops.visible_table(
            tx,
            source_schema.schema_id,
            &rename.old_table_name,
            snapshot,
        )
        .await?,
        &rename.schema_name,
        &rename.old_table_name,
    )?;
    ensure_rename_target_available_async(
        ops,
        tx,
        target_schema.schema_id,
        &rename.new_table_name,
        snapshot,
        &target_schema_name,
        table.table_id,
    )
    .await?;
    let data_path = ops.data_path(tx).await?;
    Ok(RenameTableFacts {
        source_schema,
        target_schema,
        table,
        data_path,
        target_schema_name,
    })
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) fn commit_column_nullability_sync<'tx, O>(
    ops: &O,
    tx: &O::Tx<'tx>,
    change: &ColumnNullabilityCommit,
) -> Result<SnapshotId>
where
    O: SyncColumnEvolutionCommitOps,
{
    let ids = ops.latest_snapshot_ids(tx)?;
    let snapshot = SnapshotId(ids.snapshot_id);
    let replacement = column_nullability_replacement_sync(ops, tx, change, ids, snapshot)?;
    commit_column_replacement_sync(ops, tx, &replacement, "ALTER COLUMN")
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) fn commit_column_default_sync<'tx, O>(
    ops: &O,
    tx: &O::Tx<'tx>,
    change: &ColumnDefaultCommit,
) -> Result<SnapshotId>
where
    O: SyncColumnEvolutionCommitOps,
{
    let ids = ops.latest_snapshot_ids(tx)?;
    let snapshot = SnapshotId(ids.snapshot_id);
    let plan = column_default_replacement_sync(ops, tx, change, ids, snapshot)?;
    commit_column_replacement_sync(ops, tx, &plan, "ALTER COLUMN DEFAULT")
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) fn commit_column_type_sync<'tx, O>(
    ops: &O,
    tx: &O::Tx<'tx>,
    change: &ColumnTypeCommit,
) -> Result<SnapshotId>
where
    O: SyncColumnEvolutionCommitOps,
{
    let ids = ops.latest_snapshot_ids(tx)?;
    let snapshot = SnapshotId(ids.snapshot_id);
    match column_type_change_sync(ops, tx, change, ids, snapshot)? {
        PlannedColumnTypeChange::Noop { snapshot } => Ok(snapshot),
        PlannedColumnTypeChange::Commit { replacement } => {
            commit_column_replacement_sync(ops, tx, &replacement, "ALTER COLUMN TYPE")
        }
    }
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) fn commit_column_type_tree_sync<'tx, O>(
    ops: &O,
    tx: &O::Tx<'tx>,
    change: &ColumnTypeTreeCommit,
) -> Result<SnapshotId>
where
    O: SyncColumnEvolutionCommitOps,
{
    let ids = ops.latest_snapshot_ids(tx)?;
    let snapshot = SnapshotId(ids.snapshot_id);
    match column_type_tree_change_sync(ops, tx, change, ids, snapshot)? {
        PlannedColumnTypeTreeChange::Noop { snapshot } => Ok(snapshot),
        PlannedColumnTypeTreeChange::Commit {
            metadata,
            close_columns,
            insert_columns,
        } => {
            apply_column_type_tree_change_sync(
                ops,
                tx,
                &metadata,
                &close_columns,
                &insert_columns,
            )?;
            finish_column_metadata_mutation_sync(ops, tx, &metadata)
        }
    }
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) fn commit_rename_column_sync<'tx, O>(
    ops: &O,
    tx: &O::Tx<'tx>,
    rename: &RenameColumnCommit,
) -> Result<SnapshotId>
where
    O: SyncColumnEvolutionCommitOps,
{
    let ids = ops.latest_snapshot_ids(tx)?;
    let snapshot = SnapshotId(ids.snapshot_id);
    let plan = rename_column_plan_sync(ops, tx, rename, ids, snapshot)?;
    commit_column_replacement_sync(ops, tx, &plan, "RENAME COLUMN")
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) fn commit_drop_column_sync<'tx, O>(
    ops: &O,
    tx: &O::Tx<'tx>,
    drop: &DropColumnCommit,
) -> Result<SnapshotId>
where
    O: SyncColumnEvolutionCommitOps,
{
    let ids = ops.latest_snapshot_ids(tx)?;
    let snapshot = SnapshotId(ids.snapshot_id);
    let facts = drop_column_facts_sync(ops, tx, drop, snapshot)?;
    let plan = plan_drop_column_mutation(drop.clone(), catalog_ids(ids), facts)?;
    let close = plan.close_request(snapshot);
    apply_drop_column_subtree_close_sync(ops, tx, &plan.metadata, &close)
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) fn commit_add_columns_sync<'tx, O>(
    ops: &O,
    tx: &O::Tx<'tx>,
    add: &AddColumnCommit,
) -> Result<SnapshotId>
where
    O: SyncColumnEvolutionCommitOps,
{
    let ids = ops.latest_snapshot_ids(tx)?;
    let snapshot = SnapshotId(ids.snapshot_id);
    let plan = add_columns_plan_sync(ops, tx, add, ids, snapshot)?;
    apply_add_columns_sync(ops, tx, add, plan.new_snapshot)?;
    finish_snapshot_commit_sync(ops, tx, metadata_snapshot_commit(&plan))
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) fn commit_partitioning_batch_sync<'tx, O>(
    ops: &O,
    tx: &O::Tx<'tx>,
    partitioning: &PartitioningBatchCommit,
) -> Result<SnapshotId>
where
    O: SyncPartitioningCommitOps,
{
    require_partition_metadata_tables_sync(ops, tx)?;

    let ids = ops.latest_snapshot_ids(tx)?;
    let snapshot = SnapshotId(ids.snapshot_id);
    let facts = partitioning_table_facts_sync(ops, tx, partitioning, snapshot)?;
    let plan = plan_partitioning_batch_commit(partitioning, catalog_ids(ids), facts)?;
    let changes = planned_partition_metadata_changes(&plan)?;
    apply_planned_partition_metadata_changes_sync(ops, tx, &changes)?;

    finish_snapshot_commit_sync(ops, tx, partitioning_snapshot_commit(&plan))
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) fn commit_sorted_by_sync<'tx, O>(
    ops: &O,
    tx: &O::Tx<'tx>,
    sorted_by: &SortedByCommit,
) -> Result<SnapshotId>
where
    O: SyncSortedByCommitOps,
{
    require_sort_metadata_tables_sync(ops, tx)?;

    let ids = ops.latest_snapshot_ids(tx)?;
    let snapshot = SnapshotId(ids.snapshot_id);
    let active = sorted_by_active_specs_sync(ops, tx, sorted_by, snapshot)?;
    let plan = plan_sorted_by_commit(
        sorted_by.table_id,
        sorted_by.expected_snapshot,
        catalog_ids(ids),
        &active,
        &sorted_by.expressions,
        sorted_by.commit_message.clone(),
    )?;
    let SortedByCommitPlan::Change(plan) = plan else {
        return Ok(snapshot);
    };
    let metadata = &plan.metadata;
    let writes = planned_sort_metadata_change(&plan, &sorted_by.expressions)?;
    apply_planned_sort_metadata_change_sync(ops, tx, &writes)?;
    finish_snapshot_commit_sync(ops, tx, metadata_snapshot_commit(metadata))
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
fn commit_column_replacement_sync<'tx, O>(
    ops: &O,
    tx: &O::Tx<'tx>,
    replacement: &PlannedColumnReplacement,
    operation: &'static str,
) -> Result<SnapshotId>
where
    O: SyncColumnEvolutionCommitOps,
{
    let metadata = &replacement.metadata;
    let column = &replacement.column;
    let close = replacement.close_request(operation);
    let closed = ops.close_column(tx, close.table_id, close.column_id, close.new_snapshot)?;
    ensure_closed_one(&close, closed)?;
    ops.insert_column(tx, column, metadata.new_snapshot)?;
    finish_column_metadata_mutation_sync(ops, tx, metadata)
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
fn finish_column_metadata_mutation_sync<'tx, O>(
    ops: &O,
    tx: &O::Tx<'tx>,
    metadata: &MetadataMutationPlan,
) -> Result<SnapshotId>
where
    O: SyncColumnEvolutionCommitOps,
{
    finish_snapshot_commit_sync(ops, tx, metadata_snapshot_commit(metadata))
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
fn apply_column_type_tree_change_sync<'tx, O>(
    ops: &O,
    tx: &O::Tx<'tx>,
    metadata: &MetadataMutationPlan,
    close_columns: &[ColumnCloseRequest],
    insert_columns: &[PlannedColumnTypeTreeRow],
) -> Result<()>
where
    O: SyncColumnEvolutionCommitOps,
{
    for close in close_columns {
        let closed = ops.close_column(tx, close.table_id, close.column_id, close.new_snapshot)?;
        ensure_closed_one(close, closed)?;
    }
    for column in insert_columns {
        ops.insert_type_tree_column(tx, metadata.table_id, column, metadata.new_snapshot)?;
    }
    Ok(())
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
fn apply_add_columns_sync<'tx, O>(
    ops: &O,
    tx: &O::Tx<'tx>,
    add: &AddColumnCommit,
    new_snapshot: i64,
) -> Result<()>
where
    O: SyncColumnEvolutionCommitOps,
{
    for column in &add.columns {
        ops.insert_add_column(tx, add.table_id, column, new_snapshot)?;
    }
    for stats in &add.table_column_stats {
        ops.upsert_table_column_stats(tx, stats)?;
    }
    Ok(())
}

fn ensure_closed_one(close: &ColumnCloseRequest, closed: usize) -> Result<()> {
    if closed != 1 {
        return Err(StorageError::Backend(format!(
            "DuckLake {operation} expected one active column row for table {} column {}, updated {}",
            close.table_id.0,
            close.column_id.0,
            closed,
            operation = close.operation
        )));
    }
    Ok(())
}

fn ensure_closed_subtree(close: &ColumnSubtreeCloseRequest, closed: usize) -> Result<()> {
    if closed < 1 {
        return Err(StorageError::Backend(format!(
            "DuckLake {operation} expected at least one active column row for table {} column {}, updated {}",
            close.table_id.0,
            close.column_id.0,
            closed,
            operation = close.operation
        )));
    }
    Ok(())
}

#[derive(Debug, Clone, Copy)]
struct MetadataCloseRequest {
    operation: &'static str,
    row_kind: &'static str,
    row_id: i64,
}

fn ensure_closed_metadata_row(close: &MetadataCloseRequest, closed: usize) -> Result<()> {
    if closed != 1 {
        return Err(StorageError::Backend(format!(
            "DuckLake {operation} expected one active {row_kind} row {row_id}, updated {closed}",
            operation = close.operation,
            row_kind = close.row_kind,
            row_id = close.row_id,
        )));
    }
    Ok(())
}

#[derive(Debug, Clone, Copy)]
struct PlannedPartitionClose {
    partition_id: ducklake_catalog::PartitionId,
    new_snapshot: i64,
    expectation: MetadataCloseRequest,
}

#[derive(Debug, Clone)]
struct PlannedPartitionMetadataChange<'a> {
    close: Option<PlannedPartitionClose>,
    insert: Option<PlannedPartitionInfoInsert<'a>>,
}

#[derive(Debug, Clone)]
struct PlannedPartitionInfoInsert<'a> {
    partition_id: ducklake_catalog::PartitionId,
    table_id: TableId,
    new_snapshot: i64,
    columns: Vec<PlannedPartitionColumnInsert<'a>>,
}

#[derive(Debug, Clone, Copy)]
struct PlannedPartitionColumnInsert<'a> {
    partition_key_index: i64,
    column: &'a PartitionColumnCommit,
}

fn planned_partition_metadata_changes(
    plan: &crate::PartitioningBatchPlan,
) -> Result<Vec<PlannedPartitionMetadataChange<'_>>> {
    plan.table_changes
        .iter()
        .map(|change| {
            Ok(PlannedPartitionMetadataChange {
                close: change
                    .close_partition_id
                    .map(|partition_id| PlannedPartitionClose {
                        partition_id,
                        new_snapshot: plan.new_snapshot,
                        expectation: MetadataCloseRequest {
                            operation: "SET PARTITIONED BY",
                            row_kind: "partition",
                            row_id: partition_id.0,
                        },
                    }),
                insert: change
                    .insert_partition_id
                    .map(|partition_id| -> Result<PlannedPartitionInfoInsert<'_>> {
                        Ok(PlannedPartitionInfoInsert {
                            partition_id,
                            table_id: change.table_id,
                            new_snapshot: plan.new_snapshot,
                            columns: planned_partition_column_inserts(&change.columns)?,
                        })
                    })
                    .transpose()?,
            })
        })
        .collect()
}

fn planned_partition_column_inserts(
    columns: &[PartitionColumnCommit],
) -> Result<Vec<PlannedPartitionColumnInsert<'_>>> {
    columns
        .iter()
        .enumerate()
        .map(|(idx, column)| {
            Ok(PlannedPartitionColumnInsert {
                partition_key_index: partition_key_index(idx)?,
                column,
            })
        })
        .collect()
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
fn apply_planned_partition_metadata_changes_sync<'tx, O>(
    ops: &O,
    tx: &O::Tx<'tx>,
    changes: &[PlannedPartitionMetadataChange<'_>],
) -> Result<()>
where
    O: SyncPartitioningCommitOps,
{
    for change in changes {
        if let Some(close) = change.close {
            let closed = ops.close_partition(tx, close.partition_id, close.new_snapshot)?;
            ensure_closed_metadata_row(&close.expectation, closed)?;
        }
        if let Some(insert) = &change.insert {
            ops.insert_partition_info(
                tx,
                insert.partition_id,
                insert.table_id,
                insert.new_snapshot,
            )?;
            for planned in &insert.columns {
                ops.insert_partition_column(
                    tx,
                    insert.partition_id,
                    insert.table_id,
                    planned.partition_key_index,
                    planned.column,
                )?;
            }
        }
    }
    Ok(())
}

#[cfg(any(feature = "postgres", feature = "quack"))]
async fn apply_planned_partition_metadata_changes_async<O, Tx>(
    ops: &O,
    tx: &Tx,
    changes: &[PlannedPartitionMetadataChange<'_>],
) -> Result<()>
where
    O: AsyncPartitioningCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    for change in changes {
        if let Some(close) = change.close {
            let closed = ops
                .close_partition(tx, close.partition_id, close.new_snapshot)
                .await?;
            ensure_closed_metadata_row(&close.expectation, closed)?;
        }
        if let Some(insert) = &change.insert {
            ops.insert_partition_info(
                tx,
                insert.partition_id,
                insert.table_id,
                insert.new_snapshot,
            )
            .await?;
            for planned in &insert.columns {
                ops.insert_partition_column(
                    tx,
                    insert.partition_id,
                    insert.table_id,
                    planned.partition_key_index,
                    planned.column,
                )
                .await?;
            }
        }
    }
    Ok(())
}

#[derive(Debug, Clone, Copy)]
struct PlannedSortClose {
    sort_id: ducklake_catalog::SortId,
    table_id: TableId,
    new_snapshot: i64,
    expectation: MetadataCloseRequest,
}

#[derive(Debug, Clone)]
struct PlannedSortMetadataChange<'a> {
    close: Option<PlannedSortClose>,
    insert: Option<PlannedSortInfoInsert<'a>>,
}

#[derive(Debug, Clone)]
struct PlannedSortInfoInsert<'a> {
    sort_id: ducklake_catalog::SortId,
    table_id: TableId,
    new_snapshot: i64,
    expressions: Vec<PlannedSortExpressionInsert<'a>>,
}

#[derive(Debug, Clone, Copy)]
struct PlannedSortExpressionInsert<'a> {
    sort_key_index: i64,
    expression: &'a SortExpressionCommit,
}

fn planned_sort_metadata_change<'a>(
    plan: &crate::SortedByChangePlan,
    expressions: &'a [SortExpressionCommit],
) -> Result<PlannedSortMetadataChange<'a>> {
    let table_id = plan.metadata.table_id;
    let new_snapshot = plan.metadata.new_snapshot;
    Ok(PlannedSortMetadataChange {
        close: plan.close_sort_id.map(|sort_id| PlannedSortClose {
            sort_id,
            table_id,
            new_snapshot,
            expectation: MetadataCloseRequest {
                operation: "SET SORTED BY",
                row_kind: "sort",
                row_id: sort_id.0,
            },
        }),
        insert: plan
            .insert_sort_id
            .map(|sort_id| -> Result<PlannedSortInfoInsert<'a>> {
                Ok(PlannedSortInfoInsert {
                    sort_id,
                    table_id,
                    new_snapshot,
                    expressions: planned_sort_expression_inserts(expressions)?,
                })
            })
            .transpose()?,
    })
}

fn planned_sort_expression_inserts(
    expressions: &[SortExpressionCommit],
) -> Result<Vec<PlannedSortExpressionInsert<'_>>> {
    expressions
        .iter()
        .enumerate()
        .map(|(idx, expression)| {
            Ok(PlannedSortExpressionInsert {
                sort_key_index: sort_key_index(idx)?,
                expression,
            })
        })
        .collect()
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
fn apply_planned_sort_metadata_change_sync<'tx, O>(
    ops: &O,
    tx: &O::Tx<'tx>,
    writes: &PlannedSortMetadataChange<'_>,
) -> Result<()>
where
    O: SyncSortedByCommitOps,
{
    if let Some(close) = writes.close {
        let closed = ops.close_sort(tx, close.sort_id, close.table_id, close.new_snapshot)?;
        ensure_closed_metadata_row(&close.expectation, closed)?;
    }
    if let Some(insert) = &writes.insert {
        ops.insert_sort_info(tx, insert.sort_id, insert.table_id, insert.new_snapshot)?;
        for planned in &insert.expressions {
            ops.insert_sort_expression(
                tx,
                insert.sort_id,
                insert.table_id,
                planned.sort_key_index,
                planned.expression,
            )?;
        }
    }
    Ok(())
}

#[cfg(any(feature = "postgres", feature = "quack"))]
async fn apply_planned_sort_metadata_change_async<O, Tx>(
    ops: &O,
    tx: &Tx,
    writes: &PlannedSortMetadataChange<'_>,
) -> Result<()>
where
    O: AsyncSortedByCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    if let Some(close) = writes.close {
        let closed = ops
            .close_sort(tx, close.sort_id, close.table_id, close.new_snapshot)
            .await?;
        ensure_closed_metadata_row(&close.expectation, closed)?;
    }
    if let Some(insert) = &writes.insert {
        ops.insert_sort_info(tx, insert.sort_id, insert.table_id, insert.new_snapshot)
            .await?;
        for planned in &insert.expressions {
            ops.insert_sort_expression(
                tx,
                insert.sort_id,
                insert.table_id,
                planned.sort_key_index,
                planned.expression,
            )
            .await?;
        }
    }
    Ok(())
}

#[cfg(any(feature = "postgres", feature = "quack"))]
pub(crate) async fn commit_create_schema_async<O, Tx>(
    ops: &O,
    tx: &Tx,
    create: &CreateSchemaCommit,
) -> Result<SnapshotId>
where
    O: AsyncCatalogObjectCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    require_catalog_table_exists(
        "ducklake_schema",
        ops.table_exists(tx, "ducklake_schema").await?,
    )?;
    let ids = ops.latest_snapshot_ids(tx).await?;
    let snapshot = SnapshotId(ids.snapshot_id);
    ensure_schema_absent(
        &create.schema_name,
        ops.visible_schema(tx, &create.schema_name, snapshot)
            .await?,
    )?;
    let plan = plan_create_schema_catalog_mutation(create, catalog_ids(ids))?;
    let schema_id = require_allocated_catalog_id(&plan, "CREATE SCHEMA", "schema")?;
    let schema = planned_schema_row(schema_id, plan.new_snapshot, create);
    apply_create_schema_async(ops, tx, &schema).await?;
    finish_catalog_object_mutation_async(ops, tx, &plan).await
}

#[cfg(any(feature = "postgres", feature = "quack"))]
pub(crate) async fn commit_create_table_async<O, Tx>(
    ops: &O,
    tx: &Tx,
    create: &CreateTableCommit,
) -> Result<SnapshotId>
where
    O: AsyncCatalogObjectCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    let ids = ops.latest_snapshot_ids(tx).await?;
    let snapshot = SnapshotId(ids.snapshot_id);
    let schema = create_object_target_schema_async(
        ops,
        tx,
        &create.schema_name,
        &create.table_name,
        snapshot,
    )
    .await?;
    let plan = plan_create_table_catalog_mutation(create, catalog_ids(ids))?;
    let table_id = require_allocated_catalog_id(&plan, "CREATE TABLE", "table")?;
    let table = planned_table_insert(table_id, plan.new_snapshot, schema.schema_id, create);
    apply_create_table_async(ops, tx, &table).await?;
    finish_catalog_object_mutation_async(ops, tx, &plan).await
}

#[cfg(any(feature = "postgres", feature = "quack"))]
pub(crate) async fn commit_create_table_with_files_async<O, Tx>(
    ops: &O,
    tx: &Tx,
    create: &CreateTableWithFilesCommit,
) -> Result<SnapshotId>
where
    O: AsyncCreateTableWithFilesCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    let ids = ops.latest_snapshot_ids(tx).await?;
    let snapshot = SnapshotId(ids.snapshot_id);
    let schema = create_object_target_schema_async(
        ops,
        tx,
        &create.create.schema_name,
        &create.create.table_name,
        snapshot,
    )
    .await?;
    let metadata_tables = table_data_metadata_tables_async(
        ops,
        tx,
        "CREATE TABLE AS",
        append_files_have_partition_values(&create.files),
    )
    .await?;
    let plan =
        plan_create_table_with_files_commit(create, catalog_ids(ids), metadata_tables.table_stats)?;
    let table = planned_table_insert(
        create.table_id.0,
        plan.catalog.new_snapshot,
        schema.schema_id,
        &create.create,
    );
    apply_create_table_async(ops, tx, &table).await?;
    insert_planned_data_files_async(
        ops,
        tx,
        &create.files,
        &plan.files,
        metadata_tables.data_file,
        "CREATE TABLE AS",
    )
    .await?;
    let column_stats = planned_table_column_stats_append_update(
        create.table_id,
        SnapshotId(plan.catalog.new_snapshot),
        &create.files,
    );
    apply_table_column_stats_append_async(ops, tx, column_stats).await?;
    let snapshot_id = finish_snapshot_commit_async(
        ops,
        tx,
        catalog_object_snapshot_commit_with_next_file_id(&plan.catalog, plan.new_next_file_id),
    )
    .await?;
    let stats = planned_create_table_stats_insert(create.table_id, &plan);
    apply_create_table_stats_async(ops, tx, metadata_tables.table_stats, stats).await?;
    Ok(snapshot_id)
}

#[cfg(any(feature = "postgres", feature = "quack"))]
pub(crate) async fn commit_create_table_with_inlined_data_async<O, Tx>(
    ops: &O,
    tx: &Tx,
    create: &CreateTableWithInlinedDataCommit,
) -> Result<SnapshotId>
where
    O: AsyncCreateTableWithFilesCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    let ids = ops.latest_snapshot_ids(tx).await?;
    let snapshot = SnapshotId(ids.snapshot_id);
    let schema_entry = create_object_target_schema_async(
        ops,
        tx,
        &create.create.schema_name,
        &create.create.table_name,
        snapshot,
    )
    .await?;
    let plan = plan_create_table_with_inlined_data_commit(
        create,
        catalog_ids(ids),
        ops.table_exists(tx, "ducklake_inlined_data_tables").await?,
    )?;
    let table_schema = create_table_schema_from_commit(
        create.table_id,
        plan.catalog.new_snapshot,
        &create.create.columns,
    )?;
    ops.validate_inlined_append(
        create.rows_inserted,
        &table_schema,
        &create.batches,
        None,
        "DuckLake CREATE TABLE AS inlined commit",
    )?;
    let table = planned_table_insert(
        create.table_id.0,
        plan.catalog.new_snapshot,
        schema_entry.schema_id,
        &create.create,
    );
    apply_create_table_async(ops, tx, &table).await?;
    let batch_insert = planned_create_table_inlined_batch_insert(
        create,
        &table_schema,
        plan.catalog.new_schema_version,
        plan.catalog.new_snapshot,
    );
    insert_planned_inlined_batches_async(ops, tx, &[batch_insert]).await?;
    let column_stats_files = inlined_append_column_stats_files(
        create.table_id,
        SnapshotId(plan.catalog.new_snapshot),
        [(plan.total_record_count, create.column_stats.clone())],
    );
    let column_stats = planned_table_column_stats_append_update(
        create.table_id,
        SnapshotId(plan.catalog.new_snapshot),
        &column_stats_files,
    );
    apply_table_column_stats_append_async(ops, tx, column_stats).await?;
    let snapshot_id =
        finish_snapshot_commit_async(ops, tx, catalog_object_snapshot_commit(&plan.catalog))
            .await?;
    let stats =
        planned_inlined_append_table_stats_update(create.table_id, 0, plan.total_record_count)?;
    apply_inlined_append_table_stats_async(
        ops,
        tx,
        ops.table_exists(tx, "ducklake_table_stats").await?,
        stats,
    )
    .await?;
    Ok(snapshot_id)
}

#[cfg(any(feature = "postgres", feature = "quack"))]
pub(crate) async fn commit_create_view_async<O, Tx>(
    ops: &O,
    tx: &Tx,
    create: &CreateViewCommit,
) -> Result<SnapshotId>
where
    O: AsyncCatalogObjectCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    let ids = ops.latest_snapshot_ids(tx).await?;
    let snapshot = SnapshotId(ids.snapshot_id);
    let schema = create_object_target_schema_async(
        ops,
        tx,
        &create.schema_name,
        &create.view_name,
        snapshot,
    )
    .await?;
    let plan = plan_create_view_catalog_mutation(create, catalog_ids(ids))?;
    let view_id = require_allocated_catalog_id(&plan, "CREATE VIEW", "view")?;
    let view = planned_view_row(view_id, plan.new_snapshot, schema.schema_id, create);
    apply_create_view_async(ops, tx, &view).await?;
    finish_catalog_object_mutation_async(ops, tx, &plan).await
}

#[cfg(any(feature = "postgres", feature = "quack"))]
pub(crate) async fn commit_drop_table_async<O, Tx>(
    ops: &O,
    tx: &Tx,
    drop: &DropTableCommit,
) -> Result<SnapshotId>
where
    O: AsyncCatalogObjectCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    let ids = ops.latest_snapshot_ids(tx).await?;
    let snapshot = SnapshotId(ids.snapshot_id);
    let table = drop_table_target_async(ops, tx, drop, snapshot).await?;
    let plan = plan_drop_table_catalog_mutation(drop, catalog_ids(ids), table.table_id)?;
    apply_drop_table_async(ops, tx, table.table_id, plan.new_snapshot).await?;
    finish_catalog_object_mutation_async(ops, tx, &plan).await
}

#[cfg(any(feature = "postgres", feature = "quack"))]
pub(crate) async fn commit_rename_table_async<O, Tx>(
    ops: &O,
    tx: &Tx,
    rename: &RenameTableCommit,
) -> Result<SnapshotId>
where
    O: AsyncCatalogObjectCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    let ids = ops.latest_snapshot_ids(tx).await?;
    let snapshot = SnapshotId(ids.snapshot_id);
    let facts = rename_table_facts_async(ops, tx, rename, snapshot).await?;
    let plan =
        plan_rename_table_catalog_mutation(rename, catalog_ids(ids), &facts.target_schema_name)?;
    let renamed_table = planned_renamed_table_row(
        &facts.data_path,
        &facts.source_schema,
        &facts.target_schema,
        &facts.table,
        rename,
        plan.new_snapshot,
    );
    apply_rename_table_async(
        ops,
        tx,
        facts.table.table_id,
        plan.new_snapshot,
        &renamed_table,
    )
    .await?;
    finish_catalog_object_mutation_async(ops, tx, &plan).await
}

#[cfg(any(feature = "postgres", feature = "quack"))]
pub(crate) async fn commit_drop_view_async<O, Tx>(
    ops: &O,
    tx: &Tx,
    drop: &DropViewCommit,
) -> Result<SnapshotId>
where
    O: AsyncCatalogObjectCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    let ids = ops.latest_snapshot_ids(tx).await?;
    let snapshot = SnapshotId(ids.snapshot_id);
    let view = drop_view_target_async(ops, tx, drop, snapshot).await?;
    let plan = plan_drop_view_catalog_mutation(drop, catalog_ids(ids), view.view_id)?;
    apply_drop_view_async(ops, tx, view.view_id, plan.new_snapshot).await?;
    finish_catalog_object_mutation_async(ops, tx, &plan).await
}

#[cfg(any(feature = "postgres", feature = "quack"))]
pub(crate) async fn commit_drop_schema_async<O, Tx>(
    ops: &O,
    tx: &Tx,
    drop: &DropSchemaCommit,
) -> Result<SnapshotId>
where
    O: AsyncCatalogObjectCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    let ids = ops.latest_snapshot_ids(tx).await?;
    let snapshot = SnapshotId(ids.snapshot_id);
    let schema = require_schema(
        ops.visible_schema(tx, &drop.schema_name, snapshot).await?,
        &drop.schema_name,
    )?;
    let table_ids = ops
        .visible_table_ids_for_schema(tx, schema.schema_id, snapshot)
        .await?;
    let view_ids = ops
        .visible_view_ids_for_schema(tx, schema.schema_id, snapshot)
        .await?;
    let close_views = ops.table_exists(tx, "ducklake_view").await?;
    let planned = planned_drop_schema(
        drop,
        ids,
        schema.schema_id,
        snapshot,
        table_ids,
        &view_ids,
        close_views,
    )?;
    apply_planned_drop_schema_async(ops, tx, &planned).await?;
    finish_catalog_object_mutation_async(ops, tx, &planned.plan).await
}

#[cfg(any(feature = "postgres", feature = "quack"))]
async fn finish_catalog_object_mutation_async<O, Tx>(
    ops: &O,
    tx: &Tx,
    plan: &crate::CatalogObjectMutationPlan,
) -> Result<SnapshotId>
where
    O: AsyncCatalogObjectCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    finish_snapshot_commit_async(ops, tx, catalog_object_snapshot_commit(plan)).await
}

#[cfg(any(feature = "postgres", feature = "quack"))]
async fn ensure_table_or_view_absent_async<O, Tx>(
    ops: &O,
    tx: &Tx,
    schema_id: SchemaId,
    object_name: &str,
    snapshot: SnapshotId,
    schema_name: &str,
) -> Result<()>
where
    O: AsyncCatalogObjectCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    let tables = ops
        .visible_tables_case_insensitive(tx, schema_id, object_name, snapshot)
        .await?;
    let views = if ops.table_exists(tx, "ducklake_view").await? {
        ops.visible_views_case_insensitive(tx, schema_id, object_name, snapshot)
            .await?
    } else {
        Vec::new()
    };
    ensure_table_or_view_absent_from_rows(tables, views, schema_name, object_name)
}

#[cfg(any(feature = "postgres", feature = "quack"))]
async fn ensure_rename_target_available_async<O, Tx>(
    ops: &O,
    tx: &Tx,
    schema_id: SchemaId,
    object_name: &str,
    snapshot: SnapshotId,
    schema_name: &str,
    source_table_id: TableId,
) -> Result<()>
where
    O: AsyncCatalogObjectCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    let tables = ops
        .visible_tables_case_insensitive(tx, schema_id, object_name, snapshot)
        .await?;
    let views = ops
        .visible_views_case_insensitive(tx, schema_id, object_name, snapshot)
        .await?;
    ensure_rename_target_available_from_rows(
        tables,
        views,
        schema_name,
        object_name,
        source_table_id,
    )
}

#[cfg(any(feature = "postgres", feature = "quack"))]
pub(crate) async fn commit_column_nullability_async<O, Tx>(
    ops: &O,
    tx: &Tx,
    change: &ColumnNullabilityCommit,
) -> Result<SnapshotId>
where
    O: AsyncColumnEvolutionCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    let ids = ops.latest_snapshot_ids(tx).await?;
    let snapshot = SnapshotId(ids.snapshot_id);
    let replacement = column_nullability_replacement_async(ops, tx, change, ids, snapshot).await?;
    commit_column_replacement_async(ops, tx, &replacement, "ALTER COLUMN").await
}

#[cfg(any(feature = "postgres", feature = "quack"))]
pub(crate) async fn commit_column_default_async<O, Tx>(
    ops: &O,
    tx: &Tx,
    change: &ColumnDefaultCommit,
) -> Result<SnapshotId>
where
    O: AsyncColumnEvolutionCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    let ids = ops.latest_snapshot_ids(tx).await?;
    let snapshot = SnapshotId(ids.snapshot_id);
    let plan = column_default_replacement_async(ops, tx, change, ids, snapshot).await?;
    commit_column_replacement_async(ops, tx, &plan, "ALTER COLUMN DEFAULT").await
}

#[cfg(any(feature = "postgres", feature = "quack"))]
pub(crate) async fn commit_column_type_async<O, Tx>(
    ops: &O,
    tx: &Tx,
    change: &ColumnTypeCommit,
) -> Result<SnapshotId>
where
    O: AsyncColumnEvolutionCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    let ids = ops.latest_snapshot_ids(tx).await?;
    let snapshot = SnapshotId(ids.snapshot_id);
    match column_type_change_async(ops, tx, change, ids, snapshot).await? {
        PlannedColumnTypeChange::Noop { snapshot } => Ok(snapshot),
        PlannedColumnTypeChange::Commit { replacement } => {
            commit_column_replacement_async(ops, tx, &replacement, "ALTER COLUMN TYPE").await
        }
    }
}

#[cfg(any(feature = "postgres", feature = "quack"))]
pub(crate) async fn commit_column_type_tree_async<O, Tx>(
    ops: &O,
    tx: &Tx,
    change: &ColumnTypeTreeCommit,
) -> Result<SnapshotId>
where
    O: AsyncColumnEvolutionCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    let ids = ops.latest_snapshot_ids(tx).await?;
    let snapshot = SnapshotId(ids.snapshot_id);
    match column_type_tree_change_async(ops, tx, change, ids, snapshot).await? {
        PlannedColumnTypeTreeChange::Noop { snapshot } => Ok(snapshot),
        PlannedColumnTypeTreeChange::Commit {
            metadata,
            close_columns,
            insert_columns,
        } => {
            apply_column_type_tree_change_async(
                ops,
                tx,
                &metadata,
                &close_columns,
                &insert_columns,
            )
            .await?;
            finish_column_metadata_mutation_async(ops, tx, &metadata).await
        }
    }
}

#[cfg(any(feature = "postgres", feature = "quack"))]
pub(crate) async fn commit_rename_column_async<O, Tx>(
    ops: &O,
    tx: &Tx,
    rename: &RenameColumnCommit,
) -> Result<SnapshotId>
where
    O: AsyncColumnEvolutionCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    let ids = ops.latest_snapshot_ids(tx).await?;
    let snapshot = SnapshotId(ids.snapshot_id);
    let plan = rename_column_plan_async(ops, tx, rename, ids, snapshot).await?;
    commit_column_replacement_async(ops, tx, &plan, "RENAME COLUMN").await
}

#[cfg(any(feature = "postgres", feature = "quack"))]
pub(crate) async fn commit_drop_column_async<O, Tx>(
    ops: &O,
    tx: &Tx,
    drop: &DropColumnCommit,
) -> Result<SnapshotId>
where
    O: AsyncColumnEvolutionCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    let ids = ops.latest_snapshot_ids(tx).await?;
    let snapshot = SnapshotId(ids.snapshot_id);
    let facts = drop_column_facts_async(ops, tx, drop, snapshot).await?;
    let plan = plan_drop_column_mutation(drop.clone(), catalog_ids(ids), facts)?;
    let close = plan.close_request(snapshot);
    apply_drop_column_subtree_close_async(ops, tx, &plan.metadata, &close).await
}

#[cfg(any(feature = "postgres", feature = "quack"))]
pub(crate) async fn commit_add_columns_async<O, Tx>(
    ops: &O,
    tx: &Tx,
    add: &AddColumnCommit,
) -> Result<SnapshotId>
where
    O: AsyncColumnEvolutionCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    let ids = ops.latest_snapshot_ids(tx).await?;
    let snapshot = SnapshotId(ids.snapshot_id);
    let plan = add_columns_plan_async(ops, tx, add, ids, snapshot).await?;
    apply_add_columns_async(ops, tx, add, plan.new_snapshot).await?;
    finish_snapshot_commit_async(ops, tx, metadata_snapshot_commit(&plan)).await
}

#[cfg(any(feature = "postgres", feature = "quack"))]
pub(crate) async fn commit_partitioning_batch_async<O, Tx>(
    ops: &O,
    tx: &Tx,
    partitioning: &PartitioningBatchCommit,
) -> Result<SnapshotId>
where
    O: AsyncPartitioningCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    require_partition_metadata_tables_async(ops, tx).await?;

    let ids = ops.latest_snapshot_ids(tx).await?;
    let snapshot = SnapshotId(ids.snapshot_id);
    let facts = partitioning_table_facts_async(ops, tx, partitioning, snapshot).await?;
    let plan = plan_partitioning_batch_commit(partitioning, catalog_ids(ids), facts)?;
    let changes = planned_partition_metadata_changes(&plan)?;
    apply_planned_partition_metadata_changes_async(ops, tx, &changes).await?;

    finish_snapshot_commit_async(ops, tx, partitioning_snapshot_commit(&plan)).await
}

#[cfg(any(feature = "postgres", feature = "quack"))]
pub(crate) async fn commit_sorted_by_async<O, Tx>(
    ops: &O,
    tx: &Tx,
    sorted_by: &SortedByCommit,
) -> Result<SnapshotId>
where
    O: AsyncSortedByCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    require_sort_metadata_tables_async(ops, tx).await?;

    let ids = ops.latest_snapshot_ids(tx).await?;
    let snapshot = SnapshotId(ids.snapshot_id);
    let active = sorted_by_active_specs_async(ops, tx, sorted_by, snapshot).await?;
    let plan = plan_sorted_by_commit(
        sorted_by.table_id,
        sorted_by.expected_snapshot,
        catalog_ids(ids),
        &active,
        &sorted_by.expressions,
        sorted_by.commit_message.clone(),
    )?;
    let SortedByCommitPlan::Change(plan) = plan else {
        return Ok(snapshot);
    };
    let metadata = &plan.metadata;
    let writes = planned_sort_metadata_change(&plan, &sorted_by.expressions)?;
    apply_planned_sort_metadata_change_async(ops, tx, &writes).await?;
    finish_snapshot_commit_async(ops, tx, metadata_snapshot_commit(metadata)).await
}

#[cfg(any(feature = "postgres", feature = "quack"))]
async fn commit_column_replacement_async<O, Tx>(
    ops: &O,
    tx: &Tx,
    replacement: &PlannedColumnReplacement,
    operation: &'static str,
) -> Result<SnapshotId>
where
    O: AsyncColumnEvolutionCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    let metadata = &replacement.metadata;
    let column = &replacement.column;
    let close = replacement.close_request(operation);
    let closed = ops
        .close_column(tx, close.table_id, close.column_id, close.new_snapshot)
        .await?;
    ensure_closed_one(&close, closed)?;
    ops.insert_column(tx, column, metadata.new_snapshot).await?;
    finish_column_metadata_mutation_async(ops, tx, metadata).await
}

#[cfg(any(feature = "postgres", feature = "quack"))]
async fn finish_column_metadata_mutation_async<O, Tx>(
    ops: &O,
    tx: &Tx,
    metadata: &MetadataMutationPlan,
) -> Result<SnapshotId>
where
    O: AsyncColumnEvolutionCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    finish_snapshot_commit_async(ops, tx, metadata_snapshot_commit(metadata)).await
}

#[cfg(any(feature = "postgres", feature = "quack"))]
async fn apply_column_type_tree_change_async<O, Tx>(
    ops: &O,
    tx: &Tx,
    metadata: &MetadataMutationPlan,
    close_columns: &[ColumnCloseRequest],
    insert_columns: &[PlannedColumnTypeTreeRow],
) -> Result<()>
where
    O: AsyncColumnEvolutionCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    for close in close_columns {
        let closed = ops
            .close_column(tx, close.table_id, close.column_id, close.new_snapshot)
            .await?;
        ensure_closed_one(close, closed)?;
    }
    for column in insert_columns {
        ops.insert_type_tree_column(tx, metadata.table_id, column, metadata.new_snapshot)
            .await?;
    }
    Ok(())
}

#[cfg(any(feature = "postgres", feature = "quack"))]
async fn apply_add_columns_async<O, Tx>(
    ops: &O,
    tx: &Tx,
    add: &AddColumnCommit,
    new_snapshot: i64,
) -> Result<()>
where
    O: AsyncColumnEvolutionCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    for column in &add.columns {
        ops.insert_add_column(tx, add.table_id, column, new_snapshot)
            .await?;
    }
    for stats in &add.table_column_stats {
        ops.upsert_table_column_stats(tx, stats).await?;
    }
    Ok(())
}

#[cfg(any(feature = "postgres", feature = "quack"))]
pub(crate) async fn commit_append_files_async<O, Tx>(
    ops: &O,
    tx: &Tx,
    files: &[AppendFile],
) -> Result<SnapshotId>
where
    O: AsyncTableDataCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    let append_plan = plan_append_files_commit(files)?;
    let table_id = append_plan.table_id;
    let ids = ops.latest_snapshot_ids(tx).await?;

    let metadata_tables = table_data_metadata_tables_async(
        ops,
        tx,
        "append",
        append_files_have_partition_values(files),
    )
    .await?;
    let next_row_id = ops
        .next_row_id_for_append(tx, table_id, metadata_tables.table_stats)
        .await?;
    let snapshot_plan =
        append_plan.plan_snapshot(ids.snapshot_id, ids.next_file_id, next_row_id, files)?;

    insert_planned_data_files_async(
        ops,
        tx,
        files,
        &snapshot_plan.files,
        metadata_tables.data_file,
        "append",
    )
    .await?;
    let column_stats =
        planned_table_column_stats_append_update(table_id, SnapshotId(ids.snapshot_id), files);
    apply_table_column_stats_append_async(ops, tx, column_stats).await?;
    let snapshot_id = finish_snapshot_commit_async(
        ops,
        tx,
        append_files_snapshot_commit(ids, table_id, &snapshot_plan),
    )
    .await?;
    let stats = planned_append_table_stats_update(table_id, &snapshot_plan);
    apply_append_table_stats_async(ops, tx, metadata_tables.table_stats, stats).await?;

    Ok(snapshot_id)
}

#[cfg(any(feature = "postgres", feature = "quack"))]
pub(crate) async fn commit_append_inlined_data_async<O, Tx>(
    ops: &O,
    tx: &Tx,
    append: &InlinedDataAppendCommit,
) -> Result<SnapshotId>
where
    O: AsyncTableDataCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    ops.validate_inlined_append(
        append.rows_appended,
        &append.schema,
        &append.batches,
        None,
        "DuckLake inlined append",
    )?;
    let ids = ops.latest_snapshot_ids(tx).await?;
    let has_table_stats = ops.table_exists(tx, "ducklake_table_stats").await?;
    let plan = planned_inlined_append_commit(
        ids,
        append,
        ops.table_exists(tx, "ducklake_inlined_data_tables").await?,
        has_table_stats,
        ops.next_row_id_for_append(tx, append.table_id, has_table_stats)
            .await?,
    )?;
    let batch_insert = planned_inlined_append_batch_insert(append, &plan);
    insert_planned_inlined_batches_async(ops, tx, &[batch_insert]).await?;
    let column_stats_files = inlined_append_column_stats_files(
        append.table_id,
        SnapshotId(ids.snapshot_id),
        [(append.rows_appended, append.column_stats.clone())],
    );
    let column_stats = planned_table_column_stats_append_update(
        append.table_id,
        SnapshotId(ids.snapshot_id),
        &column_stats_files,
    );
    apply_table_column_stats_append_async(ops, tx, column_stats).await?;

    let snapshot_id = finish_snapshot_commit_async(
        ops,
        tx,
        inlined_append_snapshot_commit(ids, append, plan.new_snapshot, plan.next_file_id),
    )
    .await?;
    let stats = planned_inlined_append_table_stats_update(
        append.table_id,
        plan.row_id_start,
        append.rows_appended,
    )?;
    apply_inlined_append_table_stats_async(ops, tx, plan.has_table_stats, stats).await?;

    Ok(snapshot_id)
}

#[cfg(any(feature = "postgres", feature = "quack"))]
pub(crate) async fn commit_delete_files_async<O, Tx>(
    ops: &O,
    tx: &Tx,
    files: &[DeleteFileCommit],
) -> Result<SnapshotId>
where
    O: AsyncTableDataCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    let delete_plan = plan_delete_files_commit(files)?;
    let table_id = delete_plan.table_id;
    let ids = ops.latest_snapshot_ids(tx).await?;
    let snapshot_plan = delete_plan.plan_snapshot(ids.snapshot_id, ids.next_file_id, files)?;
    let new_snapshot = snapshot_plan.new_snapshot.0;

    insert_planned_delete_files_async(ops, tx, files, new_snapshot).await?;
    let snapshot_id = finish_snapshot_commit_async(
        ops,
        tx,
        delete_files_snapshot_commit(ids, table_id, &snapshot_plan),
    )
    .await?;
    let stats = planned_delete_table_stats_update(table_id, &snapshot_plan);
    apply_delete_table_stats_async(ops, tx, stats).await?;

    Ok(snapshot_id)
}

#[cfg(any(feature = "postgres", feature = "quack"))]
pub(crate) async fn commit_table_changes_async<O, Tx>(
    ops: &O,
    tx: &Tx,
    changes: &TableChangeCommit,
) -> Result<SnapshotId>
where
    O: AsyncTableDataCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    let table_change_plan = plan_table_change_commit(changes)?;
    for append in &changes.inlined_data_appends {
        ops.validate_inlined_append(
            append.rows_appended,
            &append.schema,
            &append.batches,
            append.row_ids.as_deref(),
            "DuckLake table-change inlined append",
        )?;
    }

    let ids = ops.latest_snapshot_ids(tx).await?;
    let metadata_tables = table_data_metadata_tables_async(
        ops,
        tx,
        "table-change",
        table_change_has_partition_values(changes),
    )
    .await?;
    let next_row_id = ops
        .next_row_id_for_append(tx, changes.table_id, metadata_tables.table_stats)
        .await?;
    let snapshot_plan =
        table_change_plan.plan_snapshot(ids.snapshot_id, ids.next_file_id, next_row_id, changes)?;
    let new_snapshot = snapshot_plan.new_snapshot.0;

    insert_planned_data_files_async(
        ops,
        tx,
        &changes.append_files,
        &snapshot_plan.append_files,
        metadata_tables.data_file,
        "table-change",
    )
    .await?;
    let column_stats = planned_table_column_stats_append_update(
        changes.table_id,
        SnapshotId(ids.snapshot_id),
        &changes.append_files,
    );
    apply_table_column_stats_append_async(ops, tx, column_stats).await?;

    let batch_inserts = planned_table_change_inlined_batch_inserts(
        changes,
        &snapshot_plan.inlined_data_appends,
        new_snapshot,
    )?;
    insert_planned_inlined_batches_async(ops, tx, &batch_inserts).await?;
    let inlined_column_stats_files = inlined_append_column_stats_files(
        changes.table_id,
        SnapshotId(ids.snapshot_id),
        changes
            .inlined_data_appends
            .iter()
            .map(|append| (append.rows_appended, append.column_stats.clone())),
    );
    let inlined_column_stats = planned_table_column_stats_append_update(
        changes.table_id,
        SnapshotId(ids.snapshot_id),
        &inlined_column_stats_files,
    );
    apply_table_column_stats_append_async(ops, tx, inlined_column_stats).await?;

    insert_planned_delete_files_async(ops, tx, &changes.delete_files, new_snapshot).await?;
    let inlined_mutations = planned_table_change_inlined_mutations(changes, ids, &snapshot_plan);
    ops.write_inlined_file_deletions(
        tx,
        inlined_mutations.table_id,
        inlined_mutations.inlined_file_deletions,
        inlined_mutations.new_snapshot,
    )
    .await?;
    let inlined_data_deleted = ops
        .apply_inlined_data_deletions(
            tx,
            inlined_mutations.inlined_data_deletions,
            inlined_mutations.new_snapshot,
        )
        .await?;
    let flushed_inlined_data_deleted = ops
        .delete_flushed_inlined_data(
            tx,
            inlined_mutations.flushed_inlined_data,
            inlined_mutations.previous_snapshot,
        )
        .await?;
    ops.delete_flushed_inlined_file_deletions(
        tx,
        inlined_mutations.table_id,
        inlined_mutations.flushed_inlined_file_deletions,
    )
    .await?;
    let total_delete_count = planned_table_change_total_delete_count(
        &inlined_mutations,
        inlined_data_deleted,
        flushed_inlined_data_deleted,
    )?;

    let snapshot_id = finish_snapshot_commit_async(
        ops,
        tx,
        table_change_snapshot_commit(ids, changes, &snapshot_plan),
    )
    .await?;
    let stats =
        planned_table_change_stats_update(changes.table_id, total_delete_count, &snapshot_plan);
    apply_table_change_table_stats_async(ops, tx, metadata_tables.table_stats, stats).await?;

    Ok(snapshot_id)
}

#[cfg(any(feature = "postgres", feature = "quack"))]
pub(crate) async fn commit_truncate_table_async<O, Tx>(
    ops: &O,
    tx: &Tx,
    truncate: &TruncateTableCommit,
) -> Result<SnapshotId>
where
    O: AsyncMaintenanceCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    let truncate_plan = plan_truncate_table_commit(truncate)?;
    let ids = ops.latest_snapshot_ids(tx).await?;
    let snapshot_plan = truncate_plan.plan_snapshot(ids.snapshot_id, ids.next_file_id)?;
    let effects = planned_truncate_table_effects(truncate, &snapshot_plan);

    apply_truncate_table_effects_async(ops, tx, &truncate_plan, &effects).await?;

    let snapshot_id = finish_snapshot_commit_async(
        ops,
        tx,
        truncate_snapshot_commit(ids, truncate, &snapshot_plan),
    )
    .await?;
    let stats_reset = planned_truncate_table_stats_reset(&effects);
    apply_truncate_table_stats_async(ops, tx, stats_reset).await?;

    Ok(snapshot_id)
}

#[cfg(any(feature = "postgres", feature = "quack"))]
pub(crate) async fn commit_merge_adjacent_compaction_async<O, Tx>(
    ops: &O,
    tx: &Tx,
    compaction: &MergeAdjacentCompactionCommit,
) -> Result<SnapshotId>
where
    O: AsyncMaintenanceCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    let source_ids = plan_merge_adjacent_compaction_sources(compaction)?;
    let ids = ops.latest_snapshot_ids(tx).await?;
    let source_files = ops
        .source_data_files(tx, compaction.table_id, ids.snapshot_id, &source_ids)
        .await?;
    require_operation_table(
        "merge-adjacent compaction",
        "ducklake_files_scheduled_for_deletion",
        ops.table_exists(tx, "ducklake_files_scheduled_for_deletion")
            .await?,
    )?;
    require_file_partition_value_table(
        "merge-adjacent compaction",
        compaction.has_partition_values(),
        ops.table_exists(tx, "ducklake_file_partition_value")
            .await?,
    )?;

    let plan = plan_merge_adjacent_compaction_commit(
        compaction,
        catalog_ids(ids),
        &source_ids,
        source_files.len(),
        source_files.iter().map(|file| file.file_size_bytes),
    )?;
    let cleanup = planned_maintenance_cleanup(compaction.table_id, &source_ids, &source_files);
    apply_maintenance_cleanup_async(ops, tx, ids.snapshot_id, &cleanup).await?;

    finish_maintenance_commit_async(
        ops,
        tx,
        compaction.table_id,
        &compaction.compacted_files,
        plan,
    )
    .await
}

#[cfg(any(feature = "postgres", feature = "quack"))]
pub(crate) async fn commit_rewrite_data_files_async<O, Tx>(
    ops: &O,
    tx: &Tx,
    rewrite: &RewriteDataFilesCommit,
) -> Result<SnapshotId>
where
    O: AsyncMaintenanceCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    let source_ids = plan_rewrite_data_files_sources(rewrite)?;
    let ids = ops.latest_snapshot_ids(tx).await?;
    let source_files = ops
        .source_data_files(tx, rewrite.table_id, ids.snapshot_id, &source_ids)
        .await?;
    require_file_partition_value_table(
        "rewrite-data-files",
        rewrite.has_partition_values(),
        ops.table_exists(tx, "ducklake_file_partition_value")
            .await?,
    )?;

    let plan = plan_rewrite_data_files_commit(
        rewrite,
        catalog_ids(ids),
        &source_ids,
        source_files.len(),
        source_files.iter().map(|file| file.file_size_bytes),
    )?;
    let close = planned_rewrite_close(
        rewrite.table_id,
        &source_ids,
        plan.new_snapshot,
        ops.table_exists(tx, "ducklake_delete_file").await?,
    );
    apply_rewrite_close_async(ops, tx, &close).await?;

    finish_maintenance_commit_async(ops, tx, rewrite.table_id, &rewrite.rewritten_files, plan).await
}

#[cfg(any(feature = "postgres", feature = "quack"))]
async fn finish_maintenance_commit_async<O, Tx>(
    ops: &O,
    tx: &Tx,
    table_id: TableId,
    output_files: &[AppendFile],
    plan: MaintenanceCommitPlan,
) -> Result<SnapshotId>
where
    O: AsyncMaintenanceCommitOps<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    let finish = planned_maintenance_finish(table_id, output_files, &plan);
    apply_maintenance_output_files_async(ops, tx, &finish).await?;
    let snapshot_id =
        finish_snapshot_commit_async(ops, tx, maintenance_snapshot_commit(&plan)).await?;
    apply_maintenance_table_stats_async(ops, tx, finish.table_stats).await?;

    Ok(snapshot_id)
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
fn validate_table_change_inlined_append<O>(ops: &O, append: &InlinedDataAppend) -> Result<()>
where
    O: SyncTableDataCommitOps,
{
    ops.validate_inlined_append(
        append.rows_appended,
        &append.schema,
        &append.batches,
        append.row_ids.as_deref(),
        "DuckLake table-change inlined append",
    )
}

fn delete_format_label(format: &DeleteFormat) -> Result<&'static str> {
    match format {
        DeleteFormat::Parquet => Ok("parquet"),
        DeleteFormat::Puffin => Ok("puffin"),
        DeleteFormat::Unknown(format) => Err(StorageError::Unsupported(format!(
            "DuckLake delete file format {format} is not supported"
        ))),
    }
}

fn partition_key_index(idx: usize) -> Result<i64> {
    i64::try_from(idx)
        .map_err(|_| StorageError::Backend("DuckLake partition key index is out of range".into()))
}
