use ducklake_catalog::SnapshotId;

#[cfg(any(feature = "postgres", feature = "quack"))]
use crate::commit::{
    AsyncCatalogObjectCommitOps, AsyncColumnEvolutionCommitOps, AsyncCommitOps,
    AsyncCreateTableWithFilesCommitOps, AsyncMaintenanceCommitOps, AsyncPartitioningCommitOps,
    AsyncSortedByCommitOps,
};
#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
use crate::commit::{
    SyncCatalogObjectCommitOps, SyncColumnEvolutionCommitOps, SyncCreateTableWithFilesCommitOps,
    SyncMaintenanceCommitOps, SyncPartitioningCommitOps, SyncSortedByCommitOps,
};
use crate::{
    AddColumnCommit, AppendFile, ColumnDefaultCommit, ColumnNullabilityCommit, ColumnTypeCommit,
    ColumnTypeTreeCommit, CreateSchemaCommit, CreateTableCommit, CreateTableWithFilesCommit,
    CreateTableWithInlinedDataCommit, CreateViewCommit, DeleteFileCommit, DropColumnCommit,
    DropSchemaCommit, DropTableCommit, DropViewCommit, InlinedDataAppendCommit,
    MergeAdjacentCompactionCommit, PartitioningBatchCommit, RenameColumnCommit, RenameTableCommit,
    Result, RewriteDataFilesCommit, SortedByCommit, StorageError, TableChangeCommit, TableSchema,
    TruncateTableCommit,
};

pub(crate) enum CommitRequest {
    AppendFiles(Vec<AppendFile>),
    AppendInlinedData(InlinedDataAppendCommit),
    DeleteFiles(Vec<DeleteFileCommit>),
    TableChanges(TableChangeCommit),
    CreateSchema(CreateSchemaCommit),
    CreateTable(CreateTableCommit),
    CreateTableWithFiles(CreateTableWithFilesCommit),
    CreateTableWithInlinedData(CreateTableWithInlinedDataCommit),
    CreateView(CreateViewCommit),
    DropTable(DropTableCommit),
    RenameTable(RenameTableCommit),
    DropView(DropViewCommit),
    DropSchema(DropSchemaCommit),
    ColumnNullability(ColumnNullabilityCommit),
    ColumnDefault(ColumnDefaultCommit),
    ColumnType(ColumnTypeCommit),
    ColumnTypeTree(ColumnTypeTreeCommit),
    RenameColumn(RenameColumnCommit),
    DropColumn(DropColumnCommit),
    AddColumns(AddColumnCommit),
    PartitioningBatch(PartitioningBatchCommit),
    SortedBy(SortedByCommit),
    TruncateTable(TruncateTableCommit),
    MergeAdjacentCompaction(MergeAdjacentCompactionCommit),
    RewriteDataFiles(RewriteDataFilesCommit),
}

#[cfg(feature = "quack")]
pub(crate) struct CommitAbortContext {
    pub(crate) label: &'static str,
    pub(crate) operation: Option<&'static str>,
    pub(crate) expected_snapshot: Option<SnapshotId>,
}

impl CommitRequest {
    pub(crate) fn validate_preflight(&self) -> Result<()> {
        match self {
            Self::AppendFiles(files) => {
                crate::plan_append_files_commit(files)?;
            }
            Self::AppendInlinedData(append) => {
                validate_inlined_append(
                    append.rows_appended,
                    &append.schema,
                    &append.batches,
                    None,
                    "DuckLake inlined append",
                )?;
            }
            Self::DeleteFiles(files) => {
                crate::plan_delete_files_commit(files)?;
            }
            Self::TableChanges(changes) => {
                crate::plan_table_change_commit(changes)?;
                for append in &changes.inlined_data_appends {
                    validate_inlined_append(
                        append.rows_appended,
                        &append.schema,
                        &append.batches,
                        append.row_ids.as_deref(),
                        "DuckLake table-change inlined append",
                    )?;
                }
            }
            Self::CreateSchema(create) => {
                if create.schema_name.trim().is_empty() {
                    return Err(StorageError::Backend(
                        "DuckLake create-schema schema name is empty".into(),
                    ));
                }
            }
            Self::CreateTable(create) => {
                validate_create_table_input(create)?;
            }
            Self::CreateTableWithFiles(create) => {
                if create.files.is_empty() {
                    return Err(StorageError::Backend(
                        "DuckLake CREATE TABLE AS commit received no files".into(),
                    ));
                }
                validate_create_table_input(&create.create)?;
                for file in &create.files {
                    if file.table_id != create.table_id {
                        return Err(StorageError::Backend(
                            "DuckLake CREATE TABLE AS files span multiple tables".into(),
                        ));
                    }
                    if file.expected_snapshot != create.expected_snapshot {
                        return Err(StorageError::Backend(
                            "DuckLake CREATE TABLE AS files have inconsistent expected snapshots"
                                .into(),
                        ));
                    }
                }
            }
            Self::CreateTableWithInlinedData(create) => {
                if create.rows_inserted <= 0 {
                    return Err(StorageError::Backend(
                        "DuckLake CREATE TABLE AS inlined commit received no rows".into(),
                    ));
                }
                validate_create_table_input(&create.create)?;
                crate::validate_inlined_append_rows(
                    create.rows_inserted,
                    &create.batches,
                    None,
                    "DuckLake CREATE TABLE AS inlined commit",
                )?;
            }
            Self::CreateView(create) => {
                if create.schema_name.trim().is_empty() {
                    return Err(StorageError::Backend(
                        "DuckLake create-view schema name is empty".into(),
                    ));
                }
                if create.view_name.trim().is_empty() {
                    return Err(StorageError::Backend(
                        "DuckLake create-view view name is empty".into(),
                    ));
                }
                if create.dialect.trim().is_empty() {
                    return Err(StorageError::Backend(
                        "DuckLake create-view dialect is empty".into(),
                    ));
                }
                if create.sql.trim().is_empty() {
                    return Err(StorageError::Backend(
                        "DuckLake create-view SQL is empty".into(),
                    ));
                }
            }
            Self::DropTable(drop) => {
                if drop.schema_name.trim().is_empty() {
                    return Err(StorageError::Backend(
                        "DuckLake drop-table schema name is empty".into(),
                    ));
                }
                if drop.table_name.trim().is_empty() {
                    return Err(StorageError::Backend(
                        "DuckLake drop-table table name is empty".into(),
                    ));
                }
            }
            Self::RenameTable(rename) => {
                if rename.schema_name.trim().is_empty() {
                    return Err(StorageError::Backend(
                        "DuckLake rename-table schema name is empty".into(),
                    ));
                }
                if rename.old_table_name.trim().is_empty() {
                    return Err(StorageError::Backend(
                        "DuckLake rename-table source table name is empty".into(),
                    ));
                }
                if rename.new_table_name.trim().is_empty() {
                    return Err(StorageError::Backend(
                        "DuckLake rename-table target table name is empty".into(),
                    ));
                }
                if rename
                    .new_schema_name
                    .as_deref()
                    .is_some_and(|schema| schema.trim().is_empty())
                {
                    return Err(StorageError::Backend(
                        "DuckLake rename-table target schema name is empty".into(),
                    ));
                }
            }
            Self::DropView(drop) => {
                if drop.schema_name.trim().is_empty() {
                    return Err(StorageError::Backend(
                        "DuckLake drop-view schema name is empty".into(),
                    ));
                }
                if drop.view_name.trim().is_empty() {
                    return Err(StorageError::Backend(
                        "DuckLake drop-view view name is empty".into(),
                    ));
                }
            }
            Self::DropSchema(drop) => {
                if drop.schema_name.trim().is_empty() {
                    return Err(StorageError::Backend(
                        "DuckLake drop-schema schema name is empty".into(),
                    ));
                }
            }
            Self::ColumnNullability(change) => validate_positive_column_id(
                "DuckLake column nullability change column id",
                change.column_id.0,
            )?,
            Self::ColumnDefault(change) => validate_positive_column_id(
                "DuckLake column default change column id",
                change.column_id.0,
            )?,
            Self::ColumnType(change) => validate_positive_column_id(
                "DuckLake column type change column id",
                change.column_id.0,
            )?,
            Self::ColumnTypeTree(change) => validate_positive_column_id(
                "DuckLake column type change column id",
                change.column_id.0,
            )?,
            Self::RenameColumn(rename) => {
                validate_positive_column_id("DuckLake rename-column id", rename.column_id.0)?;
                if rename.new_column_name.trim().is_empty() {
                    return Err(StorageError::Backend(
                        "DuckLake rename-column target name is empty".into(),
                    ));
                }
            }
            Self::DropColumn(_) => {}
            Self::AddColumns(add) => crate::validate_add_column_commit(add)?,
            Self::PartitioningBatch(partitioning) => {
                if partitioning.changes.is_empty() {
                    return Err(StorageError::Backend(
                        "DuckLake partitioning batch has no table changes".into(),
                    ));
                }
            }
            Self::SortedBy(_) => {}
            Self::TruncateTable(truncate) => {
                crate::plan_truncate_table_commit(truncate)?;
            }
            Self::MergeAdjacentCompaction(compaction) => {
                crate::plan_merge_adjacent_compaction_sources(compaction)?;
            }
            Self::RewriteDataFiles(rewrite) => {
                crate::plan_rewrite_data_files_sources(rewrite)?;
            }
        }
        Ok(())
    }

    #[cfg(feature = "quack")]
    pub(crate) fn abort_context(&self) -> CommitAbortContext {
        match self {
            Self::AppendFiles(files) => CommitAbortContext {
                label: "Quack append commit",
                operation: Some("append"),
                expected_snapshot: files.first().map(|file| file.expected_snapshot),
            },
            Self::AppendInlinedData(append) => CommitAbortContext {
                label: "Quack inlined append commit",
                operation: Some("append"),
                expected_snapshot: Some(append.expected_snapshot),
            },
            Self::DeleteFiles(files) => CommitAbortContext {
                label: "Quack delete commit",
                operation: Some("delete"),
                expected_snapshot: files.first().map(|file| file.expected_snapshot),
            },
            Self::TableChanges(changes) => CommitAbortContext {
                label: "Quack table-change commit",
                operation: Some("table-change"),
                expected_snapshot: Some(changes.expected_snapshot),
            },
            Self::CreateSchema(_) => CommitAbortContext {
                label: "Quack create-schema commit",
                operation: None,
                expected_snapshot: None,
            },
            Self::CreateTable(_) => CommitAbortContext {
                label: "Quack create-table commit",
                operation: None,
                expected_snapshot: None,
            },
            Self::CreateTableWithFiles(create) => CommitAbortContext {
                label: "Quack create-table-as commit",
                operation: Some("create-table-as"),
                expected_snapshot: Some(create.expected_snapshot),
            },
            Self::CreateTableWithInlinedData(create) => CommitAbortContext {
                label: "Quack create-table-as inlined commit",
                operation: Some("create-table-as"),
                expected_snapshot: Some(create.expected_snapshot),
            },
            Self::CreateView(_) => CommitAbortContext {
                label: "Quack create-view commit",
                operation: None,
                expected_snapshot: None,
            },
            Self::DropTable(_) => CommitAbortContext {
                label: "Quack drop-table commit",
                operation: None,
                expected_snapshot: None,
            },
            Self::RenameTable(_) => CommitAbortContext {
                label: "Quack rename-table commit",
                operation: None,
                expected_snapshot: None,
            },
            Self::DropView(_) => CommitAbortContext {
                label: "Quack drop-view commit",
                operation: None,
                expected_snapshot: None,
            },
            Self::DropSchema(_) => CommitAbortContext {
                label: "Quack drop-schema commit",
                operation: None,
                expected_snapshot: None,
            },
            Self::ColumnNullability(change) => CommitAbortContext {
                label: "Quack alter-column-nullability commit",
                operation: Some("alter-column-nullability"),
                expected_snapshot: Some(change.expected_snapshot),
            },
            Self::ColumnDefault(change) => CommitAbortContext {
                label: "Quack alter-column-default commit",
                operation: Some("alter-column-default"),
                expected_snapshot: Some(change.expected_snapshot),
            },
            Self::ColumnType(change) => CommitAbortContext {
                label: "Quack alter-column-type commit",
                operation: Some("alter-column-type"),
                expected_snapshot: Some(change.expected_snapshot),
            },
            Self::ColumnTypeTree(change) => CommitAbortContext {
                label: "Quack alter-column-complex-type commit",
                operation: Some("alter-column-type"),
                expected_snapshot: Some(change.expected_snapshot),
            },
            Self::RenameColumn(rename) => CommitAbortContext {
                label: "Quack rename-column commit",
                operation: Some("rename-column"),
                expected_snapshot: Some(rename.expected_snapshot),
            },
            Self::DropColumn(drop) => CommitAbortContext {
                label: "Quack drop-column commit",
                operation: Some("drop-column"),
                expected_snapshot: Some(drop.expected_snapshot),
            },
            Self::AddColumns(add) => CommitAbortContext {
                label: "Quack add-column commit",
                operation: Some("add-column"),
                expected_snapshot: Some(add.expected_snapshot),
            },
            Self::PartitioningBatch(partitioning) => CommitAbortContext {
                label: "Quack alter-partitioning commit",
                operation: Some("alter-partitioning"),
                expected_snapshot: Some(partitioning.expected_snapshot),
            },
            Self::SortedBy(sorted_by) => CommitAbortContext {
                label: "Quack alter-sorted-by commit",
                operation: Some("alter-sorted-by"),
                expected_snapshot: Some(sorted_by.expected_snapshot),
            },
            Self::TruncateTable(truncate) => CommitAbortContext {
                label: "Quack truncate commit",
                operation: Some("truncate"),
                expected_snapshot: Some(truncate.expected_snapshot),
            },
            Self::MergeAdjacentCompaction(compaction) => CommitAbortContext {
                label: "Quack merge-adjacent compaction",
                operation: Some("merge-adjacent"),
                expected_snapshot: Some(compaction.expected_snapshot),
            },
            Self::RewriteDataFiles(rewrite) => CommitAbortContext {
                label: "Quack rewrite-data-files",
                operation: Some("rewrite-data-files"),
                expected_snapshot: Some(rewrite.expected_snapshot),
            },
        }
    }
}

fn validate_create_table_input(create: &CreateTableCommit) -> Result<()> {
    if create.schema_name.trim().is_empty() {
        return Err(StorageError::Backend(
            "DuckLake create-table schema name is empty".into(),
        ));
    }
    if create.table_name.trim().is_empty() {
        return Err(StorageError::Backend(
            "DuckLake create-table table name is empty".into(),
        ));
    }
    if create.columns.is_empty() {
        return Err(StorageError::Backend(
            "DuckLake create-table received no columns".into(),
        ));
    }
    crate::validate_create_table_columns(&create.columns)
}

fn validate_inlined_append(
    rows_appended: i64,
    schema: &TableSchema,
    batches: &[arrow::record_batch::RecordBatch],
    preserved_row_ids: Option<&[i64]>,
    context: &str,
) -> Result<()> {
    crate::validate_inlined_append_rows(rows_appended, batches, preserved_row_ids, context)?;
    for field in &schema.columns {
        if !field.children.is_empty() {
            return Err(StorageError::Unsupported(format!(
                "Inlined append writes do not yet support nested column {}",
                field.name
            )));
        }
        validate_inlined_field_type(field)?;
    }
    Ok(())
}

fn validate_inlined_field_type(field: &ducklake_catalog::SchemaField) -> Result<()> {
    match field.ducklake_type {
        ducklake_catalog::DuckLakeType::Boolean
        | ducklake_catalog::DuckLakeType::Int8
        | ducklake_catalog::DuckLakeType::Int16
        | ducklake_catalog::DuckLakeType::Int32
        | ducklake_catalog::DuckLakeType::Int64
        | ducklake_catalog::DuckLakeType::UInt8
        | ducklake_catalog::DuckLakeType::UInt16
        | ducklake_catalog::DuckLakeType::UInt32
        | ducklake_catalog::DuckLakeType::UInt64
        | ducklake_catalog::DuckLakeType::Int128
        | ducklake_catalog::DuckLakeType::UInt128
        | ducklake_catalog::DuckLakeType::Float32
        | ducklake_catalog::DuckLakeType::Float64
        | ducklake_catalog::DuckLakeType::Decimal { .. }
        | ducklake_catalog::DuckLakeType::Date
        | ducklake_catalog::DuckLakeType::Time
        | ducklake_catalog::DuckLakeType::TimeNs
        | ducklake_catalog::DuckLakeType::Timestamp
        | ducklake_catalog::DuckLakeType::TimestampS
        | ducklake_catalog::DuckLakeType::TimestampMs
        | ducklake_catalog::DuckLakeType::TimestampUs
        | ducklake_catalog::DuckLakeType::TimestampNs
        | ducklake_catalog::DuckLakeType::TimestampTz
        | ducklake_catalog::DuckLakeType::TimeTz
        | ducklake_catalog::DuckLakeType::Interval
        | ducklake_catalog::DuckLakeType::Blob
        | ducklake_catalog::DuckLakeType::Varchar
        | ducklake_catalog::DuckLakeType::Json
        | ducklake_catalog::DuckLakeType::Uuid => Ok(()),
        ref other => Err(StorageError::Unsupported(format!(
            "Inlined append writes do not support DuckLake type {other}"
        ))),
    }
}

fn validate_positive_column_id(label: &str, column_id: i64) -> Result<()> {
    if column_id <= 0 {
        return Err(StorageError::Backend(format!(
            "{label} {column_id} is not positive"
        )));
    }
    Ok(())
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) fn apply_sync_commit_request<'tx, O>(
    ops: &O,
    tx: &O::Tx<'tx>,
    request: &CommitRequest,
) -> Result<SnapshotId>
where
    O: SyncMaintenanceCommitOps
        + SyncCatalogObjectCommitOps
        + SyncCreateTableWithFilesCommitOps
        + SyncPartitioningCommitOps
        + SyncSortedByCommitOps
        + SyncColumnEvolutionCommitOps,
{
    match request {
        CommitRequest::AppendFiles(files) => {
            crate::commit::commit_append_files_sync(ops, tx, files)
        }
        CommitRequest::AppendInlinedData(append) => {
            crate::commit::commit_append_inlined_data_sync(ops, tx, append)
        }
        CommitRequest::DeleteFiles(files) => {
            crate::commit::commit_delete_files_sync(ops, tx, files)
        }
        CommitRequest::TableChanges(changes) => {
            crate::commit::commit_table_changes_sync(ops, tx, changes)
        }
        CommitRequest::CreateSchema(create) => {
            crate::commit::commit_create_schema_sync(ops, tx, create)
        }
        CommitRequest::CreateTable(create) => {
            crate::commit::commit_create_table_sync(ops, tx, create)
        }
        CommitRequest::CreateTableWithFiles(create) => {
            crate::commit::commit_create_table_with_files_sync(ops, tx, create)
        }
        CommitRequest::CreateTableWithInlinedData(create) => {
            crate::commit::commit_create_table_with_inlined_data_sync(ops, tx, create)
        }
        CommitRequest::CreateView(create) => {
            crate::commit::commit_create_view_sync(ops, tx, create)
        }
        CommitRequest::DropTable(drop) => crate::commit::commit_drop_table_sync(ops, tx, drop),
        CommitRequest::RenameTable(rename) => {
            crate::commit::commit_rename_table_sync(ops, tx, rename)
        }
        CommitRequest::DropView(drop) => crate::commit::commit_drop_view_sync(ops, tx, drop),
        CommitRequest::DropSchema(drop) => crate::commit::commit_drop_schema_sync(ops, tx, drop),
        CommitRequest::ColumnNullability(change) => {
            crate::commit::commit_column_nullability_sync(ops, tx, change)
        }
        CommitRequest::ColumnDefault(change) => {
            crate::commit::commit_column_default_sync(ops, tx, change)
        }
        CommitRequest::ColumnType(change) => {
            crate::commit::commit_column_type_sync(ops, tx, change)
        }
        CommitRequest::ColumnTypeTree(change) => {
            crate::commit::commit_column_type_tree_sync(ops, tx, change)
        }
        CommitRequest::RenameColumn(rename) => {
            crate::commit::commit_rename_column_sync(ops, tx, rename)
        }
        CommitRequest::DropColumn(drop) => crate::commit::commit_drop_column_sync(ops, tx, drop),
        CommitRequest::AddColumns(add) => crate::commit::commit_add_columns_sync(ops, tx, add),
        CommitRequest::PartitioningBatch(partitioning) => {
            crate::commit::commit_partitioning_batch_sync(ops, tx, partitioning)
        }
        CommitRequest::SortedBy(sorted_by) => {
            crate::commit::commit_sorted_by_sync(ops, tx, sorted_by)
        }
        CommitRequest::TruncateTable(truncate) => {
            crate::commit::commit_truncate_table_sync(ops, tx, truncate)
        }
        CommitRequest::MergeAdjacentCompaction(compaction) => {
            crate::commit::commit_merge_adjacent_compaction_sync(ops, tx, compaction)
        }
        CommitRequest::RewriteDataFiles(rewrite) => {
            crate::commit::commit_rewrite_data_files_sync(ops, tx, rewrite)
        }
    }
}

#[cfg(any(feature = "postgres", feature = "quack"))]
pub(crate) async fn apply_async_commit_request<O, Tx>(
    ops: &O,
    tx: &Tx,
    request: &CommitRequest,
) -> Result<SnapshotId>
where
    O: AsyncMaintenanceCommitOps<Tx>
        + AsyncCatalogObjectCommitOps<Tx>
        + AsyncCreateTableWithFilesCommitOps<Tx>
        + AsyncPartitioningCommitOps<Tx>
        + AsyncSortedByCommitOps<Tx>
        + AsyncColumnEvolutionCommitOps<Tx>
        + Sync,
    Tx: Sync + ?Sized,
{
    AsyncCommitOps::prepare_commit(ops, tx).await?;
    match request {
        CommitRequest::AppendFiles(files) => {
            crate::commit::commit_append_files_async(ops, tx, files).await
        }
        CommitRequest::AppendInlinedData(append) => {
            crate::commit::commit_append_inlined_data_async(ops, tx, append).await
        }
        CommitRequest::DeleteFiles(files) => {
            crate::commit::commit_delete_files_async(ops, tx, files).await
        }
        CommitRequest::TableChanges(changes) => {
            crate::commit::commit_table_changes_async(ops, tx, changes).await
        }
        CommitRequest::CreateSchema(create) => {
            crate::commit::commit_create_schema_async(ops, tx, create).await
        }
        CommitRequest::CreateTable(create) => {
            crate::commit::commit_create_table_async(ops, tx, create).await
        }
        CommitRequest::CreateTableWithFiles(create) => {
            crate::commit::commit_create_table_with_files_async(ops, tx, create).await
        }
        CommitRequest::CreateTableWithInlinedData(create) => {
            crate::commit::commit_create_table_with_inlined_data_async(ops, tx, create).await
        }
        CommitRequest::CreateView(create) => {
            crate::commit::commit_create_view_async(ops, tx, create).await
        }
        CommitRequest::DropTable(drop) => {
            crate::commit::commit_drop_table_async(ops, tx, drop).await
        }
        CommitRequest::RenameTable(rename) => {
            crate::commit::commit_rename_table_async(ops, tx, rename).await
        }
        CommitRequest::DropView(drop) => crate::commit::commit_drop_view_async(ops, tx, drop).await,
        CommitRequest::DropSchema(drop) => {
            crate::commit::commit_drop_schema_async(ops, tx, drop).await
        }
        CommitRequest::ColumnNullability(change) => {
            crate::commit::commit_column_nullability_async(ops, tx, change).await
        }
        CommitRequest::ColumnDefault(change) => {
            crate::commit::commit_column_default_async(ops, tx, change).await
        }
        CommitRequest::ColumnType(change) => {
            crate::commit::commit_column_type_async(ops, tx, change).await
        }
        CommitRequest::ColumnTypeTree(change) => {
            crate::commit::commit_column_type_tree_async(ops, tx, change).await
        }
        CommitRequest::RenameColumn(rename) => {
            crate::commit::commit_rename_column_async(ops, tx, rename).await
        }
        CommitRequest::DropColumn(drop) => {
            crate::commit::commit_drop_column_async(ops, tx, drop).await
        }
        CommitRequest::AddColumns(add) => {
            crate::commit::commit_add_columns_async(ops, tx, add).await
        }
        CommitRequest::PartitioningBatch(partitioning) => {
            crate::commit::commit_partitioning_batch_async(ops, tx, partitioning).await
        }
        CommitRequest::SortedBy(sorted_by) => {
            crate::commit::commit_sorted_by_async(ops, tx, sorted_by).await
        }
        CommitRequest::TruncateTable(truncate) => {
            crate::commit::commit_truncate_table_async(ops, tx, truncate).await
        }
        CommitRequest::MergeAdjacentCompaction(compaction) => {
            crate::commit::commit_merge_adjacent_compaction_async(ops, tx, compaction).await
        }
        CommitRequest::RewriteDataFiles(rewrite) => {
            crate::commit::commit_rewrite_data_files_async(ops, tx, rewrite).await
        }
    }
}
