macro_rules! impl_embedded_sync_commit_ops {
    (
        $ops:ty,
        tx_type = $tx_type:ty,
        tx = $tx:ty,
        table_exists = $table_exists:path
    ) => {
        impl $crate::commit::SyncCommitOps for $ops {
            type Tx<'tx> = $tx_type;

            fn latest_snapshot_ids(
                &self,
                tx: &$tx,
            ) -> $crate::Result<$crate::commit::CommitSnapshotIds> {
                run_catalog_query(
                    tx,
                    &$crate::catalog_queries::latest_snapshot_ids(),
                    $crate::catalog_queries::latest_snapshot_ids_value,
                )?
                .into_iter()
                .next()
                .ok_or_else(|| {
                    $crate::StorageError::Backend("DuckLake commit found no snapshots".into())
                })
            }

            fn table_exists(&self, tx: &$tx, table_name: &str) -> $crate::Result<bool> {
                $table_exists(tx, table_name)
            }

            fn insert_snapshot(
                &self,
                tx: &$tx,
                snapshot_id: i64,
                schema_version: i64,
                next_catalog_id: i64,
                next_file_id: i64,
            ) -> $crate::Result<()> {
                let now_micros = $crate::sql_utils::current_time_micros()?;
                let now_timestamp = $crate::embedded_rows::format_timestamp_micros(now_micros)?;
                execute_catalog_command(
                    tx,
                    &$crate::catalog_queries::insert_snapshot_with_timestamp_text(
                        snapshot_id,
                        now_timestamp,
                        schema_version,
                        next_catalog_id,
                        next_file_id,
                    ),
                )?;
                Ok(())
            }

            fn insert_snapshot_change(
                &self,
                tx: &$tx,
                snapshot_id: i64,
                changes_made: &str,
                commit_message: &str,
            ) -> $crate::Result<()> {
                execute_catalog_command(
                    tx,
                    &$crate::catalog_queries::insert_snapshot_change(
                        snapshot_id,
                        changes_made,
                        commit_message,
                    ),
                )?;
                Ok(())
            }
        }
    };
}

macro_rules! impl_embedded_sync_catalog_object_commit_ops {
    (
        $ops:ty,
        $tx:ty,
        uuid_style = $uuid_style:expr,
        data_path = $data_path:path,
        close_inlined = $close_inlined:path
    ) => {
        impl $crate::commit::SyncCatalogObjectCommitOps for $ops {
            fn catalog_object_uuid_style(&self) -> $crate::catalog_queries::UuidParamStyle {
                $uuid_style
            }

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

            fn catalog_object_close_inlined_data_rows_for_table(
                &self,
                tx: &$tx,
                table_id: ducklake_catalog::TableId,
                new_snapshot: i64,
            ) -> $crate::Result<()> {
                $close_inlined(tx, table_id, new_snapshot)
            }

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

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

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

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

            fn data_path(&self, tx: &$tx) -> $crate::Result<String> {
                $data_path(tx)
            }
        }
    };
}

macro_rules! impl_embedded_sync_table_data_commit_ops {
    (
        $ops:ty,
        $tx:ty,
        dialect = $dialect:expr,
        next_row_id_for_append = $next_row_id_for_append:path,
        insert_inlined_batches = $insert_inlined_batches:path,
        overwritten_cleanup_data_path = |$cleanup_tx:ident| $cleanup_data_path:expr,
        write_inlined_file_deletions = $write_inlined_file_deletions:path,
        apply_inlined_data_deletions = $apply_inlined_data_deletions:path,
        delete_flushed_inlined_data = $delete_flushed_inlined_data:path,
        delete_flushed_inlined_file_deletions = $delete_flushed_inlined_file_deletions:path
        $(, insert_table_stats_from_active_files = $insert_table_stats_from_active_files:path)?
    ) => {
        impl $crate::commit::SyncTableDataCommitOps for $ops {
            fn table_data_execute(
                &self,
                tx: &$tx,
                query: &$crate::catalog_queries::CatalogQuery,
            ) -> $crate::Result<usize> {
                execute_catalog_command(tx, query)
            }

            fn next_row_id_for_append(
                &self,
                tx: &$tx,
                table_id: ducklake_catalog::TableId,
                has_table_stats: bool,
            ) -> $crate::Result<Option<i64>> {
                $next_row_id_for_append(
                    |query| {
                        Ok(run_catalog_query(
                            tx,
                            &query,
                            $crate::catalog_queries::opt_i64_value,
                        )?
                        .into_iter()
                        .next()
                        .flatten())
                    },
                    table_id,
                    has_table_stats,
                )
            }

            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<()> {
                $crate::stats_engine::update_table_column_stats_for_append_sync(
                    || {
                        <$ops as $crate::commit::SyncCommitOps>::table_exists(
                            self,
                            tx,
                            "ducklake_table_column_stats",
                        )
                    },
                    |query| {
                        run_catalog_query(tx, &query, $crate::catalog_queries::column_type_value)
                    },
                    |query| {
                        run_catalog_query(
                            tx,
                            &query,
                            $crate::embedded_reads::table_column_stats_row,
                        )
                    },
                    |stats| {
                        $crate::stats_engine::upsert_table_column_stats_by_update_sync(
                            stats,
                            |query| execute_catalog_command(tx, &query),
                        )
                    },
                    table_id,
                    latest_snapshot,
                    files,
                )
            }

            $(
                fn insert_table_stats_from_active_files(
                    &self,
                    tx: &$tx,
                    table_id: ducklake_catalog::TableId,
                ) -> $crate::Result<()> {
                    $insert_table_stats_from_active_files(tx, table_id)
                }
            )?

            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,
                    $dialect,
                )
            }

            fn ensure_inlined_data_table(
                &self,
                tx: &$tx,
                table_id: ducklake_catalog::TableId,
                schema_version: i64,
                schema: &ducklake_catalog::TableSchema,
            ) -> $crate::Result<String> {
                $crate::inlined_values::ensure_inlined_data_table_sync(
                    table_id,
                    schema_version,
                    schema,
                    $dialect,
                    $crate::embedded_reads::quote_identifier,
                    $crate::embedded_reads::quote_identifier,
                    |table_id, schema_version| {
                        Ok(run_catalog_query(
                            tx,
                            &$crate::catalog_queries::inlined_data_table_name(
                                table_id,
                                schema_version,
                            ),
                            $crate::catalog_queries::string_value,
                        )?
                        .into_iter()
                        .next())
                    },
                    |table_id, table_name, schema_version| {
                        execute_catalog_command(
                            tx,
                            &$crate::catalog_queries::insert_inlined_data_table_ref(
                                table_id,
                                table_name,
                                schema_version,
                            ),
                        )?;
                        Ok(())
                    },
                    |create_sql| {
                        tx.execute(create_sql, []).map_err(sql_err)?;
                        Ok(())
                    },
                )
            }

            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<()> {
                $insert_inlined_batches(
                    tx,
                    table_name,
                    schema,
                    batches,
                    preserved_row_ids,
                    row_id_start,
                    begin_snapshot,
                )
            }

            fn replace_overwritten_delete_file(
                &self,
                tx: &$tx,
                file: &$crate::DeleteFileCommit,
                new_snapshot: i64,
            ) -> $crate::Result<()> {
                $crate::embedded_maintenance::replace_overwritten_delete_file(
                    &**tx,
                    file,
                    new_snapshot,
                    |overwritten, table_id| {
                        let $cleanup_tx = tx;
                        let data_path = $cleanup_data_path?;
                        $crate::embedded_maintenance::load_overwritten_cleanup_file_sync(
                            overwritten,
                            table_id,
                            &data_path,
                            |query, mapper| run_catalog_query(tx, query, mapper),
                        )
                    },
                )
            }

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

            fn apply_inlined_data_deletions(
                &self,
                tx: &$tx,
                deletions: &[$crate::InlinedDataDeletionCommit],
                new_snapshot: i64,
            ) -> $crate::Result<i64> {
                $apply_inlined_data_deletions(tx, deletions, new_snapshot)
            }

            fn delete_flushed_inlined_data(
                &self,
                tx: &$tx,
                flushed_tables: &[$crate::FlushedInlinedDataCommit],
                flush_snapshot: i64,
            ) -> $crate::Result<i64> {
                $delete_flushed_inlined_data(tx, flushed_tables, flush_snapshot)
            }

            fn delete_flushed_inlined_file_deletions(
                &self,
                tx: &$tx,
                table_id: ducklake_catalog::TableId,
                flushed_deletions: &[$crate::FlushedInlinedFileDeletionCommit],
            ) -> $crate::Result<()> {
                $delete_flushed_inlined_file_deletions(tx, table_id, flushed_deletions)
            }
        }
    };
}

macro_rules! impl_embedded_sync_maintenance_commit_ops {
    (
        $ops:ty,
        $tx:ty
        $(, recompute_table_column_stats_from_active_files = $recompute_table_column_stats_from_active_files:path)?
    ) => {
        impl $crate::commit::SyncMaintenanceCommitOps for $ops {
            fn maintenance_execute(
                &self,
                tx: &$tx,
                query: &$crate::catalog_queries::CatalogQuery,
            ) -> $crate::Result<usize> {
                execute_catalog_command(tx, query)
            }

            fn source_data_files(
                &self,
                tx: &$tx,
                table_id: ducklake_catalog::TableId,
                snapshot: i64,
                source_ids: &[i64],
            ) -> $crate::Result<Vec<$crate::commit::MaintenanceSourceFile>> {
                $crate::embedded_maintenance::compaction_source_rows(
                    &**tx, table_id, snapshot, source_ids,
                )
            }

            fn delete_file_cleanup_rows(
                &self,
                tx: &$tx,
                latest_snapshot: i64,
                source_ids: &[i64],
            ) -> $crate::Result<Vec<$crate::CleanupMetadataFile>> {
                $crate::embedded_maintenance::compaction_delete_rows(
                    &**tx,
                    latest_snapshot,
                    source_ids,
                )
            }

            fn schedule_cleanup_files(
                &self,
                tx: &$tx,
                files: &[$crate::CleanupMetadataFile],
            ) -> $crate::Result<()> {
                $crate::embedded_maintenance::schedule_cleanup_files(&**tx, files)
            }

            fn delete_data_file_metadata(
                &self,
                tx: &$tx,
                source_ids: &[i64],
            ) -> $crate::Result<()> {
                $crate::embedded_maintenance::delete_file_metadata(&**tx, source_ids)
            }

            fn delete_delete_files_by_data_file_id(
                &self,
                tx: &$tx,
                source_ids: &[i64],
            ) -> $crate::Result<()> {
                $crate::embedded_maintenance::delete_by_ids(
                    &**tx,
                    "ducklake_delete_file",
                    "data_file_id",
                    source_ids,
                )
            }

            fn delete_inlined_file_deletions_for_files(
                &self,
                tx: &$tx,
                table_id: ducklake_catalog::TableId,
                source_ids: &[i64],
            ) -> $crate::Result<()> {
                $crate::embedded_maintenance::delete_inlined_file_deletions(
                    &**tx, table_id, source_ids,
                )
            }

            $(
                fn recompute_table_column_stats_from_active_files(
                    &self,
                    tx: &$tx,
                    table_id: ducklake_catalog::TableId,
                    snapshot: ducklake_catalog::SnapshotId,
                ) -> $crate::Result<()> {
                    $recompute_table_column_stats_from_active_files(tx, table_id, snapshot)
                }
            )?
        }
    };
}

macro_rules! impl_embedded_sync_partitioning_commit_ops {
    ($ops:ty, $tx:ty) => {
        impl $crate::commit::SyncPartitioningCommitOps for $ops {
            fn partitioning_execute(
                &self,
                tx: &$tx,
                query: &$crate::catalog_queries::CatalogQuery,
            ) -> $crate::Result<usize> {
                execute_catalog_command(tx, query)
            }

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

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

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

macro_rules! impl_embedded_sync_sorted_by_commit_ops {
    ($ops:ty, $tx:ty) => {
        impl $crate::commit::SyncSortedByCommitOps for $ops {
            fn sorted_by_execute(
                &self,
                tx: &$tx,
                query: &$crate::catalog_queries::CatalogQuery,
            ) -> $crate::Result<usize> {
                execute_catalog_command(tx, query)
            }

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

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

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

macro_rules! impl_embedded_sync_column_evolution_commit_ops {
    ($ops:ty, $tx:ty, $rewrite_sort_expressions_for_rename:path) => {
        impl $crate::commit::SyncColumnEvolutionCommitOps for $ops {
            fn column_execute(
                &self,
                tx: &$tx,
                query: &$crate::catalog_queries::CatalogQuery,
            ) -> $crate::Result<usize> {
                execute_catalog_command(tx, query)
            }

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

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

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

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

            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(run_catalog_query(
                    tx,
                    &$crate::catalog_queries::visible_column(table_id, column_id, snapshot),
                    $crate::embedded_reads::column_entry,
                )?
                .into_iter()
                .next())
            }

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

            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> {
                $rewrite_sort_expressions_for_rename(
                    tx,
                    table_id,
                    latest_snapshot.0,
                    new_snapshot,
                    next_catalog_id,
                    old_column_name,
                    new_column_name,
                )
            }
        }
    };
}

macro_rules! impl_embedded_sync_metadata_commit_ops {
    (
        $ops:ty,
        $tx:ty,
        rewrite_sort_expressions_for_rename = $rewrite_sort_expressions_for_rename:path
    ) => {
        impl_embedded_sync_partitioning_commit_ops!($ops, $tx);
        impl_embedded_sync_sorted_by_commit_ops!($ops, $tx);
        impl_embedded_sync_column_evolution_commit_ops!(
            $ops,
            $tx,
            $rewrite_sort_expressions_for_rename
        );
    };
}

macro_rules! impl_embedded_inlined_mutation_helpers {
    (
        tx = $tx:ty,
        table_exists = $table_exists:path,
        apply_inlined_data_deletions = $apply_inlined_data_deletions:ident,
        delete_flushed_inlined_data = $delete_flushed_inlined_data:ident,
        delete_flushed_inlined_file_deletions = $delete_flushed_inlined_file_deletions:ident,
        write_inlined_file_deletions = $write_inlined_file_deletions:ident,
        close_inlined_data_rows_for_table = $close_inlined_data_rows_for_table:ident $(,)?
    ) => {
        fn $apply_inlined_data_deletions(
            tx: &$tx,
            deletions: &[$crate::InlinedDataDeletionCommit],
            new_snapshot: i64,
        ) -> $crate::Result<i64> {
            $crate::inlined_values::apply_inlined_data_deletions_row_by_row_sync(
                deletions,
                new_snapshot,
                $crate::embedded_reads::quote_identifier,
                "?1",
                "?2",
                "!=",
                |sql, row_ids, new_snapshot| {
                    let mut stmt = tx.prepare(&sql).map_err(sql_err)?;
                    row_ids
                        .iter()
                        .map(|row_id| {
                            let changed = stmt
                                .execute(params![new_snapshot, row_id])
                                .map_err(sql_err)?;
                            Ok((*row_id, changed))
                        })
                        .collect()
                },
            )
        }

        fn $delete_flushed_inlined_data(
            tx: &$tx,
            flushed_tables: &[$crate::FlushedInlinedDataCommit],
            flush_snapshot: i64,
        ) -> $crate::Result<i64> {
            $crate::inlined_values::delete_flushed_inlined_data_row_by_row_sync(
                flushed_tables,
                flush_snapshot,
                $crate::embedded_reads::quote_identifier,
                "?1",
                "?2",
                |sql, row_ids, flush_snapshot| {
                    let mut stmt = tx.prepare(&sql).map_err(sql_err)?;
                    let mut deleted = 0_usize;
                    for row_id in row_ids {
                        let changed = stmt
                            .execute(params![flush_snapshot, row_id])
                            .map_err(sql_err)?;
                        deleted = deleted.checked_add(changed).ok_or_else(|| {
                            $crate::StorageError::Backend(
                                "DuckLake flushed inlined-data row count overflow".into(),
                            )
                        })?;
                    }
                    Ok(deleted)
                },
            )
        }

        fn $delete_flushed_inlined_file_deletions(
            tx: &$tx,
            table_id: ducklake_catalog::TableId,
            flushed_deletions: &[$crate::FlushedInlinedFileDeletionCommit],
        ) -> $crate::Result<()> {
            $crate::inlined_values::delete_flushed_inlined_file_deletions_sync(
                table_id,
                flushed_deletions,
                |table_name| $table_exists(tx, table_name),
                |table_name, row| {
                    let quoted_table = $crate::embedded_reads::quote_identifier(table_name);
                    let sql = $crate::inlined_values::delete_flushed_inlined_file_deletion_sql(
                        &quoted_table,
                        "?1",
                        "?2",
                        "?3",
                    );
                    tx.execute(
                        &sql,
                        params![row.data_file_id.0, row.row_id, row.begin_snapshot.0],
                    )
                    .map_err(sql_err)
                    .map_err(Into::into)
                },
            )
        }

        fn $write_inlined_file_deletions(
            tx: &$tx,
            table_id: ducklake_catalog::TableId,
            deletions: &[$crate::InlinedFileDeletionCommit],
            new_snapshot: i64,
        ) -> $crate::Result<()> {
            $crate::inlined_values::write_inlined_file_deletions_sync(
                table_id,
                deletions,
                new_snapshot,
                $crate::embedded_reads::quote_identifier,
                "?1, ?2, ?3",
                |create_sql| {
                    tx.execute(create_sql, []).map_err(sql_err)?;
                    Ok(())
                },
                |insert_sql, rows, new_snapshot| {
                    let mut stmt = tx.prepare(insert_sql).map_err(sql_err)?;
                    for row in rows {
                        stmt.execute(params![row.data_file_id, row.row_id, new_snapshot])
                            .map_err(sql_err)?;
                    }
                    Ok(())
                },
            )
        }

        fn $close_inlined_data_rows_for_table(
            tx: &$tx,
            table_id: ducklake_catalog::TableId,
            new_snapshot: i64,
        ) -> $crate::Result<()> {
            $crate::inlined_values::close_inlined_data_rows_for_table_sync(
                table_id,
                new_snapshot,
                |query| run_catalog_query(tx, query, $crate::catalog_queries::string_value),
                |table_name| $table_exists(tx, table_name),
                |table_name, new_snapshot| {
                    tx.execute(
                        &$crate::inlined_values::close_all_inlined_data_rows_sql(
                            &$crate::embedded_reads::quote_identifier(table_name),
                            "?1",
                        ),
                        params![new_snapshot],
                    )
                    .map_err(sql_err)?;
                    Ok(())
                },
            )
        }
    };
}

macro_rules! impl_embedded_insert_inlined_batches_helper {
    (
        fn $name:ident,
        tx = $tx:ty,
        params_from_iter = $params_from_iter:path,
        metadata_i64_value = |$metadata_i64_value:ident| $metadata_i64_expr:expr,
        metadata_null_value = $metadata_null_expr:expr,
        placeholder = |$placeholder_idx:ident, $placeholder_field:ident| $placeholder:expr,
        value = $value:path $(,)?
    ) => {
        fn $name(
            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 placeholders = $crate::inlined_values::inlined_insert_placeholders(
                schema,
                vec!["?".to_string(); 3],
                |$placeholder_idx, $placeholder_field| $placeholder,
            )?;
            let sql = $crate::inlined_values::inlined_insert_sql(
                &$crate::embedded_reads::quote_identifier(table_name),
                &placeholders,
            );
            let mut stmt = tx.prepare(&sql).map_err(sql_err)?;
            $crate::inlined_values::write_inlined_batch_rows_sync(
                schema,
                batches,
                preserved_row_ids,
                row_id_start,
                begin_snapshot,
                |row_id, begin_snapshot| {
                    $crate::inlined_values::inlined_insert_metadata_values(
                        row_id,
                        begin_snapshot,
                        |$metadata_i64_value| $metadata_i64_expr,
                        || $metadata_null_expr,
                    )
                },
                $value,
                |values| {
                    stmt.execute($params_from_iter(values)).map_err(sql_err)?;
                    Ok(())
                },
            )
        }
    };
}

macro_rules! impl_embedded_sort_rename_rewrite_helper {
    (
        fn $name:ident,
        tx = $tx:ty,
        table_exists = $table_exists:path $(,)?
    ) => {
        fn $name(
            tx: &$tx,
            table_id: ducklake_catalog::TableId,
            latest_snapshot: i64,
            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_sync(
                $crate::sort_rewrite_engine::SortRenameRewrite {
                    table_id,
                    latest_snapshot: ducklake_catalog::SnapshotId(latest_snapshot),
                    new_snapshot,
                    next_catalog_id,
                    old_column_name,
                    new_column_name,
                },
                |table_name| $table_exists(tx, table_name),
                |query| run_catalog_query(tx, &query, $crate::embedded_reads::sort_info),
                |query| run_catalog_query(tx, &query, $crate::embedded_reads::sort_expression),
                |sort_id, table_id, new_snapshot| {
                    execute_catalog_command(
                        tx,
                        &$crate::catalog_queries::close_sort(sort_id, table_id, new_snapshot),
                    )
                },
                |query| {
                    execute_catalog_command(tx, &query)?;
                    Ok(())
                },
            )
        }
    };
}
