macro_rules! catalog_async_method {
    (|$this:ident| fn $name:ident (&self $(, $arg:ident : $arg_ty:ty)* $(,)?) -> $ret:ty $body:block) => {
        fn $name<'life0, 'async_trait>(
            &'life0 self,
            $($arg: $arg_ty),*
        ) -> std::pin::Pin<
            Box<dyn std::future::Future<Output = $ret> + Send + 'async_trait>,
        >
        where
            'life0: 'async_trait,
            Self: Sync + 'async_trait,
        {
            let $this = self;
            Box::pin(async move $body)
        }
    };
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
macro_rules! impl_embedded_async_read_query_executor {
    (impl $ty:ty) => {
        #[async_trait::async_trait]
        impl $crate::catalog_async_query_methods::AsyncReadQueryExecutor for $ty {
            fn read_query_render(&self) -> $crate::catalog_async_query_methods::ReadQueryRender {
                $crate::catalog_async_query_methods::ReadQueryRender::Embedded
            }

            async fn execute_rendered_read_query<T>(
                &self,
                sql: String,
                params: Vec<i64>,
                mapper: for<'row> fn(
                    &'row dyn $crate::embedded_rows::EmbeddedRow,
                ) -> $crate::Result<T>,
            ) -> $crate::Result<Vec<T>>
            where
                T: Send + 'static,
            {
                self.with_conn(move |conn| {
                    let mut stmt = conn.prepare(&sql).map_err(sql_err)?;
                    collect_shared_rows(&mut stmt, params_from_iter(params), mapper)
                })
                .await
            }
        }
    };
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
macro_rules! impl_embedded_inlined_read_methods {
    (file_deletion_table_exists_sql = $file_deletion_table_exists_sql:expr) => {
        catalog_async_method! { |this|
            fn inlined_file_deletions_at(
                &self,
                table: ducklake_catalog::TableId,
                snapshot: ducklake_catalog::SnapshotId,
            ) -> $crate::Result<Vec<ducklake_catalog::InlinedFileDeletion>> {
                this.with_conn(move |conn| {
                    let table_name = $crate::inlined_values::inlined_file_deletion_table_name(table);
                    let exists = conn
                        .query_row(
                            $file_deletion_table_exists_sql,
                            params![table_name],
                            |_| Ok(()),
                        )
                        .optional()
                        .map_err(sql_err)?
                        .is_some();
                    if !exists {
                        return Ok(Vec::new());
                    }

                    run_read_query(
                        &conn,
                        &$crate::embedded_reads::inlined_file_deletions(&table_name, snapshot),
                    )
                })
                .await
            }
        }

        fn inlined_data<'life0, 'life1, 'life2, 'async_trait>(
            &'life0 self,
            table: &'life1 ducklake_catalog::InlinedDataTableRef,
            schema: &'life2 ducklake_catalog::TableSchema,
            snapshot: ducklake_catalog::SnapshotId,
        ) -> std::pin::Pin<
            Box<
                dyn std::future::Future<
                        Output = $crate::Result<Vec<arrow::record_batch::RecordBatch>>,
                    > + Send
                    + 'async_trait,
            >,
        >
        where
            'life0: 'async_trait,
            'life1: 'async_trait,
            'life2: 'async_trait,
            Self: Sync + 'async_trait,
        {
            let this = self;
            Box::pin(async move {
                let table_name = table.table_name.clone();
                let schema = schema.clone();
                this.with_conn(move |conn| read_inlined_data(&conn, &table_name, &schema, snapshot))
                    .await
            })
        }

        fn inlined_data_with_metadata<'life0, 'life1, 'life2, 'async_trait>(
            &'life0 self,
            table: &'life1 ducklake_catalog::InlinedDataTableRef,
            schema: &'life2 ducklake_catalog::TableSchema,
            snapshot: ducklake_catalog::SnapshotId,
        ) -> std::pin::Pin<
            Box<
                dyn std::future::Future<Output = $crate::Result<Vec<$crate::InlinedDataBatch>>>
                    + Send
                    + 'async_trait,
            >,
        >
        where
            'life0: 'async_trait,
            'life1: 'async_trait,
            'life2: 'async_trait,
            Self: Sync + 'async_trait,
        {
            let this = self;
            Box::pin(async move {
                let table_name = table.table_name.clone();
                let schema = schema.clone();
                this.with_conn(move |conn| {
                    read_inlined_data_with_metadata(&conn, &table_name, &schema, snapshot)
                })
                .await
            })
        }
    };
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
macro_rules! impl_embedded_inlined_data_reader {
    (
        dialect = $dialect:expr,
        projection = |$field:ident| $projection:expr,
        decoder = $decoder:expr $(,)?
    ) => {
        fn read_inlined_data(
            conn: &Connection,
            table_name: &str,
            schema: &ducklake_catalog::TableSchema,
            snapshot: ducklake_catalog::SnapshotId,
        ) -> $crate::Result<Vec<arrow::record_batch::RecordBatch>> {
            let projection =
                $crate::inlined_values::inlined_read_projection(schema, |$field| $projection)?;
            let table = $crate::embedded_reads::quote_identifier(table_name);
            let sql = $crate::inlined_values::read_inlined_data_sql(&table, &projection, "?1");

            $crate::inlined_values::read_inlined_record_batches(schema, $dialect, |columns| {
                let mut stmt = conn.prepare(&sql).map_err(sql_err)?;
                let mut rows = stmt.query(params![snapshot.0]).map_err(sql_err)?;
                while let Some(row) = rows.next().map_err(sql_err)? {
                    for (idx, column) in columns.iter_mut().enumerate() {
                        $crate::inlined_values::push_decoded_inlined_column(
                            column,
                            row.get_ref(idx).map_err(sql_err)?,
                            &$decoder,
                        )?;
                    }
                }
                Ok(())
            })
        }

        fn read_inlined_data_with_metadata(
            conn: &Connection,
            table_name: &str,
            schema: &ducklake_catalog::TableSchema,
            snapshot: ducklake_catalog::SnapshotId,
        ) -> $crate::Result<Vec<$crate::InlinedDataBatch>> {
            let projection =
                $crate::inlined_values::inlined_read_projection(schema, |$field| $projection)?;
            let table = $crate::embedded_reads::quote_identifier(table_name);
            let sql = $crate::inlined_values::read_inlined_data_with_metadata_sql(
                &table,
                &projection,
                "?1",
            );

            $crate::inlined_values::read_inlined_data_batches(
                schema,
                $dialect,
                |columns, row_ids, begin_snapshots| {
                    let mut stmt = conn.prepare(&sql).map_err(sql_err)?;
                    let mut rows = stmt.query(params![snapshot.0]).map_err(sql_err)?;
                    while let Some(row) = rows.next().map_err(sql_err)? {
                        row_ids.push(row.get(0).map_err(sql_err)?);
                        begin_snapshots
                            .push(ducklake_catalog::SnapshotId(row.get(1).map_err(sql_err)?));
                        for (idx, column) in columns.iter_mut().enumerate() {
                            $crate::inlined_values::push_decoded_inlined_column(
                                column,
                                row.get_ref(idx + 2).map_err(sql_err)?,
                                &$decoder,
                            )?;
                        }
                    }
                    Ok(())
                },
            )
        }
    };
}

macro_rules! impl_catalog_backend_read_methods_direct {
    () => {
        catalog_async_method! { |this|
            fn snapshots(&self) -> $crate::Result<Vec<ducklake_catalog::Snapshot>> {
                $crate::catalog_async_query_methods::AsyncReadQueryExecutor::run_read_query(this, &$crate::embedded_reads::snapshots()).await
            }
        }

        catalog_async_method! { |this|
            fn snapshot_changes(&self) -> $crate::Result<Vec<ducklake_catalog::SnapshotChange>> {
                $crate::catalog_async_query_methods::AsyncReadQueryExecutor::run_read_query(this, &$crate::embedded_reads::snapshot_changes())
                    .await
            }
        }

        catalog_async_method! { |this|
            fn schema_versions(
                &self,
                table: ducklake_catalog::TableId,
            ) -> $crate::Result<Vec<ducklake_catalog::SchemaVersion>> {
                $crate::catalog_async_query_methods::AsyncReadQueryExecutor::run_read_query(this, &$crate::embedded_reads::schema_versions(table))
                    .await
            }
        }

        catalog_async_method! { |this|
            fn schemas_at(
                &self,
                snapshot: ducklake_catalog::SnapshotId,
            ) -> $crate::Result<Vec<ducklake_catalog::SchemaEntry>> {
                $crate::catalog_async_query_methods::AsyncReadQueryExecutor::run_read_query(this, &$crate::embedded_reads::schemas_at(snapshot))
                    .await
            }
        }

        catalog_async_method! { |this|
            fn tables_at(
                &self,
                schema: ducklake_catalog::SchemaId,
                snapshot: ducklake_catalog::SnapshotId,
            ) -> $crate::Result<Vec<ducklake_catalog::TableEntry>> {
                $crate::catalog_async_query_methods::AsyncReadQueryExecutor::run_read_query(this, &$crate::embedded_reads::tables_at(schema, snapshot))
                    .await
            }
        }

        catalog_async_method! { |this|
            fn views_at(
                &self,
                schema: ducklake_catalog::SchemaId,
                snapshot: ducklake_catalog::SnapshotId,
            ) -> $crate::Result<Vec<ducklake_catalog::ViewEntry>> {
                $crate::catalog_async_query_methods::AsyncReadQueryExecutor::run_read_query(this, &$crate::embedded_reads::views_at(schema, snapshot))
                    .await
            }
        }

        catalog_async_method! { |this|
            fn tags_at(
                &self,
                snapshot: ducklake_catalog::SnapshotId,
            ) -> $crate::Result<Vec<ducklake_catalog::TagEntry>> {
                $crate::catalog_async_query_methods::AsyncReadQueryExecutor::run_read_query(this, &$crate::embedded_reads::tags_at(snapshot))
                    .await
            }
        }

        catalog_async_method! { |this|
            fn column_tags_at(
                &self,
                table: ducklake_catalog::TableId,
                snapshot: ducklake_catalog::SnapshotId,
            ) -> $crate::Result<Vec<ducklake_catalog::ColumnTagEntry>> {
                $crate::catalog_async_query_methods::AsyncReadQueryExecutor::run_read_query(this, &$crate::embedded_reads::column_tags_at(table, snapshot))
                    .await
            }
        }

        catalog_async_method! { |this|
            fn columns_at(
                &self,
                table: ducklake_catalog::TableId,
                snapshot: ducklake_catalog::SnapshotId,
            ) -> $crate::Result<Vec<ducklake_catalog::ColumnEntry>> {
                $crate::catalog_async_query_methods::AsyncReadQueryExecutor::run_read_query(this, &$crate::embedded_reads::columns_at(table, snapshot))
                    .await
            }
        }

        catalog_async_method! { |this|
            fn data_files_at(
                &self,
                table: ducklake_catalog::TableId,
                snapshot: ducklake_catalog::SnapshotId,
            ) -> $crate::Result<Vec<ducklake_catalog::DataFile>> {
                $crate::catalog_async_query_methods::AsyncReadQueryExecutor::run_read_query(this, &$crate::embedded_reads::data_files_at(table, snapshot))
                    .await
            }
        }

        catalog_async_method! { |this|
            fn delete_files_at(
                &self,
                table: ducklake_catalog::TableId,
                snapshot: ducklake_catalog::SnapshotId,
            ) -> $crate::Result<Vec<ducklake_catalog::DeleteFile>> {
                $crate::catalog_async_query_methods::AsyncReadQueryExecutor::run_read_query(this, &$crate::embedded_reads::delete_files_at(table, snapshot))
                    .await
            }
        }

        catalog_async_method! { |this|
            fn file_stats(
                &self,
                table: ducklake_catalog::TableId,
            ) -> $crate::Result<Vec<ducklake_catalog::FileColumnStats>> {
                $crate::catalog_async_query_methods::AsyncReadQueryExecutor::run_read_query(this, &$crate::embedded_reads::file_stats(table))
                    .await
            }
        }

        catalog_async_method! { |this|
            fn file_variant_stats(
                &self,
                table: ducklake_catalog::TableId,
            ) -> $crate::Result<Vec<ducklake_catalog::FileVariantStats>> {
                $crate::catalog_async_query_methods::AsyncReadQueryExecutor::run_read_query(this, &$crate::embedded_reads::file_variant_stats(table))
                    .await
            }
        }

        catalog_async_method! { |this|
            fn table_stats(
                &self,
                table: ducklake_catalog::TableId,
            ) -> $crate::Result<Option<ducklake_catalog::TableStats>> {
                let rows =
                    $crate::catalog_async_query_methods::AsyncReadQueryExecutor::run_read_query(
                        this,
                        &$crate::embedded_reads::table_stats(table),
                    )
                    .await?;
                match rows.len() {
                    0 => Ok(None),
                    1 => Ok(rows.into_iter().next()),
                    count => Err($crate::StorageError::Backend(format!(
                        "DuckLake table_stats returned {count} rows for table {}",
                        table.0
                    ))),
                }
            }
        }

        catalog_async_method! { |this|
            fn table_column_stats(
                &self,
                table: ducklake_catalog::TableId,
            ) -> $crate::Result<Vec<ducklake_catalog::TableColumnStats>> {
                $crate::catalog_async_query_methods::AsyncReadQueryExecutor::run_read_query(this, &$crate::embedded_reads::table_column_stats(table))
                    .await
            }
        }

        catalog_async_method! { |this|
            fn partition_info_at(
                &self,
                table: ducklake_catalog::TableId,
                snapshot: ducklake_catalog::SnapshotId,
            ) -> $crate::Result<Vec<ducklake_catalog::PartitionInfo>> {
                $crate::catalog_async_query_methods::AsyncReadQueryExecutor::run_read_query(this, &$crate::embedded_reads::partition_info_at(table, snapshot))
                    .await
            }
        }

        catalog_async_method! { |this|
            fn partition_columns(
                &self,
                partition: ducklake_catalog::PartitionId,
            ) -> $crate::Result<Vec<ducklake_catalog::PartitionColumn>> {
                $crate::catalog_async_query_methods::AsyncReadQueryExecutor::run_read_query(this, &$crate::embedded_reads::partition_columns(partition))
                    .await
            }
        }

        catalog_async_method! { |this|
            fn file_partition_values(
                &self,
                table: ducklake_catalog::TableId,
            ) -> $crate::Result<Vec<ducklake_catalog::FilePartitionValue>> {
                $crate::catalog_async_query_methods::AsyncReadQueryExecutor::run_read_query(this, &$crate::embedded_reads::file_partition_values(table))
                    .await
            }
        }

        catalog_async_method! { |this|
            fn files_scheduled_for_deletion(
                &self,
            ) -> $crate::Result<Vec<ducklake_catalog::FileScheduledForDeletion>> {
                $crate::catalog_async_query_methods::AsyncReadQueryExecutor::run_read_query(this, &$crate::embedded_reads::files_scheduled_for_deletion())
                    .await
            }
        }

        catalog_async_method! { |this|
            fn column_mapping(
                &self,
                mapping: ducklake_catalog::MappingId,
            ) -> $crate::Result<(
                ducklake_catalog::ColumnMapping,
                Vec<ducklake_catalog::NameMappingEntry>,
            )> {
                let mapping_rows =
                    $crate::catalog_async_query_methods::AsyncReadQueryExecutor::run_read_query(
                        this,
                        &$crate::embedded_reads::column_mapping_head(mapping),
                    )
                    .await?;
                let column_mapping = match mapping_rows.len() {
                    0 => {
                        return Err($crate::StorageError::Backend(format!(
                            "mapping {} not found",
                            mapping.0
                        )));
                    }
                    1 => mapping_rows.into_iter().next().expect("one mapping row"),
                    count => {
                        return Err($crate::StorageError::Backend(format!(
                            "DuckLake column mapping {} returned {count} rows",
                            mapping.0
                        )));
                    }
                };
                let entries =
                    $crate::catalog_async_query_methods::AsyncReadQueryExecutor::run_read_query(
                        this,
                        &$crate::embedded_reads::name_mappings(mapping),
                    )
                    .await?;
                Ok((column_mapping, entries))
            }
        }

        catalog_async_method! { |this|
            fn inlined_tables(
                &self,
                table: ducklake_catalog::TableId,
                schema_version: i64,
            ) -> $crate::Result<Vec<ducklake_catalog::InlinedDataTableRef>> {
                $crate::catalog_async_query_methods::AsyncReadQueryExecutor::run_read_query(this, &$crate::embedded_reads::inlined_tables(table, schema_version))
                    .await
            }
        }

        catalog_async_method! { |this|
            fn macros_at(
                &self,
                schema: ducklake_catalog::SchemaId,
                snapshot: ducklake_catalog::SnapshotId,
            ) -> $crate::Result<Vec<ducklake_catalog::MacroEntry>> {
                $crate::catalog_async_query_methods::AsyncReadQueryExecutor::run_read_query(this, &$crate::embedded_reads::macros_at(schema, snapshot))
                    .await
            }
        }

        catalog_async_method! { |this|
            fn macro_impls(
                &self,
                macro_id: ducklake_catalog::MacroId,
            ) -> $crate::Result<Vec<ducklake_catalog::MacroImpl>> {
                $crate::catalog_async_query_methods::AsyncReadQueryExecutor::run_read_query(this, &$crate::embedded_reads::macro_impls(macro_id))
                    .await
            }
        }

        catalog_async_method! { |this|
            fn macro_parameters(
                &self,
                macro_id: ducklake_catalog::MacroId,
                impl_id: ducklake_catalog::MacroImplId,
            ) -> $crate::Result<Vec<ducklake_catalog::MacroParameter>> {
                $crate::catalog_async_query_methods::AsyncReadQueryExecutor::run_read_query(this, &$crate::embedded_reads::macro_parameters(macro_id, impl_id))
                    .await
            }
        }

        catalog_async_method! { |this|
            fn sort_info_at(
                &self,
                table: ducklake_catalog::TableId,
                snapshot: ducklake_catalog::SnapshotId,
            ) -> $crate::Result<Vec<ducklake_catalog::SortInfo>> {
                $crate::catalog_async_query_methods::AsyncReadQueryExecutor::run_read_query(this, &$crate::embedded_reads::sort_info_at(table, snapshot))
                    .await
            }
        }

        catalog_async_method! { |this|
            fn sort_expressions(
                &self,
                sort_id: ducklake_catalog::SortId,
            ) -> $crate::Result<Vec<ducklake_catalog::SortExpression>> {
                $crate::catalog_async_query_methods::AsyncReadQueryExecutor::run_read_query(this, &$crate::embedded_reads::sort_expressions(sort_id))
                    .await
            }
        }
    };
}
