use async_trait::async_trait;
use std::future::Future;

#[derive(Debug, Clone, Copy)]
pub(crate) struct CompactionSourceSqlRefs<'a> {
    pub(crate) data_file_ref: &'a str,
    pub(crate) table_ref: &'a str,
    pub(crate) schema_ref: &'a str,
    pub(crate) table_id_expr: &'a str,
    pub(crate) snapshot_expr: &'a str,
}

#[derive(Debug, Clone, Copy)]
pub(crate) struct CompactionDeleteSqlRefs<'a> {
    pub(crate) delete_file_ref: &'a str,
    pub(crate) table_ref: &'a str,
    pub(crate) schema_ref: &'a str,
    pub(crate) snapshot_expr: &'a str,
}

#[derive(Debug, Clone, Copy)]
pub(crate) struct SnapshotTableSqlRefs<'a> {
    pub(crate) table_ref: &'a str,
    pub(crate) snapshot_ref: &'a str,
}

#[derive(Debug, Clone, Copy)]
pub(crate) struct SnapshotDataFileSqlRefs<'a> {
    pub(crate) data_file_ref: &'a str,
    pub(crate) table_ref: &'a str,
    pub(crate) schema_ref: &'a str,
    pub(crate) snapshot_ref: &'a str,
}

#[derive(Debug, Clone, Copy)]
pub(crate) struct SnapshotDeleteFileSqlRefs<'a> {
    pub(crate) delete_file_ref: &'a str,
    pub(crate) table_ref: &'a str,
    pub(crate) schema_ref: &'a str,
    pub(crate) snapshot_ref: &'a str,
}

macro_rules! impl_async_commit_ops {
    (
        impl<$($generics:lifetime),+> $ops:ty, $tx:ty {
            latest_snapshot_ids = |$latest_self:ident, $latest_tx:ident| $latest:expr,
            table_exists = |$exists_self:ident, $exists_tx:ident, $table_name:ident| $exists:expr,
            insert_snapshot = |$snapshot_self:ident, $snapshot_tx:ident, $snapshot_id:ident, $schema_version:ident, $next_catalog_id:ident, $next_file_id:ident| $insert_snapshot:expr,
            insert_snapshot_change = |$change_self:ident, $change_tx:ident, $change_snapshot_id:ident, $changes_made:ident, $commit_message:ident| $insert_change:expr
            $(, extra { $($extra:tt)* })? $(,)?
        }
    ) => {
        impl_async_commit_ops!(
            @impl [<$($generics),+>] $ops, $tx {
                latest_snapshot_ids = |$latest_self, $latest_tx| $latest,
                table_exists = |$exists_self, $exists_tx, $table_name| $exists,
                insert_snapshot = |$snapshot_self, $snapshot_tx, $snapshot_id, $schema_version, $next_catalog_id, $next_file_id| $insert_snapshot,
                insert_snapshot_change = |$change_self, $change_tx, $change_snapshot_id, $changes_made, $commit_message| $insert_change
                $(, extra { $($extra)* })?
            }
        );
    };
    (
        impl $ops:ty, $tx:ty {
            latest_snapshot_ids = |$latest_self:ident, $latest_tx:ident| $latest:expr,
            table_exists = |$exists_self:ident, $exists_tx:ident, $table_name:ident| $exists:expr,
            insert_snapshot = |$snapshot_self:ident, $snapshot_tx:ident, $snapshot_id:ident, $schema_version:ident, $next_catalog_id:ident, $next_file_id:ident| $insert_snapshot:expr,
            insert_snapshot_change = |$change_self:ident, $change_tx:ident, $change_snapshot_id:ident, $changes_made:ident, $commit_message:ident| $insert_change:expr
            $(, extra { $($extra:tt)* })? $(,)?
        }
    ) => {
        impl_async_commit_ops!(
            @impl [] $ops, $tx {
                latest_snapshot_ids = |$latest_self, $latest_tx| $latest,
                table_exists = |$exists_self, $exists_tx, $table_name| $exists,
                insert_snapshot = |$snapshot_self, $snapshot_tx, $snapshot_id, $schema_version, $next_catalog_id, $next_file_id| $insert_snapshot,
                insert_snapshot_change = |$change_self, $change_tx, $change_snapshot_id, $changes_made, $commit_message| $insert_change
                $(, extra { $($extra)* })?
            }
        );
    };
    (
        @impl [$($impl_generics:tt)*] $ops:ty, $tx:ty {
            latest_snapshot_ids = |$latest_self:ident, $latest_tx:ident| $latest:expr,
            table_exists = |$exists_self:ident, $exists_tx:ident, $table_name:ident| $exists:expr,
            insert_snapshot = |$snapshot_self:ident, $snapshot_tx:ident, $snapshot_id:ident, $schema_version:ident, $next_catalog_id:ident, $next_file_id:ident| $insert_snapshot:expr,
            insert_snapshot_change = |$change_self:ident, $change_tx:ident, $change_snapshot_id:ident, $changes_made:ident, $commit_message:ident| $insert_change:expr
            $(, extra { $($extra:tt)* })? $(,)?
        }
    ) => {
        #[async_trait]
        impl$($impl_generics)* $crate::commit::AsyncCommitOps<$tx> for $ops {
            async fn latest_snapshot_ids(
                &self,
                $latest_tx: &$tx,
            ) -> $crate::Result<$crate::commit::CommitSnapshotIds> {
                let $latest_self = self;
                $latest.await
            }

            async fn table_exists(
                &self,
                $exists_tx: &$tx,
                $table_name: &str,
            ) -> $crate::Result<bool> {
                let $exists_self = self;
                $exists.await
            }

            async fn insert_snapshot(
                &self,
                $snapshot_tx: &$tx,
                $snapshot_id: i64,
                $schema_version: i64,
                $next_catalog_id: i64,
                $next_file_id: i64,
            ) -> $crate::Result<()> {
                let $snapshot_self = self;
                $insert_snapshot.await
            }

            async fn insert_snapshot_change(
                &self,
                $change_tx: &$tx,
                $change_snapshot_id: i64,
                $changes_made: &str,
                $commit_message: &str,
            ) -> $crate::Result<()> {
                let $change_self = self;
                $insert_change.await
            }

            $($($extra)*)?
        }
    };
}

#[async_trait]
pub(crate) trait AsyncCatalogQueryCommitExecutor<Tx: Sync + ?Sized>:
    crate::commit::AsyncCommitOps<Tx>
{
    async fn exec_catalog_execute(
        &self,
        tx: &Tx,
        query: &crate::catalog_queries::CatalogQuery,
    ) -> crate::Result<usize>;

    async fn exec_catalog_query<T: Send>(
        &self,
        tx: &Tx,
        query: &crate::catalog_queries::CatalogQuery,
        mapper: for<'row> fn(&'row dyn crate::embedded_rows::EmbeddedRow) -> crate::Result<T>,
    ) -> crate::Result<Vec<T>>;

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

pub(crate) async fn load_latest_snapshot_ids<Ops, Tx>(
    ops: &Ops,
    tx: &Tx,
    operation: &str,
) -> crate::Result<crate::commit::CommitSnapshotIds>
where
    Ops: AsyncCatalogQueryCommitExecutor<Tx> + Sync,
    Tx: Sync + ?Sized,
{
    ops.exec_catalog_query(
        tx,
        &crate::catalog_queries::latest_snapshot_ids(),
        crate::catalog_queries::latest_snapshot_ids_value,
    )
    .await?
    .into_iter()
    .next()
    .ok_or_else(|| crate::StorageError::Backend(format!("DuckLake {operation} found no snapshots")))
}

pub(crate) async fn replace_overwritten_delete_file<
    Ops,
    Tx,
    LoadDataPath,
    LoadDataPathFut,
    ScheduleCleanup,
    ScheduleCleanupFut,
>(
    ops: &Ops,
    tx: &Tx,
    file: &crate::DeleteFileCommit,
    new_snapshot: i64,
    load_data_path: LoadDataPath,
    schedule_cleanup: ScheduleCleanup,
) -> crate::Result<()>
where
    Ops: AsyncCatalogQueryCommitExecutor<Tx> + Sync,
    Tx: Sync + ?Sized,
    LoadDataPath: FnOnce() -> LoadDataPathFut,
    LoadDataPathFut: Future<Output = crate::Result<String>>,
    ScheduleCleanup: FnOnce(crate::CleanupMetadataFile) -> ScheduleCleanupFut,
    ScheduleCleanupFut: Future<Output = crate::Result<()>>,
{
    let Some(overwritten) = file.overwritten_delete_file_id else {
        return Ok(());
    };

    if file.begin_snapshot.is_none() {
        let query = crate::catalog_queries::close_overwritten_delete_file(file, new_snapshot)
            .ok_or_else(|| {
                crate::StorageError::Backend("DuckLake overwrite missing delete file id".into())
            })?;
        let updated = ops.exec_catalog_execute(tx, &query).await?;
        if updated != 1 {
            return Err(crate::StorageError::Backend(format!(
                "DuckLake overwrite expected one active delete file {}, updated {} rows",
                overwritten.0, updated
            )));
        }
        return Ok(());
    }

    let data_path = load_data_path().await?;
    let cleanup_row = crate::embedded_maintenance::single_overwritten_cleanup_row(
        ops.exec_catalog_query(
            tx,
            &crate::catalog_queries::overwritten_delete_file_cleanup_row(
                overwritten,
                file.table_id,
            ),
            crate::embedded_maintenance::cleanup_file_path_row,
        )
        .await?,
        overwritten,
    )?
    .into_cleanup_file(&data_path)?;

    let query = crate::catalog_queries::delete_overwritten_delete_file(file).ok_or_else(|| {
        crate::StorageError::Backend("DuckLake overwrite missing delete file id".into())
    })?;
    let deleted = ops.exec_catalog_execute(tx, &query).await?;
    if deleted != 1 {
        return Err(crate::StorageError::Backend(format!(
            "DuckLake overwrite expected one delete file {}, deleted {} rows",
            overwritten.0, deleted
        )));
    }

    schedule_cleanup(cleanup_row).await
}

macro_rules! impl_async_catalog_query_commit_executor {
    (
        impl<$($generics:lifetime),+> $ops:ty, $tx:ty {
            execute = |$execute_self:ident, $execute_tx:ident, $execute_query:ident| $execute:expr,
            query = |$query_self:ident, $query_tx:ident, $query_query:ident, $mapper:ident| $query:expr,
            table_exists = |$exists_self:ident, $exists_tx:ident, $table_name:ident| $exists:expr $(,)?
        }
    ) => {
        impl_async_catalog_query_commit_executor!(
            @impl [<$($generics),+>] $ops, $tx {
                execute = |$execute_self, $execute_tx, $execute_query| $execute,
                query = |$query_self, $query_tx, $query_query, $mapper| $query,
                table_exists = |$exists_self, $exists_tx, $table_name| $exists,
            }
        );
    };
    (
        impl $ops:ty, $tx:ty {
            execute = |$execute_self:ident, $execute_tx:ident, $execute_query:ident| $execute:expr,
            query = |$query_self:ident, $query_tx:ident, $query_query:ident, $mapper:ident| $query:expr,
            table_exists = |$exists_self:ident, $exists_tx:ident, $table_name:ident| $exists:expr $(,)?
        }
    ) => {
        impl_async_catalog_query_commit_executor!(
            @impl [] $ops, $tx {
                execute = |$execute_self, $execute_tx, $execute_query| $execute,
                query = |$query_self, $query_tx, $query_query, $mapper| $query,
                table_exists = |$exists_self, $exists_tx, $table_name| $exists,
            }
        );
    };
    (
        @impl [$($impl_generics:tt)*] $ops:ty, $tx:ty {
            execute = |$execute_self:ident, $execute_tx:ident, $execute_query:ident| $execute:expr,
            query = |$query_self:ident, $query_tx:ident, $query_query:ident, $mapper:ident| $query:expr,
            table_exists = |$exists_self:ident, $exists_tx:ident, $table_name:ident| $exists:expr $(,)?
        }
    ) => {
        #[async_trait]
        impl$($impl_generics)* $crate::catalog_async_commit_methods::AsyncCatalogQueryCommitExecutor<$tx>
            for $ops
        {
            async fn exec_catalog_execute(
                &self,
                $execute_tx: &$tx,
                $execute_query: &$crate::catalog_queries::CatalogQuery,
            ) -> $crate::Result<usize> {
                let $execute_self = self;
                $execute.await
            }

            async fn exec_catalog_query<T: Send>(
                &self,
                $query_tx: &$tx,
                $query_query: &$crate::catalog_queries::CatalogQuery,
                $mapper: for<'row> fn(
                    &'row dyn $crate::embedded_rows::EmbeddedRow,
                ) -> $crate::Result<T>,
            ) -> $crate::Result<Vec<T>> {
                let $query_self = self;
                $query.await
            }

            async fn exec_catalog_table_exists(
                &self,
                $exists_tx: &$tx,
                $table_name: &str,
            ) -> $crate::Result<bool> {
                let $exists_self = self;
                $exists.await
            }
        }
    };
}

pub(crate) async fn load_compaction_source_rows<
    CheckContext,
    CheckContextFut,
    LoadDataPath,
    LoadDataPathFut,
    QueryWithoutPath,
    QueryWithoutPathFut,
    QueryWithPath,
    QueryWithPathFut,
>(
    table_id: ducklake_catalog::TableId,
    snapshot: i64,
    source_ids: &[i64],
    refs: CompactionSourceSqlRefs<'_>,
    cleanup_path_context_available: CheckContext,
    load_data_path: LoadDataPath,
    query_without_table_path: QueryWithoutPath,
    query_with_table_path: QueryWithPath,
) -> crate::Result<Vec<crate::commit::MaintenanceSourceFile>>
where
    CheckContext: FnOnce() -> CheckContextFut,
    CheckContextFut: Future<Output = crate::Result<bool>>,
    LoadDataPath: FnOnce() -> LoadDataPathFut,
    LoadDataPathFut: Future<Output = crate::Result<String>>,
    QueryWithoutPath: FnOnce(String, Vec<i64>) -> QueryWithoutPathFut,
    QueryWithoutPathFut: Future<Output = crate::Result<Vec<crate::commit::MaintenanceSourceFile>>>,
    QueryWithPath: FnOnce(String, Vec<i64>, String) -> QueryWithPathFut,
    QueryWithPathFut: Future<Output = crate::Result<Vec<crate::commit::MaintenanceSourceFile>>>,
{
    let params = vec![table_id.0, snapshot];
    if !cleanup_path_context_available().await? {
        let sql = crate::embedded_maintenance::compaction_source_rows_without_table_path_sql(
            source_ids,
            refs.data_file_ref,
            refs.table_id_expr,
            refs.snapshot_expr,
        );
        return query_without_table_path(sql, params).await;
    }

    let sql = crate::embedded_maintenance::compaction_source_rows_sql_with_refs(
        source_ids,
        refs.data_file_ref,
        refs.table_ref,
        refs.schema_ref,
        refs.table_id_expr,
        refs.snapshot_expr,
    );
    let data_path = load_data_path().await?;
    query_with_table_path(sql, params, data_path).await
}

pub(crate) async fn load_compaction_delete_rows<
    DeleteTableExists,
    DeleteTableExistsFut,
    CheckContext,
    CheckContextFut,
    LoadDataPath,
    LoadDataPathFut,
    QueryWithoutPath,
    QueryWithoutPathFut,
    QueryWithPath,
    QueryWithPathFut,
>(
    snapshot: i64,
    source_ids: &[i64],
    refs: CompactionDeleteSqlRefs<'_>,
    delete_table_exists: DeleteTableExists,
    cleanup_path_context_available: CheckContext,
    load_data_path: LoadDataPath,
    query_without_table_path: QueryWithoutPath,
    query_with_table_path: QueryWithPath,
) -> crate::Result<Vec<crate::CleanupMetadataFile>>
where
    DeleteTableExists: FnOnce() -> DeleteTableExistsFut,
    DeleteTableExistsFut: Future<Output = crate::Result<bool>>,
    CheckContext: FnOnce() -> CheckContextFut,
    CheckContextFut: Future<Output = crate::Result<bool>>,
    LoadDataPath: FnOnce() -> LoadDataPathFut,
    LoadDataPathFut: Future<Output = crate::Result<String>>,
    QueryWithoutPath: FnOnce(String, Vec<i64>) -> QueryWithoutPathFut,
    QueryWithoutPathFut: Future<Output = crate::Result<Vec<crate::CleanupMetadataFile>>>,
    QueryWithPath: FnOnce(String, Vec<i64>, String) -> QueryWithPathFut,
    QueryWithPathFut: Future<Output = crate::Result<Vec<crate::CleanupMetadataFile>>>,
{
    if !delete_table_exists().await? {
        return Ok(Vec::new());
    }

    let params = vec![snapshot];
    if !cleanup_path_context_available().await? {
        let sql = crate::embedded_maintenance::compaction_delete_rows_without_table_path_sql(
            source_ids,
            refs.delete_file_ref,
            refs.snapshot_expr,
        );
        return query_without_table_path(sql, params).await;
    }

    let sql = crate::embedded_maintenance::compaction_delete_rows_sql_with_refs(
        source_ids,
        refs.delete_file_ref,
        refs.table_ref,
        refs.schema_ref,
        refs.snapshot_expr,
    );
    let data_path = load_data_path().await?;
    query_with_table_path(sql, params, data_path).await
}

pub(crate) async fn load_tables_unreferenced_by_snapshots<
    TableExists,
    TableExistsFut,
    Query,
    QueryFut,
>(
    refs: SnapshotTableSqlRefs<'_>,
    table_exists: TableExists,
    query: Query,
) -> crate::Result<Vec<i64>>
where
    TableExists: FnOnce() -> TableExistsFut,
    TableExistsFut: Future<Output = crate::Result<bool>>,
    Query: FnOnce(String) -> QueryFut,
    QueryFut: Future<Output = crate::Result<Vec<i64>>>,
{
    if !table_exists().await? {
        return Ok(Vec::new());
    }
    query(
        crate::embedded_maintenance::tables_unreferenced_by_snapshots_sql_with_refs(
            refs.table_ref,
            refs.snapshot_ref,
        ),
    )
    .await
}

pub(crate) async fn load_data_files_unreferenced_by_snapshots<
    DataFileTableExists,
    DataFileTableExistsFut,
    CheckContext,
    CheckContextFut,
    LoadDataPath,
    LoadDataPathFut,
    QueryWithoutPath,
    QueryWithoutPathFut,
    QueryWithPath,
    QueryWithPathFut,
>(
    deleted_table_ids: &[i64],
    refs: SnapshotDataFileSqlRefs<'_>,
    data_file_table_exists: DataFileTableExists,
    cleanup_path_context_available: CheckContext,
    load_data_path: LoadDataPath,
    query_without_table_path: QueryWithoutPath,
    query_with_table_path: QueryWithPath,
) -> crate::Result<Vec<crate::CleanupMetadataFile>>
where
    DataFileTableExists: FnOnce() -> DataFileTableExistsFut,
    DataFileTableExistsFut: Future<Output = crate::Result<bool>>,
    CheckContext: FnOnce() -> CheckContextFut,
    CheckContextFut: Future<Output = crate::Result<bool>>,
    LoadDataPath: FnOnce() -> LoadDataPathFut,
    LoadDataPathFut: Future<Output = crate::Result<String>>,
    QueryWithoutPath: FnOnce(String) -> QueryWithoutPathFut,
    QueryWithoutPathFut: Future<Output = crate::Result<Vec<crate::CleanupMetadataFile>>>,
    QueryWithPath: FnOnce(String, String) -> QueryWithPathFut,
    QueryWithPathFut: Future<Output = crate::Result<Vec<crate::CleanupMetadataFile>>>,
{
    if !data_file_table_exists().await? {
        return Ok(Vec::new());
    }
    if !cleanup_path_context_available().await? {
        return query_without_table_path(
            crate::embedded_maintenance::data_files_unreferenced_by_snapshots_without_table_path_sql(
                deleted_table_ids,
                refs.data_file_ref,
                refs.snapshot_ref,
            ),
        )
        .await;
    }
    let sql = crate::embedded_maintenance::data_files_unreferenced_by_snapshots_sql_with_refs(
        deleted_table_ids,
        refs.data_file_ref,
        refs.table_ref,
        refs.schema_ref,
        refs.snapshot_ref,
    );
    let data_path = load_data_path().await?;
    query_with_table_path(sql, data_path).await
}

pub(crate) async fn load_delete_files_unreferenced_by_snapshots<
    DeleteFileTableExists,
    DeleteFileTableExistsFut,
    CheckContext,
    CheckContextFut,
    LoadDataPath,
    LoadDataPathFut,
    QueryWithoutPath,
    QueryWithoutPathFut,
    QueryWithPath,
    QueryWithPathFut,
>(
    deleted_table_ids: &[i64],
    deleted_data_file_ids: &[i64],
    refs: SnapshotDeleteFileSqlRefs<'_>,
    delete_file_table_exists: DeleteFileTableExists,
    cleanup_path_context_available: CheckContext,
    load_data_path: LoadDataPath,
    query_without_table_path: QueryWithoutPath,
    query_with_table_path: QueryWithPath,
) -> crate::Result<Vec<crate::CleanupMetadataFile>>
where
    DeleteFileTableExists: FnOnce() -> DeleteFileTableExistsFut,
    DeleteFileTableExistsFut: Future<Output = crate::Result<bool>>,
    CheckContext: FnOnce() -> CheckContextFut,
    CheckContextFut: Future<Output = crate::Result<bool>>,
    LoadDataPath: FnOnce() -> LoadDataPathFut,
    LoadDataPathFut: Future<Output = crate::Result<String>>,
    QueryWithoutPath: FnOnce(String) -> QueryWithoutPathFut,
    QueryWithoutPathFut: Future<Output = crate::Result<Vec<crate::CleanupMetadataFile>>>,
    QueryWithPath: FnOnce(String, String) -> QueryWithPathFut,
    QueryWithPathFut: Future<Output = crate::Result<Vec<crate::CleanupMetadataFile>>>,
{
    if !delete_file_table_exists().await? {
        return Ok(Vec::new());
    }
    if !cleanup_path_context_available().await? {
        return query_without_table_path(
            crate::embedded_maintenance::delete_files_unreferenced_by_snapshots_without_table_path_sql(
                deleted_table_ids,
                deleted_data_file_ids,
                refs.delete_file_ref,
                refs.snapshot_ref,
            ),
        )
        .await;
    }
    let sql = crate::embedded_maintenance::delete_files_unreferenced_by_snapshots_sql_with_refs(
        deleted_table_ids,
        deleted_data_file_ids,
        refs.delete_file_ref,
        refs.table_ref,
        refs.schema_ref,
        refs.snapshot_ref,
    );
    let data_path = load_data_path().await?;
    query_with_table_path(sql, data_path).await
}

#[async_trait]
pub(crate) trait AsyncTableDataCommitExecutor<Tx: Sync + ?Sized>:
    AsyncCatalogQueryCommitExecutor<Tx>
{
    async fn exec_table_data_execute(
        &self,
        tx: &Tx,
        query: &crate::catalog_queries::CatalogQuery,
    ) -> crate::Result<usize> {
        self.exec_catalog_execute(tx, query).await
    }

    async fn exec_table_data_query_optional_i64(
        &self,
        tx: &Tx,
        query: &crate::catalog_queries::CatalogQuery,
    ) -> crate::Result<Option<i64>> {
        Ok(self
            .exec_catalog_query(tx, query, crate::catalog_queries::opt_i64_value)
            .await?
            .into_iter()
            .next()
            .flatten())
    }

    async fn exec_table_column_stats_exists(&self, tx: &Tx) -> crate::Result<bool> {
        self.exec_catalog_table_exists(tx, "ducklake_table_column_stats")
            .await
    }

    async fn exec_table_data_query_column_types(
        &self,
        tx: &Tx,
        query: &crate::catalog_queries::CatalogQuery,
    ) -> crate::Result<Vec<(ducklake_catalog::ColumnId, ducklake_catalog::DuckLakeType)>> {
        self.exec_catalog_query(tx, query, crate::catalog_queries::column_type_value)
            .await
    }

    async fn exec_table_data_query_table_column_stats(
        &self,
        tx: &Tx,
        query: &crate::catalog_queries::CatalogQuery,
    ) -> crate::Result<Vec<ducklake_catalog::TableColumnStats>> {
        self.exec_catalog_query(tx, query, crate::embedded_reads::table_column_stats_row)
            .await
    }

    async fn exec_upsert_table_column_stats(
        &self,
        tx: &Tx,
        stats: ducklake_catalog::TableColumnStats,
    ) -> crate::Result<()>;

    async fn exec_next_row_id_for_append(
        &self,
        tx: &Tx,
        table_id: ducklake_catalog::TableId,
        has_table_stats: bool,
    ) -> crate::Result<Option<i64>> {
        crate::stats_engine::next_row_id_for_append_from_table_stats_async_query(
            |query| async move { self.exec_table_data_query_optional_i64(tx, &query).await },
            table_id,
            has_table_stats,
        )
        .await
    }

    async fn exec_update_table_column_stats_for_append(
        &self,
        tx: &Tx,
        table_id: ducklake_catalog::TableId,
        latest_snapshot: ducklake_catalog::SnapshotId,
        files: &[crate::AppendFile],
    ) -> crate::Result<()> {
        crate::stats_engine::update_table_column_stats_for_append_async(
            || async move { self.exec_table_column_stats_exists(tx).await },
            |query| async move { self.exec_table_data_query_column_types(tx, &query).await },
            |query| async move {
                self.exec_table_data_query_table_column_stats(tx, &query)
                    .await
            },
            |stats| async move { self.exec_upsert_table_column_stats(tx, stats).await },
            table_id,
            latest_snapshot,
            files,
        )
        .await
    }

    fn exec_inlined_sql_dialect(&self) -> crate::inlined_values::InlinedSqlDialect;

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

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

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

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

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

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

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

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

macro_rules! impl_async_table_data_commit_executor {
    (
        impl<$($generics:lifetime),+> $ops:ty, $tx:ty {
            dialect = $dialect:expr,
            upsert_table_column_stats = |$upsert_self:ident, $upsert_tx:ident, $stats:ident| $upsert:expr,
            inlined_executor = |$inlined_self:ident, $inlined_tx:ident| $inlined_executor:expr,
            replace_overwritten_delete_file = |$replace_self:ident, $replace_tx:ident, $file:ident, $new_snapshot:ident| $replace:expr,
            $(extra { $($extra:tt)* })? $(,)?
        }
    ) => {
        impl_async_table_data_commit_executor!(
            @impl [<$($generics),+>] $ops, $tx {
                dialect = $dialect,
                upsert_table_column_stats = |$upsert_self, $upsert_tx, $stats| $upsert,
                inlined_executor = |$inlined_self, $inlined_tx| $inlined_executor,
                replace_overwritten_delete_file = |$replace_self, $replace_tx, $file, $new_snapshot| $replace,
                extra { $($($extra)*)? }
            }
        );
    };
    (
        impl $ops:ty, $tx:ty {
            dialect = $dialect:expr,
            upsert_table_column_stats = |$upsert_self:ident, $upsert_tx:ident, $stats:ident| $upsert:expr,
            inlined_executor = |$inlined_self:ident, $inlined_tx:ident| $inlined_executor:expr,
            replace_overwritten_delete_file = |$replace_self:ident, $replace_tx:ident, $file:ident, $new_snapshot:ident| $replace:expr,
            $(extra { $($extra:tt)* })? $(,)?
        }
    ) => {
        impl_async_table_data_commit_executor!(
            @impl [] $ops, $tx {
                dialect = $dialect,
                upsert_table_column_stats = |$upsert_self, $upsert_tx, $stats| $upsert,
                inlined_executor = |$inlined_self, $inlined_tx| $inlined_executor,
                replace_overwritten_delete_file = |$replace_self, $replace_tx, $file, $new_snapshot| $replace,
                extra { $($($extra)*)? }
            }
        );
    };
    (
        @impl [$($impl_generics:tt)*] $ops:ty, $tx:ty {
            dialect = $dialect:expr,
            upsert_table_column_stats = |$upsert_self:ident, $upsert_tx:ident, $stats:ident| $upsert:expr,
            inlined_executor = |$inlined_self:ident, $inlined_tx:ident| $inlined_executor:expr,
            replace_overwritten_delete_file = |$replace_self:ident, $replace_tx:ident, $file:ident, $new_snapshot:ident| $replace:expr,
            extra { $($extra:tt)* }
        }
    ) => {
        #[async_trait]
        impl$($impl_generics)* $crate::catalog_async_commit_methods::AsyncTableDataCommitExecutor<$tx>
            for $ops
        {
            async fn exec_upsert_table_column_stats(
                &self,
                $upsert_tx: &$tx,
                $stats: ducklake_catalog::TableColumnStats,
            ) -> $crate::Result<()> {
                let $upsert_self = self;
                $upsert.await
            }

            fn exec_inlined_sql_dialect(&self) -> $crate::inlined_values::InlinedSqlDialect {
                $dialect
            }

            async fn exec_ensure_inlined_data_table(
                &self,
                tx: &$tx,
                table_id: ducklake_catalog::TableId,
                schema_version: i64,
                schema: &ducklake_catalog::TableSchema,
            ) -> $crate::Result<String> {
                let $inlined_self = self;
                let $inlined_tx = tx;
                let executor = $inlined_executor;
                $crate::catalog_async_inlined_mutation_methods::ensure_inlined_data_table(
                    &executor,
                    table_id,
                    schema_version,
                    schema,
                )
                .await
            }

            async fn exec_insert_inlined_batches(
                &self,
                tx: &$tx,
                table_name: &str,
                schema: &ducklake_catalog::TableSchema,
                batches: &[arrow::record_batch::RecordBatch],
                preserved_row_ids: Option<&[i64]>,
                row_id_start: i64,
                begin_snapshot: i64,
            ) -> $crate::Result<()> {
                let $inlined_self = self;
                let $inlined_tx = tx;
                let executor = $inlined_executor;
                $crate::catalog_async_inlined_mutation_methods::insert_inlined_batches(
                    &executor,
                    table_name,
                    schema,
                    batches,
                    preserved_row_ids,
                    row_id_start,
                    begin_snapshot,
                )
                .await
            }

            async fn exec_replace_overwritten_delete_file(
                &self,
                $replace_tx: &$tx,
                $file: &$crate::DeleteFileCommit,
                $new_snapshot: i64,
            ) -> $crate::Result<()> {
                let $replace_self = self;
                $replace.await
            }

            async fn exec_write_inlined_file_deletions(
                &self,
                tx: &$tx,
                table_id: ducklake_catalog::TableId,
                deletions: &[$crate::InlinedFileDeletionCommit],
                new_snapshot: i64,
            ) -> $crate::Result<()> {
                let $inlined_self = self;
                let $inlined_tx = tx;
                let executor = $inlined_executor;
                $crate::catalog_async_inlined_mutation_methods::write_inlined_file_deletions(
                    &executor,
                    table_id,
                    deletions,
                    new_snapshot,
                )
                .await
            }

            async fn exec_apply_inlined_data_deletions(
                &self,
                tx: &$tx,
                deletions: &[$crate::InlinedDataDeletionCommit],
                new_snapshot: i64,
            ) -> $crate::Result<i64> {
                let $inlined_self = self;
                let $inlined_tx = tx;
                let executor = $inlined_executor;
                $crate::catalog_async_inlined_mutation_methods::apply_inlined_data_deletions(
                    &executor,
                    deletions,
                    new_snapshot,
                )
                .await
            }

            async fn exec_delete_flushed_inlined_data(
                &self,
                tx: &$tx,
                flushed_tables: &[$crate::FlushedInlinedDataCommit],
                flush_snapshot: i64,
            ) -> $crate::Result<i64> {
                let $inlined_self = self;
                let $inlined_tx = tx;
                let executor = $inlined_executor;
                $crate::catalog_async_inlined_mutation_methods::delete_flushed_inlined_data(
                    &executor,
                    flushed_tables,
                    flush_snapshot,
                )
                .await
            }

            async fn exec_delete_flushed_inlined_file_deletions(
                &self,
                tx: &$tx,
                table_id: ducklake_catalog::TableId,
                flushed_deletions: &[$crate::FlushedInlinedFileDeletionCommit],
            ) -> $crate::Result<()> {
                let $inlined_self = self;
                let $inlined_tx = tx;
                let executor = $inlined_executor;
                $crate::catalog_async_inlined_mutation_methods::delete_flushed_inlined_file_deletions(
                    &executor,
                    table_id,
                    flushed_deletions,
                )
                .await
            }

            $($extra)*
        }
    };
}

#[async_trait]
impl<Tx, O> crate::commit::AsyncTableDataCommitOps<Tx> for O
where
    Tx: Sync + ?Sized,
    O: AsyncTableDataCommitExecutor<Tx> + Sync,
{
    async fn table_data_execute(
        &self,
        tx: &Tx,
        query: &crate::catalog_queries::CatalogQuery,
    ) -> crate::Result<usize> {
        self.exec_table_data_execute(tx, query).await
    }

    async fn next_row_id_for_append(
        &self,
        tx: &Tx,
        table_id: ducklake_catalog::TableId,
        has_table_stats: bool,
    ) -> crate::Result<Option<i64>> {
        self.exec_next_row_id_for_append(tx, table_id, has_table_stats)
            .await
    }

    async fn update_table_column_stats_for_append(
        &self,
        tx: &Tx,
        table_id: ducklake_catalog::TableId,
        latest_snapshot: ducklake_catalog::SnapshotId,
        files: &[crate::AppendFile],
    ) -> crate::Result<()> {
        self.exec_update_table_column_stats_for_append(tx, table_id, latest_snapshot, files)
            .await
    }

    fn validate_inlined_append(
        &self,
        rows_appended: i64,
        schema: &ducklake_catalog::TableSchema,
        batches: &[arrow::record_batch::RecordBatch],
        preserved_row_ids: Option<&[i64]>,
        context: &str,
    ) -> crate::Result<()> {
        crate::inlined_values::validate_inlined_append_for_dialect(
            rows_appended,
            schema,
            batches,
            preserved_row_ids,
            context,
            self.exec_inlined_sql_dialect(),
        )
    }

    async fn ensure_inlined_data_table(
        &self,
        tx: &Tx,
        table_id: ducklake_catalog::TableId,
        schema_version: i64,
        schema: &ducklake_catalog::TableSchema,
    ) -> crate::Result<String> {
        self.exec_ensure_inlined_data_table(tx, table_id, schema_version, schema)
            .await
    }

    async fn insert_inlined_batches(
        &self,
        tx: &Tx,
        table_name: &str,
        schema: &ducklake_catalog::TableSchema,
        batches: &[arrow::record_batch::RecordBatch],
        preserved_row_ids: Option<&[i64]>,
        row_id_start: i64,
        begin_snapshot: i64,
    ) -> crate::Result<()> {
        self.exec_insert_inlined_batches(
            tx,
            table_name,
            schema,
            batches,
            preserved_row_ids,
            row_id_start,
            begin_snapshot,
        )
        .await
    }

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

    async fn replace_overwritten_delete_file(
        &self,
        tx: &Tx,
        file: &crate::DeleteFileCommit,
        new_snapshot: i64,
    ) -> crate::Result<()> {
        self.exec_replace_overwritten_delete_file(tx, file, new_snapshot)
            .await
    }

    async fn write_inlined_file_deletions(
        &self,
        tx: &Tx,
        table_id: ducklake_catalog::TableId,
        deletions: &[crate::InlinedFileDeletionCommit],
        new_snapshot: i64,
    ) -> crate::Result<()> {
        self.exec_write_inlined_file_deletions(tx, table_id, deletions, new_snapshot)
            .await
    }

    async fn apply_inlined_data_deletions(
        &self,
        tx: &Tx,
        deletions: &[crate::InlinedDataDeletionCommit],
        new_snapshot: i64,
    ) -> crate::Result<i64> {
        self.exec_apply_inlined_data_deletions(tx, deletions, new_snapshot)
            .await
    }

    async fn delete_flushed_inlined_data(
        &self,
        tx: &Tx,
        flushed_tables: &[crate::FlushedInlinedDataCommit],
        flush_snapshot: i64,
    ) -> crate::Result<i64> {
        self.exec_delete_flushed_inlined_data(tx, flushed_tables, flush_snapshot)
            .await
    }

    async fn delete_flushed_inlined_file_deletions(
        &self,
        tx: &Tx,
        table_id: ducklake_catalog::TableId,
        flushed_deletions: &[crate::FlushedInlinedFileDeletionCommit],
    ) -> crate::Result<()> {
        self.exec_delete_flushed_inlined_file_deletions(tx, table_id, flushed_deletions)
            .await
    }
}

#[async_trait]
pub(crate) trait AsyncMaintenanceCommitExecutor<Tx: Sync + ?Sized>:
    crate::commit::AsyncTableDataCommitOps<Tx> + AsyncCatalogQueryCommitExecutor<Tx>
{
    async fn exec_maintenance_execute(
        &self,
        tx: &Tx,
        query: &crate::catalog_queries::CatalogQuery,
    ) -> crate::Result<usize> {
        self.exec_catalog_execute(tx, query).await
    }

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

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

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

    async fn exec_delete_by_ids(
        &self,
        tx: &Tx,
        table: &str,
        column: &str,
        ids: &[i64],
    ) -> crate::Result<()>;

    async fn exec_delete_data_file_metadata(
        &self,
        tx: &Tx,
        source_ids: &[i64],
    ) -> crate::Result<()> {
        for target in crate::metadata_prune_engine::FILE_METADATA_DELETE_TARGETS {
            self.exec_delete_by_ids(tx, target.table, target.column, source_ids)
                .await?;
        }
        Ok(())
    }

    async fn exec_delete_delete_files_by_data_file_id(
        &self,
        tx: &Tx,
        source_ids: &[i64],
    ) -> crate::Result<()> {
        self.exec_delete_by_ids(tx, "ducklake_delete_file", "data_file_id", source_ids)
            .await
    }

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

macro_rules! impl_async_maintenance_commit_executor {
    (
        impl<$($generics:lifetime),+> $ops:ty, $tx:ty {
            cleanup_executor = |$cleanup_self:ident, $cleanup_tx:ident| $cleanup_executor:expr,
            inlined_executor = |$inlined_self:ident, $inlined_tx:ident| $inlined_executor:expr $(,)?
        }
    ) => {
        impl_async_maintenance_commit_executor!(
            @impl [<$($generics),+>] $ops, $tx {
                cleanup_executor = |$cleanup_self, $cleanup_tx| $cleanup_executor,
                inlined_executor = |$inlined_self, $inlined_tx| $inlined_executor,
            }
        );
    };
    (
        impl $ops:ty, $tx:ty {
            cleanup_executor = |$cleanup_self:ident, $cleanup_tx:ident| $cleanup_executor:expr,
            inlined_executor = |$inlined_self:ident, $inlined_tx:ident| $inlined_executor:expr $(,)?
        }
    ) => {
        impl_async_maintenance_commit_executor!(
            @impl [] $ops, $tx {
                cleanup_executor = |$cleanup_self, $cleanup_tx| $cleanup_executor,
                inlined_executor = |$inlined_self, $inlined_tx| $inlined_executor,
            }
        );
    };
    (
        @impl [$($impl_generics:tt)*] $ops:ty, $tx:ty {
            cleanup_executor = |$cleanup_self:ident, $cleanup_tx:ident| $cleanup_executor:expr,
            inlined_executor = |$inlined_self:ident, $inlined_tx:ident| $inlined_executor:expr $(,)?
        }
    ) => {
        #[async_trait]
        impl$($impl_generics)* $crate::catalog_async_commit_methods::AsyncMaintenanceCommitExecutor<$tx>
            for $ops
        {
            async fn exec_source_data_files(
                &self,
                tx: &$tx,
                table_id: ducklake_catalog::TableId,
                snapshot: i64,
                source_ids: &[i64],
            ) -> $crate::Result<Vec<$crate::commit::MaintenanceSourceFile>> {
                let $cleanup_self = self;
                let $cleanup_tx = tx;
                let executor = $cleanup_executor;
                $crate::catalog_async_maintenance_methods::compaction_source_rows(
                    &executor,
                    table_id,
                    snapshot,
                    source_ids,
                )
                .await
            }

            async fn exec_delete_file_cleanup_rows(
                &self,
                tx: &$tx,
                latest_snapshot: i64,
                source_ids: &[i64],
            ) -> $crate::Result<Vec<$crate::CleanupMetadataFile>> {
                let $cleanup_self = self;
                let $cleanup_tx = tx;
                let executor = $cleanup_executor;
                $crate::catalog_async_maintenance_methods::compaction_delete_rows(
                    &executor,
                    latest_snapshot,
                    source_ids,
                )
                .await
            }

            async fn exec_schedule_cleanup_files(
                &self,
                tx: &$tx,
                files: &[$crate::CleanupMetadataFile],
            ) -> $crate::Result<()> {
                let $cleanup_self = self;
                let $cleanup_tx = tx;
                let executor = $cleanup_executor;
                $crate::catalog_async_maintenance_methods::schedule_cleanup_metadata_files(
                    &executor,
                    files,
                )
                .await
            }

            async fn exec_delete_by_ids(
                &self,
                tx: &$tx,
                table: &str,
                column: &str,
                ids: &[i64],
            ) -> $crate::Result<()> {
                let $cleanup_self = self;
                let $cleanup_tx = tx;
                let executor = $cleanup_executor;
                $crate::metadata_prune_engine::delete_by_ids_async(
                    &executor,
                    table,
                    column,
                    ids,
                )
                .await
            }

            async fn exec_delete_inlined_file_deletions_for_files(
                &self,
                tx: &$tx,
                table_id: ducklake_catalog::TableId,
                source_ids: &[i64],
            ) -> $crate::Result<()> {
                let $inlined_self = self;
                let $inlined_tx = tx;
                let executor = $inlined_executor;
                $crate::catalog_async_inlined_mutation_methods::delete_inlined_file_deletions_for_files(
                    &executor,
                    table_id,
                    source_ids,
                )
                .await
            }
        }
    };
}

#[async_trait]
impl<Tx, O> crate::commit::AsyncMaintenanceCommitOps<Tx> for O
where
    Tx: Sync + ?Sized,
    O: AsyncMaintenanceCommitExecutor<Tx> + Sync,
{
    async fn maintenance_execute(
        &self,
        tx: &Tx,
        query: &crate::catalog_queries::CatalogQuery,
    ) -> crate::Result<usize> {
        self.exec_maintenance_execute(tx, query).await
    }

    async fn source_data_files(
        &self,
        tx: &Tx,
        table_id: ducklake_catalog::TableId,
        snapshot: i64,
        source_ids: &[i64],
    ) -> crate::Result<Vec<crate::commit::MaintenanceSourceFile>> {
        self.exec_source_data_files(tx, table_id, snapshot, source_ids)
            .await
    }

    async fn delete_file_cleanup_rows(
        &self,
        tx: &Tx,
        latest_snapshot: i64,
        source_ids: &[i64],
    ) -> crate::Result<Vec<crate::CleanupMetadataFile>> {
        self.exec_delete_file_cleanup_rows(tx, latest_snapshot, source_ids)
            .await
    }

    async fn schedule_cleanup_files(
        &self,
        tx: &Tx,
        files: &[crate::CleanupMetadataFile],
    ) -> crate::Result<()> {
        self.exec_schedule_cleanup_files(tx, files).await
    }

    async fn delete_data_file_metadata(&self, tx: &Tx, source_ids: &[i64]) -> crate::Result<()> {
        self.exec_delete_data_file_metadata(tx, source_ids).await
    }

    async fn delete_delete_files_by_data_file_id(
        &self,
        tx: &Tx,
        source_ids: &[i64],
    ) -> crate::Result<()> {
        self.exec_delete_delete_files_by_data_file_id(tx, source_ids)
            .await
    }

    async fn delete_inlined_file_deletions_for_files(
        &self,
        tx: &Tx,
        table_id: ducklake_catalog::TableId,
        source_ids: &[i64],
    ) -> crate::Result<()> {
        self.exec_delete_inlined_file_deletions_for_files(tx, table_id, source_ids)
            .await
    }
}

macro_rules! impl_async_catalog_object_commit_ops {
    (
        impl<$($generics:lifetime),+> $ops:ty, $tx:ty,
        uuid_style = $uuid_style:expr,
        inlined_executor = |$inlined_self:ident, $inlined_tx:ident| $inlined_executor:expr,
        data_path = |$path_self:ident, $path_tx:ident| $data_path:expr
        $(, extra { $($extra:tt)* })? $(,)?
    ) => {
        impl_async_catalog_object_commit_ops!(
            @impl [<$($generics),+>] $ops, $tx,
            uuid_style = $uuid_style,
            inlined_executor = |$inlined_self, $inlined_tx| $inlined_executor,
            data_path = |$path_self, $path_tx| $data_path,
            extra { $($($extra)*)? }
        );
    };
    (
        impl $ops:ty, $tx:ty,
        uuid_style = $uuid_style:expr,
        inlined_executor = |$inlined_self:ident, $inlined_tx:ident| $inlined_executor:expr,
        data_path = |$path_self:ident, $path_tx:ident| $data_path:expr
        $(, extra { $($extra:tt)* })? $(,)?
    ) => {
        impl_async_catalog_object_commit_ops!(
            @impl [] $ops, $tx,
            uuid_style = $uuid_style,
            inlined_executor = |$inlined_self, $inlined_tx| $inlined_executor,
            data_path = |$path_self, $path_tx| $data_path,
            extra { $($($extra)*)? }
        );
    };
    (
        @impl [$($impl_generics:tt)*] $ops:ty, $tx:ty,
        uuid_style = $uuid_style:expr,
        inlined_executor = |$inlined_self:ident, $inlined_tx:ident| $inlined_executor:expr,
        data_path = |$path_self:ident, $path_tx:ident| $data_path:expr,
        extra { $($extra:tt)* }
    ) => {
        #[async_trait]
        impl$($impl_generics)* $crate::commit::AsyncCatalogObjectCommitOps<$tx> for $ops {
            fn catalog_object_uuid_style(&self) -> $crate::catalog_queries::UuidParamStyle {
                $uuid_style
            }

            async fn catalog_object_execute(
                &self,
                tx: &$tx,
                query: &$crate::catalog_queries::CatalogQuery,
            ) -> $crate::Result<usize> {
                self.exec_catalog_execute(tx, query).await
            }

            async fn catalog_object_close_inlined_data_rows_for_table(
                &self,
                tx: &$tx,
                table_id: ducklake_catalog::TableId,
                new_snapshot: i64,
            ) -> $crate::Result<()> {
                let $inlined_self = self;
                let $inlined_tx = tx;
                let executor = $inlined_executor;
                $crate::catalog_async_inlined_mutation_methods::close_inlined_data_rows_for_table(
                    &executor,
                    table_id,
                    new_snapshot,
                )
                .await
            }

            async fn catalog_object_query_schemas(
                &self,
                tx: &$tx,
                query: &$crate::catalog_queries::CatalogQuery,
            ) -> $crate::Result<Vec<ducklake_catalog::SchemaEntry>> {
                self.exec_catalog_query(tx, query, $crate::embedded_reads::schema_entry)
                    .await
            }

            async fn catalog_object_query_tables(
                &self,
                tx: &$tx,
                query: &$crate::catalog_queries::CatalogQuery,
            ) -> $crate::Result<Vec<ducklake_catalog::TableEntry>> {
                self.exec_catalog_query(tx, query, $crate::embedded_reads::table_entry)
                    .await
            }

            async fn catalog_object_query_views(
                &self,
                tx: &$tx,
                query: &$crate::catalog_queries::CatalogQuery,
            ) -> $crate::Result<Vec<ducklake_catalog::ViewEntry>> {
                self.exec_catalog_query(tx, query, $crate::embedded_reads::view_entry)
                    .await
            }

            async fn catalog_object_query_i64s(
                &self,
                tx: &$tx,
                query: &$crate::catalog_queries::CatalogQuery,
            ) -> $crate::Result<Vec<i64>> {
                self.exec_catalog_query(tx, query, $crate::catalog_queries::i64_value)
                    .await
            }

            async fn data_path(&self, tx: &$tx) -> $crate::Result<String> {
                let $path_self = self;
                let $path_tx = tx;
                $data_path.await
            }

            $($extra)*
        }
    };
}

macro_rules! impl_async_partitioning_commit_ops {
    (impl<$($generics:lifetime),+> $ops:ty, $tx:ty $(, extra { $($extra:tt)* })?) => {
        impl_async_partitioning_commit_ops!(
            @impl [<$($generics),+>] $ops, $tx, extra { $($($extra)*)? }
        );
    };
    (impl $ops:ty, $tx:ty $(, extra { $($extra:tt)* })?) => {
        impl_async_partitioning_commit_ops!(
            @impl [] $ops, $tx, extra { $($($extra)*)? }
        );
    };
    (@impl [$($impl_generics:tt)*] $ops:ty, $tx:ty, extra { $($extra:tt)* }) => {
        #[async_trait]
        impl$($impl_generics)* $crate::commit::AsyncPartitioningCommitOps<$tx> for $ops {
            async fn partitioning_execute(
                &self,
                tx: &$tx,
                query: &$crate::catalog_queries::CatalogQuery,
            ) -> $crate::Result<usize> {
                self.exec_catalog_execute(tx, query).await
            }

            async fn partitioning_query_column_parent_refs(
                &self,
                tx: &$tx,
                query: &$crate::catalog_queries::CatalogQuery,
            ) -> $crate::Result<Vec<(ducklake_catalog::ColumnId, Option<i64>)>> {
                self.exec_catalog_query(tx, query, $crate::catalog_queries::column_parent_ref)
                    .await
            }

            async fn partitioning_query_infos(
                &self,
                tx: &$tx,
                query: &$crate::catalog_queries::CatalogQuery,
            ) -> $crate::Result<Vec<ducklake_catalog::PartitionInfo>> {
                self.exec_catalog_query(tx, query, $crate::embedded_reads::partition_info)
                    .await
            }

            async fn partitioning_query_columns(
                &self,
                tx: &$tx,
                query: &$crate::catalog_queries::CatalogQuery,
            ) -> $crate::Result<Vec<ducklake_catalog::PartitionColumn>> {
                self.exec_catalog_query(tx, query, $crate::embedded_reads::partition_column)
                    .await
            }

            $($extra)*
        }
    };
}

macro_rules! impl_async_sorted_by_commit_ops {
    (impl<$($generics:lifetime),+> $ops:ty, $tx:ty $(, extra { $($extra:tt)* })?) => {
        impl_async_sorted_by_commit_ops!(
            @impl [<$($generics),+>] $ops, $tx, extra { $($($extra)*)? }
        );
    };
    (impl $ops:ty, $tx:ty $(, extra { $($extra:tt)* })?) => {
        impl_async_sorted_by_commit_ops!(
            @impl [] $ops, $tx, extra { $($($extra)*)? }
        );
    };
    (@impl [$($impl_generics:tt)*] $ops:ty, $tx:ty, extra { $($extra:tt)* }) => {
        #[async_trait]
        impl$($impl_generics)* $crate::commit::AsyncSortedByCommitOps<$tx> for $ops {
            async fn sorted_by_execute(
                &self,
                tx: &$tx,
                query: &$crate::catalog_queries::CatalogQuery,
            ) -> $crate::Result<usize> {
                self.exec_catalog_execute(tx, query).await
            }

            async fn sorted_by_query_column_names(
                &self,
                tx: &$tx,
                query: &$crate::catalog_queries::CatalogQuery,
            ) -> $crate::Result<Vec<String>> {
                self.exec_catalog_query(tx, query, $crate::catalog_queries::string_value)
                    .await
            }

            async fn sorted_by_query_infos(
                &self,
                tx: &$tx,
                query: &$crate::catalog_queries::CatalogQuery,
            ) -> $crate::Result<Vec<ducklake_catalog::SortInfo>> {
                self.exec_catalog_query(tx, query, $crate::embedded_reads::sort_info)
                    .await
            }

            async fn sorted_by_query_expressions(
                &self,
                tx: &$tx,
                query: &$crate::catalog_queries::CatalogQuery,
            ) -> $crate::Result<Vec<ducklake_catalog::SortExpression>> {
                self.exec_catalog_query(tx, query, $crate::embedded_reads::sort_expression)
                    .await
            }

            $($extra)*
        }
    };
}

macro_rules! impl_async_column_evolution_commit_ops {
    (@impl [$($impl_generics:tt)*] $ops:ty, $tx:ty, extra { $($extra:tt)* }) => {
        #[async_trait]
        impl$($impl_generics)* $crate::commit::AsyncColumnEvolutionCommitOps<$tx> for $ops {
            async fn column_execute(
                &self,
                tx: &$tx,
                query: &$crate::catalog_queries::CatalogQuery,
            ) -> $crate::Result<usize> {
                self.exec_catalog_execute(tx, query).await
            }

            async fn column_query_i64s(
                &self,
                tx: &$tx,
                query: &$crate::catalog_queries::CatalogQuery,
            ) -> $crate::Result<Vec<i64>> {
                self.exec_catalog_query(tx, query, $crate::catalog_queries::i64_value)
                    .await
            }

            async fn column_query_strings(
                &self,
                tx: &$tx,
                query: &$crate::catalog_queries::CatalogQuery,
            ) -> $crate::Result<Vec<String>> {
                self.exec_catalog_query(tx, query, $crate::catalog_queries::string_value)
                    .await
            }

            async fn column_query_opt_bools(
                &self,
                tx: &$tx,
                query: &$crate::catalog_queries::CatalogQuery,
            ) -> $crate::Result<Vec<Option<bool>>> {
                self.exec_catalog_query(tx, query, $crate::catalog_queries::opt_bool_value)
                    .await
            }

            async fn column_query_active_refs(
                &self,
                tx: &$tx,
                query: &$crate::catalog_queries::CatalogQuery,
            ) -> $crate::Result<Vec<(
                ducklake_catalog::ColumnId,
                String,
                Option<ducklake_catalog::ColumnId>,
            )>> {
                self.exec_catalog_query(tx, query, $crate::catalog_queries::active_column_ref)
                    .await
            }

            async fn visible_column(
                &self,
                tx: &$tx,
                table_id: ducklake_catalog::TableId,
                column_id: ducklake_catalog::ColumnId,
                snapshot: ducklake_catalog::SnapshotId,
            ) -> $crate::Result<Option<ducklake_catalog::ColumnEntry>> {
                Ok(self
                    .exec_catalog_query(
                        tx,
                        &$crate::catalog_queries::visible_column(table_id, column_id, snapshot),
                        $crate::embedded_reads::column_entry,
                    )
                    .await?
                    .into_iter()
                    .next())
            }

            async fn visible_columns(
                &self,
                tx: &$tx,
                table_id: ducklake_catalog::TableId,
                snapshot: ducklake_catalog::SnapshotId,
            ) -> $crate::Result<Vec<ducklake_catalog::ColumnEntry>> {
                self.exec_catalog_query(
                    tx,
                    &$crate::catalog_queries::visible_columns(table_id, snapshot),
                    $crate::embedded_reads::column_entry,
                )
                .await
            }

            async fn rewrite_sort_expressions_for_rename(
                &self,
                tx: &$tx,
                table_id: ducklake_catalog::TableId,
                latest_snapshot: ducklake_catalog::SnapshotId,
                new_snapshot: i64,
                next_catalog_id: i64,
                old_column_name: &str,
                new_column_name: &str,
            ) -> $crate::Result<i64> {
                $crate::sort_rewrite_engine::rewrite_sort_expressions_for_rename_async(
                    $crate::sort_rewrite_engine::SortRenameRewrite {
                        table_id,
                        latest_snapshot,
                        new_snapshot,
                        next_catalog_id,
                        old_column_name,
                        new_column_name,
                    },
                    |table_name| async move {
                        self.exec_catalog_table_exists(tx, &table_name).await
                    },
                    |query| async move {
                        self.exec_catalog_query(
                            tx,
                            &query,
                            $crate::embedded_reads::sort_info,
                        )
                        .await
                    },
                    |query| async move {
                        self.exec_catalog_query(
                            tx,
                            &query,
                            $crate::embedded_reads::sort_expression,
                        )
                        .await
                    },
                    |sort_id, table_id, new_snapshot| async move {
                        self.exec_catalog_execute(
                            tx,
                            &$crate::catalog_queries::close_sort(
                                sort_id,
                                table_id,
                                new_snapshot,
                            ),
                        )
                        .await
                    },
                    |query| async move {
                        self.exec_catalog_execute(tx, &query).await?;
                        Ok(())
                    },
                )
                .await
            }

            $($extra)*
        }
    };
    (impl<$($generics:lifetime),+> $ops:ty, $tx:ty, extra { $($extra:tt)* }) => {
        impl_async_column_evolution_commit_ops! {
            @impl [<$($generics),+>] $ops, $tx, extra { $($extra)* }
        }
    };
    (impl $ops:ty, $tx:ty, extra { $($extra:tt)* }) => {
        impl_async_column_evolution_commit_ops! {
            @impl [] $ops, $tx, extra { $($extra)* }
        }
    };
}
