//! PostgreSQL DuckLake catalog backend.

#[cfg(test)]
use std::collections::HashMap;

#[cfg(test)]
use crate::{
    AddColumnCommit, CleanupOldFilesOptions, ColumnDefaultCommit, ColumnTypeCommit,
    ColumnTypeTreeCommit, CreateSchemaCommit, CreateTableCommit, CreateViewCommit,
    DeleteFileCommit, DropColumnCommit, DropSchemaCommit, DropTableCommit, DropViewCommit,
    ExpireSnapshotsOptions, FlushedInlinedDataCommit, FlushedInlinedFileDeletionCommit,
    InlinedFileDeletionCommit, MergeAdjacentCompactionCommit, RenameColumnCommit,
    RenameTableCommit, RewriteDataFilesCommit, SortedByCommit, TableChangeCommit,
    TruncateTableCommit,
};
use arrow::array::Array;
#[cfg(test)]
use arrow::array::{
    BinaryArray, Date32Array, Decimal128Array, Decimal256Array, FixedSizeBinaryArray, Float32Array,
    Float64Array, Int8Array, Int16Array, IntervalMonthDayNanoArray, StringArray,
    Time64MicrosecondArray, Time64NanosecondArray, TimestampMicrosecondArray,
    TimestampMillisecondArray, TimestampNanosecondArray, TimestampSecondArray, UInt8Array,
    UInt16Array, UInt32Array, UInt64Array,
};
#[cfg(test)]
use arrow::datatypes::{IntervalMonthDayNano, i256};
use arrow::record_batch::RecordBatch;
#[cfg(test)]
use arrow_schema::DataType;
use async_trait::async_trait;
#[cfg(test)]
use ducklake_catalog::{
    ColumnId, DataFileId, FileColumnStats, MacroId, MacroImplId, MappingId, PartitionId, SchemaId,
    SortId, TableId,
};
use ducklake_catalog::{SchemaField, SnapshotId, TableColumnStats, TableSchema};
use tokio_postgres::types::{FromSqlOwned, ToSql};
use tokio_postgres::{Client, GenericClient, NoTls, Row, Transaction};

use crate::catalog_async_commit_methods::AsyncCatalogQueryCommitExecutor;
use crate::catalog_async_query_methods::{AsyncCatalogQueryExecutor, AsyncReadQueryExecutor};
#[cfg(test)]
use crate::inlined_arrow_field_from_schema;
#[cfg(test)]
use crate::inlined_values::{
    InlinedColumnBuffer as InlinedColumn, inlined_column_type, push_decoded_inlined_column,
};
use crate::inlined_values::{
    InlinedSqlDialect, TextInlinedValueDecoder, postgres_inlined_placeholder_cast,
};
#[cfg(test)]
use crate::inlined_values::{
    parse_inlined_bool_text, parse_inlined_decimal128, parse_inlined_decimal256,
    parse_inlined_hex_bytes, parse_inlined_i64_text, parse_inlined_u64_text,
    parse_inlined_utf8_hex_bytes, parse_inlined_uuid,
};
use crate::{
    CatalogBackend, CleanupMetadataFile, DuckLakeCatalogConnectionOptions, InlinedDataBatch,
    Result, StorageError, ducklake_quote_identifier as quote_ident,
};

#[derive(Debug, Clone)]
pub struct PostgresCatalog {
    config: String,
    metadata_schema: String,
    connection_options: DuckLakeCatalogConnectionOptions,
}

impl PostgresCatalog {
    pub fn new(config: impl Into<String>) -> Self {
        Self {
            config: config.into(),
            metadata_schema: "ducklake_metadata".into(),
            connection_options: DuckLakeCatalogConnectionOptions::default(),
        }
    }

    pub fn with_metadata_schema(
        config: impl Into<String>,
        metadata_schema: impl Into<String>,
    ) -> Self {
        Self {
            config: config.into(),
            metadata_schema: metadata_schema.into(),
            connection_options: DuckLakeCatalogConnectionOptions::default(),
        }
    }

    impl_connection_options_builder_method!();

    pub fn config(&self) -> &str {
        &self.config
    }

    impl_metadata_schema_accessor_method!(&str, |this| &this.metadata_schema);

    async fn client(&self) -> Result<Client> {
        let (client, connection) = tokio_postgres::connect(&self.config, NoTls)
            .await
            .map_err(pg_err)?;
        tokio::spawn(async move {
            let _ = connection.await;
        });
        Ok(client)
    }

    async fn with_transaction<T, F>(&self, f: F) -> Result<T>
    where
        T: Send,
        F: for<'tx> FnOnce(
            &'tx Transaction<'tx>,
        ) -> std::pin::Pin<
            Box<dyn std::future::Future<Output = Result<T>> + Send + 'tx>,
        >,
    {
        let mut client = self.client().await?;
        let tx = client.transaction().await.map_err(pg_err)?;
        let result = f(&tx).await?;
        tx.commit().await.map_err(pg_err)?;
        Ok(result)
    }

    async fn run_commit_request(
        &self,
        request: crate::commit_engine::CommitRequest,
    ) -> Result<SnapshotId> {
        request.validate_preflight()?;
        let metadata_schema = self.metadata_schema.clone();
        self.with_transaction(|tx| {
            Box::pin(async move {
                let ops = PostgresCommitOps {
                    metadata_schema: &metadata_schema,
                };
                crate::commit_engine::apply_async_commit_request(&ops, tx, &request).await
            })
        })
        .await
    }

    fn table(&self, name: &str) -> String {
        format!(
            "{}.{}",
            quote_ident(&self.metadata_schema),
            quote_ident(name)
        )
    }
}

#[async_trait]
impl AsyncReadQueryExecutor for PostgresCatalog {
    fn read_query_render(&self) -> crate::catalog_async_query_methods::ReadQueryRender {
        crate::catalog_async_query_methods::ReadQueryRender::Postgres
    }

    fn read_query_metadata_schema(&self) -> Option<&str> {
        Some(&self.metadata_schema)
    }

    async fn execute_rendered_read_query<T>(
        &self,
        sql: String,
        params: Vec<i64>,
        mapper: for<'row> fn(&'row dyn crate::embedded_rows::EmbeddedRow) -> Result<T>,
    ) -> Result<Vec<T>>
    where
        T: Send + 'static,
    {
        let param_refs = params
            .iter()
            .map(|param| param as &(dyn ToSql + Sync))
            .collect::<Vec<_>>();
        let client = self.client().await?;
        let rows = client.query(&sql, &param_refs).await.map_err(pg_err)?;
        crate::embedded_rows::collect_embedded_rows(rows.iter(), mapper)
    }
}

struct PostgresCommitOps<'a> {
    metadata_schema: &'a str,
}

impl_async_catalog_query_commit_executor!(
    impl<'schema, 'tx> PostgresCommitOps<'schema>, Transaction<'tx> {
        execute = |this, tx, query| async move {
            let executor = postgres_catalog_query_executor(tx, this.metadata_schema);
            executor.execute_catalog_command(query).await
        },
        query = |this, tx, query, mapper| async move {
            let executor = postgres_catalog_query_executor(tx, this.metadata_schema);
            executor.run_catalog_query(query, mapper).await
        },
        table_exists = |this, tx, table_name| {
            postgres_table_exists(tx, this.metadata_schema, table_name)
        },
    }
);

impl_async_commit_ops!(
    impl<'schema, 'tx> PostgresCommitOps<'schema>, Transaction<'tx> {
        latest_snapshot_ids = |this, tx| crate::catalog_async_commit_methods::load_latest_snapshot_ids(this, tx, "commit"),
        table_exists = |this, tx, table_name| {
            postgres_table_exists(tx, this.metadata_schema, table_name)
        },
        insert_snapshot = |this, tx, snapshot_id, schema_version, next_catalog_id, next_file_id| async {
            let now_micros = crate::sql_utils::current_time_micros()?;
            let now_seconds = now_micros as f64 / 1_000_000.0;
            tx.execute(
                &format!(
                    "INSERT INTO {} VALUES ($1, to_timestamp($2::DOUBLE PRECISION), $3, $4, $5)",
                    postgres_qualified_table(this.metadata_schema, "ducklake_snapshot")
                ),
                &[
                    &snapshot_id,
                    &now_seconds,
                    &schema_version,
                    &next_catalog_id,
                    &next_file_id,
                ],
            )
            .await
            .map_err(pg_err)?;
            Ok(())
        },
        insert_snapshot_change = |this, tx, snapshot_id, changes_made, commit_message| async {
            let executor = postgres_catalog_query_executor(tx, this.metadata_schema);
            executor
                .execute_catalog_command(&crate::catalog_queries::insert_snapshot_change(
                    snapshot_id,
                    changes_made,
                    commit_message,
                ))
                .await?;
            Ok(())
        },
        extra {
            async fn prepare_commit(&self, tx: &Transaction<'tx>) -> Result<()> {
                tx.batch_execute(&format!(
                    "LOCK TABLE {} IN EXCLUSIVE MODE",
                    postgres_qualified_table(self.metadata_schema, "ducklake_snapshot")
                ))
                .await
                .map_err(pg_err)?;
                Ok(())
            }
        },
    }
);

impl_async_table_data_commit_executor!(
    impl<'schema, 'tx> PostgresCommitOps<'schema>, Transaction<'tx> {
        dialect = crate::inlined_values::InlinedSqlDialect::Postgres,
        upsert_table_column_stats = |this, tx, stats| {
            upsert_postgres_table_column_stats(tx, this.metadata_schema, &stats)
        },
        inlined_executor = |this, tx| {
            PostgresInlinedMutationOps {
                tx,
                metadata_schema: this.metadata_schema,
            }
        },
        replace_overwritten_delete_file = |this, tx, file, new_snapshot| async move {
            crate::catalog_async_commit_methods::replace_overwritten_delete_file(
                this,
                tx,
                file,
                new_snapshot,
                || async move {
                    postgres_global_metadata_value_tx(tx, this.metadata_schema, "data_path")
                        .await
                        .map(|value| value.unwrap_or_default())
                },
                |cleanup_row| async move {
                    let ops = PostgresSnapshotPruneOps {
                        tx,
                        schema: this.metadata_schema,
                    };
                    crate::catalog_async_maintenance_methods::schedule_cleanup_metadata_files(
                        &ops,
                        &[cleanup_row],
                    )
                    .await
                },
            )
            .await
        },
    }
);

impl_async_maintenance_commit_executor!(
    impl<'schema, 'tx> PostgresCommitOps<'schema>, Transaction<'tx> {
        cleanup_executor = |this, tx| {
            PostgresSnapshotPruneOps {
                tx,
                schema: this.metadata_schema,
            }
        },
        inlined_executor = |this, tx| {
            PostgresInlinedMutationOps {
                tx,
                metadata_schema: this.metadata_schema,
            }
        },
    }
);

impl_async_catalog_object_commit_ops!(
    impl<'schema, 'tx> PostgresCommitOps<'schema>,
    Transaction<'tx>,
    uuid_style = crate::catalog_queries::UuidParamStyle::CastTextAsUuid,
    inlined_executor = |this, tx| {
        PostgresInlinedMutationOps {
            tx,
            metadata_schema: this.metadata_schema,
        }
    },
    data_path = |this, tx| async move {
        postgres_global_metadata_value_tx(tx, this.metadata_schema, "data_path")
            .await
            .map(|value| value.unwrap_or_default())
    }
);

impl<'schema, 'tx> crate::commit::AsyncCreateTableWithFilesCommitOps<Transaction<'tx>>
    for PostgresCommitOps<'schema>
{
}

impl_async_partitioning_commit_ops!(impl<'schema, 'tx> PostgresCommitOps<'schema>, Transaction<'tx>);
impl_async_sorted_by_commit_ops!(impl<'schema, 'tx> PostgresCommitOps<'schema>, Transaction<'tx>);
impl_async_column_evolution_commit_ops! {
    impl<'schema, 'tx> PostgresCommitOps<'schema>, Transaction<'tx>,
    extra {}
}

impl_async_catalog_maintenance_executor!(
    impl PostgresCatalog {
        load_cleanup_data_path = |this| async {
            let client = this.client().await?;
            Ok(postgres_global_metadata_value(
                &client,
                &this.metadata_schema,
                "data_path",
            )
            .await?
            .unwrap_or_default())
        },
        load_scheduled_cleanup_files = |this| async {
            let client = this.client().await?;
            let rows = client
                .query(
                    &format!(
                        "SELECT data_file_id, path, path_is_relative, \
                         (EXTRACT(EPOCH FROM schedule_start) * 1000000)::BIGINT \
                         FROM {} ORDER BY schedule_start, data_file_id",
                        this.table("ducklake_files_scheduled_for_deletion")
                    ),
                    &[],
                )
                .await
                .map_err(pg_err)?;

            rows.iter()
                .map(|row| crate::embedded_reads::file_scheduled_for_deletion(row))
                .collect()
        },
        delete_scheduled_cleanup_files = |this, data_file_ids| async {
            if data_file_ids.is_empty() {
                return Ok(());
            }
            let table = this.table("ducklake_files_scheduled_for_deletion");
            let data_file_ids = data_file_ids.to_vec();
            this.with_transaction(|tx| {
                Box::pin(async move {
                    for data_file_id in data_file_ids {
                        tx.execute(
                            &format!("DELETE FROM {table} WHERE data_file_id = $1"),
                            &[&data_file_id],
                        )
                        .await
                        .map_err(pg_err)?;
                    }
                    Ok(())
                })
            })
            .await
        },
        load_expirable_snapshots = |this, options| async {
            crate::catalog_async_maintenance_methods::load_expirable_snapshots_from_read_executor(
                this,
                options,
            )
            .await
        },
        prune_expired_snapshots = |this, snapshots| async {
            let metadata_schema = this.metadata_schema.clone();
            let snapshots = snapshots.to_vec();
            this.with_transaction(|tx| {
                Box::pin(async move {
                    crate::catalog_async_maintenance_methods::prune_expired_snapshots(
                        &PostgresSnapshotPruneOps {
                            tx,
                            schema: &metadata_schema,
                        },
                        &snapshots,
                    )
                    .await
                })
            })
            .await
        },
    }
);

impl_async_catalog_metadata_executor!(
    impl PostgresCatalog {
        connection_options = |this| &this.connection_options,
        replace_metadata_option = |this, key, value, scope| async {
            let normalized_key = crate::metadata_option::normalized_metadata_option_key(&key);
            let sql = crate::metadata_option::render_set_metadata_option(
                &this.table("ducklake_metadata"),
                crate::metadata_option::MetadataPlaceholderStyle::Dollar,
                crate::metadata_option::NullSafeEquality::IsNotDistinctFrom,
            );
            this.with_transaction(|tx| {
                Box::pin(async move {
                    tx.execute(sql.delete.as_str(), &[&key, &scope.scope, &scope.scope_id])
                        .await
                        .map_err(pg_err)?;
                    tx.execute(
                        sql.insert.as_str(),
                        &[&normalized_key, &value, &scope.scope, &scope.scope_id],
                    )
                    .await
                    .map_err(pg_err)?;
                    Ok(())
                })
            })
            .await
        },
    }
);

impl_async_inlined_read_executor!(
    impl PostgresCatalog {
        file_deletion_table_exists = |this, table_name| async {
            let client = this.client().await?;
            postgres_table_exists(&client, &this.metadata_schema, table_name).await
        },
        load_file_deletions = |this, table_name, snapshot| async {
            crate::catalog_async_query_methods::AsyncReadQueryExecutor::run_read_query(
                this,
                &crate::embedded_reads::inlined_file_deletions(table_name, snapshot),
            )
            .await
        },
        load_data = |this, table, schema, snapshot| async {
            let client = this.client().await?;
            read_inlined_data(&client, &this.table(&table.table_name), schema, snapshot).await
        },
        load_data_with_metadata = |this, table, schema, snapshot| async {
            let client = this.client().await?;
            read_inlined_data_with_metadata(
                &client,
                &this.table(&table.table_name),
                schema,
                snapshot,
            )
            .await
        },
    }
);

#[async_trait]
impl crate::catalog_initialization_methods::CatalogInitializationExecutor for PostgresCatalog {
    impl_catalog_initialization_info_methods!(
        catalog_type = "postgres",
        default_data_path = |_this| None,
        snapshot_time_sql = |_this| Ok("NOW()".into())
    );

    async fn run_initialization_request(
        &self,
        request: crate::catalog_initialization_methods::CatalogInitializationRequest,
    ) -> Result<crate::catalog_initialization_methods::CatalogInitializationResponse> {
        let metadata_schema = self.metadata_schema.clone();
        self.with_transaction(|tx| {
            Box::pin(async move {
                request
                    .execute_async(
                        || async {
                            postgres_table_exists(tx, &metadata_schema, "ducklake_metadata").await
                        },
                        || async {
                            let executor = postgres_catalog_query_executor(tx, &metadata_schema);
                            crate::catalog_async_metadata_methods::load_catalog_metadata(&executor)
                                .await
                        },
                        |bootstrap| {
                            let create_schema =
                                crate::catalog_ddl::render_create_schema(&metadata_schema);
                            let initialization_batch =
                                crate::catalog_ddl::render_initialization_batch(
                                    Some(&metadata_schema),
                                    &bootstrap,
                                );
                            async move {
                                tx.batch_execute(&create_schema).await.map_err(pg_err)?;
                                tx.batch_execute(&initialization_batch)
                                    .await
                                    .map_err(pg_err)?;
                                Ok(())
                            }
                        },
                    )
                    .await
            })
        })
        .await
    }
}

#[async_trait]
impl CatalogBackend for PostgresCatalog {
    impl_catalog_type_method!("postgres");

    impl_async_catalog_metadata_methods!();

    impl_catalog_initialization_method!();

    impl_catalog_backend_read_methods_direct!();

    impl_async_inlined_read_methods!();

    impl_catalog_backend_commit_methods!();

    impl_async_catalog_maintenance_methods!();
}

struct PostgresSnapshotPruneOps<'a, 'tx> {
    tx: &'a Transaction<'tx>,
    schema: &'a str,
}

type PostgresMetadataPruneOps<'a, 'tx> = PostgresSnapshotPruneOps<'a, 'tx>;

#[async_trait]
impl crate::catalog_async_maintenance_methods::AsyncCleanupSqlExecutor
    for PostgresSnapshotPruneOps<'_, '_>
{
    fn metadata_table_ref(&self, table: &str) -> String {
        postgres_qualified_table(self.schema, table)
    }

    fn i64_placeholder(&self, index: usize) -> String {
        format!("${index}")
    }

    async fn cleanup_table_exists(&self, table: &str) -> Result<bool> {
        postgres_table_exists(self.tx, self.schema, table).await
    }

    async fn cleanup_path_context_available(&self) -> Result<bool> {
        Ok(
            postgres_table_exists(self.tx, self.schema, "ducklake_table").await?
                && postgres_table_exists(self.tx, self.schema, "ducklake_schema").await?,
        )
    }

    async fn cleanup_data_path(&self) -> Result<String> {
        postgres_global_metadata_value_tx(self.tx, self.schema, "data_path")
            .await
            .map(|value| value.unwrap_or_default())
    }

    async fn query_i64_rows<T, F>(&self, sql: String, params: Vec<i64>, mapper: F) -> Result<Vec<T>>
    where
        T: Send + 'static,
        F: FnMut(&dyn crate::embedded_rows::EmbeddedRow) -> Result<T> + Send,
    {
        let param_refs = params
            .iter()
            .map(|param| param as &(dyn ToSql + Sync))
            .collect::<Vec<_>>();
        let rows = self.tx.query(&sql, &param_refs).await.map_err(pg_err)?;
        crate::embedded_rows::collect_embedded_rows(rows.iter(), mapper)
    }

    async fn insert_scheduled_cleanup_file(&self, file: CleanupMetadataFile) -> Result<()> {
        self.tx
            .execute(
                &crate::embedded_maintenance::insert_scheduled_cleanup_file_sql(
                    &postgres_qualified_table(self.schema, "ducklake_files_scheduled_for_deletion"),
                    "$1",
                    "$2",
                    "$3",
                    "CURRENT_TIMESTAMP",
                ),
                &[&file.id, &file.path, &file.path_is_relative],
            )
            .await
            .map_err(pg_err)?;
        Ok(())
    }
}

struct PostgresInlinedMutationOps<'a, 'tx> {
    tx: &'a Transaction<'tx>,
    metadata_schema: &'a str,
}

#[async_trait]
impl crate::catalog_async_inlined_mutation_methods::AsyncInlinedMutationExecutor
    for PostgresInlinedMutationOps<'_, '_>
{
    fn inlined_sql_dialect(&self) -> crate::inlined_values::InlinedSqlDialect {
        InlinedSqlDialect::Postgres
    }

    fn inlined_metadata_table_identifier(&self, table_name: &str) -> String {
        postgres_qualified_table(self.metadata_schema, table_name)
    }

    async fn inlined_table_exists(&self, table_name: String) -> Result<bool> {
        postgres_table_exists(self.tx, self.metadata_schema, &table_name).await
    }

    async fn execute_inlined_sql(&self, sql: String) -> Result<()> {
        self.tx.execute(&sql, &[]).await.map_err(pg_err)?;
        Ok(())
    }

    async fn query_inlined_i64(&self, sql: String) -> Result<i64> {
        let row = self.tx.query_one(&sql, &[]).await.map_err(pg_err)?;
        get(&row, 0)
    }

    async fn query_optional_inlined_string(&self, sql: String) -> Result<Option<String>> {
        self.tx
            .query_opt(&sql, &[])
            .await
            .map_err(pg_err)?
            .map(|row| get::<String>(&row, 0))
            .transpose()
    }
}

#[async_trait]
impl crate::catalog_async_inlined_mutation_methods::AsyncInlinedBatchInsertExecutor
    for PostgresInlinedMutationOps<'_, '_>
{
    type Value = Box<dyn ToSql + Sync + Send>;
    type InsertCommand = String;

    fn prepare_inlined_insert(&self, table_name: &str, schema: &TableSchema) -> Result<String> {
        Ok(crate::inlined_values::inlined_insert_sql(
            &postgres_qualified_table(self.metadata_schema, table_name),
            &postgres_inlined_insert_placeholders(schema)?,
        ))
    }

    fn inlined_insert_i64_value(&self, value: i64) -> Self::Value {
        Box::new(value)
    }

    fn inlined_insert_null_value(&self) -> Self::Value {
        Box::new(Option::<i64>::None)
    }

    fn inlined_insert_value(
        &self,
        field: &SchemaField,
        array: &dyn Array,
        row: usize,
    ) -> Result<Self::Value> {
        postgres_inlined_value(field, array, row)
    }

    async fn execute_prepared_inlined_insert(
        &self,
        command: &Self::InsertCommand,
        values: Vec<Self::Value>,
    ) -> Result<()> {
        let params = values
            .iter()
            .map(|value| value.as_ref() as &(dyn ToSql + Sync))
            .collect::<Vec<_>>();
        self.tx.execute(command, &params).await.map_err(pg_err)?;
        Ok(())
    }
}

impl_async_metadata_prune_executor!(
    impl PostgresMetadataPruneOps<'_, '_> {
        table_exists = |this, table| postgres_table_exists(this.tx, this.schema, table),
        inlined_table_names = |this, table_ids| async {
            let table = postgres_qualified_table(this.schema, "ducklake_inlined_data_tables");
            let rows = this
                .tx
                .query(
                    &crate::metadata_prune_engine::inlined_table_names_for_table_ids_sql(
                        &table,
                        table_ids,
                    ),
                    &[],
                )
                .await
                .map_err(pg_err)?;
            crate::embedded_rows::collect_embedded_rows(rows.iter(), |row| row.get_string(0))
        },
        drop_inlined_table = |this, table_name| async {
            this.tx
                .execute(
                    &crate::metadata_prune_engine::drop_table_if_exists_sql(
                        &postgres_qualified_table(this.schema, table_name),
                    ),
                    &[],
                )
                .await
                .map_err(pg_err)?;
            Ok(())
        },
        snapshot_table_ref = |this| postgres_qualified_table(this.schema, "ducklake_snapshot"),
        macro_table_ref = |this| postgres_qualified_table(this.schema, "ducklake_macro"),
        macro_dependent_target_ref = |_this, table| {
            postgres_dependent_metadata_alias(table).to_string()
        },
        column_mapping_table_ref = |this| {
            postgres_qualified_table(this.schema, "ducklake_column_mapping")
        },
        name_mapping_target_ref = |_this| "nm".to_string(),
        delete_where = |this, table, target_ref, predicate| {
            postgres_delete_where(this.tx, this.schema, table, target_ref, predicate)
        },
    }
);

fn postgres_dependent_metadata_alias(table: &str) -> &'static str {
    match table {
        "ducklake_macro_impl" => "mi",
        "ducklake_macro_parameters" => "mp",
        _ => "target",
    }
}

async fn postgres_delete_where(
    tx: &Transaction<'_>,
    schema: &str,
    table: &str,
    target_ref: Option<&str>,
    predicate: &str,
) -> Result<()> {
    if !postgres_table_exists(tx, schema, table).await? {
        return Ok(());
    }
    tx.execute(
        &crate::metadata_prune_engine::delete_where_sql(
            &postgres_qualified_table(schema, table),
            target_ref,
            predicate,
        ),
        &[],
    )
    .await
    .map_err(pg_err)?;
    Ok(())
}

fn postgres_inlined_insert_placeholders(schema: &TableSchema) -> Result<String> {
    crate::inlined_values::inlined_insert_placeholders(
        schema,
        (1..=3).map(|idx| format!("${idx}")),
        |idx, field| {
            let parameter_index = idx + 4;
            Ok(match postgres_inlined_placeholder_cast(field) {
                Some(cast) => format!("CAST(${parameter_index} AS {cast})"),
                None => format!("${parameter_index}"),
            })
        },
    )
}

fn postgres_inlined_value(
    field: &SchemaField,
    array: &dyn Array,
    row: usize,
) -> Result<Box<dyn ToSql + Sync + Send>> {
    if matches!(
        field.ducklake_type,
        ducklake_catalog::DuckLakeType::Struct
            | ducklake_catalog::DuckLakeType::List
            | ducklake_catalog::DuckLakeType::Map
    ) {
        if array.is_null(row) {
            return Ok(Box::new(Option::<String>::None));
        }
        return crate::inlined_values::inlined_duckdb_literal_value(
            field,
            array,
            row,
            "PostgreSQL inlined append",
            crate::inlined_values::InlinedIntervalFormat::MicrosecondsOnly("PostgreSQL"),
        )
        .map(|value| Box::new(value) as Box<dyn ToSql + Sync + Send>);
    }
    crate::inlined_values::encode_inlined_scalar_value(
        field,
        array,
        row,
        "PostgreSQL inlined append",
        crate::inlined_values::InlinedIntervalFormat::MicrosecondsOnly("PostgreSQL"),
        &PostgresInlinedValueEncoder,
    )
}

struct PostgresInlinedValueEncoder;

impl_inlined_scalar_value_encoder!(
    impl PostgresInlinedValueEncoder,
    Value = Box<dyn ToSql + Sync + Send>,
    {
        null(field) => postgres_inlined_null_value(field),
        boolean(value) => Ok(Box::new(value)),
        int8(value) => Ok(Box::new(i16::from(value))),
        int16(value) => Ok(Box::new(value)),
        int32(value) => Ok(Box::new(value)),
        int64(value) => Ok(Box::new(value)),
        uint8(value) => Ok(Box::new(i32::from(value))),
        uint16(value) => Ok(Box::new(i32::from(value))),
        uint32(value) => Ok(Box::new(i64::from(value))),
        uint64(_field, value) => Ok(Box::new(value.to_string())),
        float32(value) => Ok(Box::new(value)),
        float64(value) => Ok(Box::new(value)),
        text(field, value) => postgres_inlined_text_value(field, value),
        bytes(field, value) => postgres_inlined_bytes_value(field, value),
    }
);

fn postgres_inlined_text_value(
    field: &SchemaField,
    value: String,
) -> Result<Box<dyn ToSql + Sync + Send>> {
    match &field.ducklake_type {
        ducklake_catalog::DuckLakeType::Int128
        | ducklake_catalog::DuckLakeType::UInt128
        | ducklake_catalog::DuckLakeType::Decimal { .. }
        | ducklake_catalog::DuckLakeType::Date
        | ducklake_catalog::DuckLakeType::Time
        | ducklake_catalog::DuckLakeType::TimeNs
        | ducklake_catalog::DuckLakeType::Timestamp
        | ducklake_catalog::DuckLakeType::TimestampS
        | ducklake_catalog::DuckLakeType::TimestampMs
        | ducklake_catalog::DuckLakeType::TimestampUs
        | ducklake_catalog::DuckLakeType::TimestampNs
        | ducklake_catalog::DuckLakeType::TimestampTz
        | ducklake_catalog::DuckLakeType::TimeTz
        | ducklake_catalog::DuckLakeType::Interval
        | ducklake_catalog::DuckLakeType::Uuid => Ok(Box::new(value)),
        ducklake_catalog::DuckLakeType::Varchar | ducklake_catalog::DuckLakeType::Json => {
            Ok(Box::new(value.into_bytes()))
        }
        other => Err(StorageError::Unsupported(format!(
            "PostgreSQL inlined append cannot encode DuckLake type {other} as text"
        ))),
    }
}

fn postgres_inlined_bytes_value(
    field: &SchemaField,
    value: Vec<u8>,
) -> Result<Box<dyn ToSql + Sync + Send>> {
    match field.ducklake_type {
        ducklake_catalog::DuckLakeType::Blob => Ok(Box::new(value)),
        ref other => Err(StorageError::Unsupported(format!(
            "PostgreSQL inlined append cannot encode DuckLake type {other} as bytes"
        ))),
    }
}

fn postgres_inlined_null_value(field: &SchemaField) -> Result<Box<dyn ToSql + Sync + Send>> {
    match &field.ducklake_type {
        ducklake_catalog::DuckLakeType::Boolean => Ok(Box::new(Option::<bool>::None)),
        ducklake_catalog::DuckLakeType::Int8 => Ok(Box::new(Option::<i16>::None)),
        ducklake_catalog::DuckLakeType::Int16 => Ok(Box::new(Option::<i16>::None)),
        ducklake_catalog::DuckLakeType::Int32 => Ok(Box::new(Option::<i32>::None)),
        ducklake_catalog::DuckLakeType::Int64 => Ok(Box::new(Option::<i64>::None)),
        ducklake_catalog::DuckLakeType::UInt8 => Ok(Box::new(Option::<i32>::None)),
        ducklake_catalog::DuckLakeType::UInt16 => Ok(Box::new(Option::<i32>::None)),
        ducklake_catalog::DuckLakeType::UInt32 => Ok(Box::new(Option::<i64>::None)),
        ducklake_catalog::DuckLakeType::UInt64
        | ducklake_catalog::DuckLakeType::Int128
        | ducklake_catalog::DuckLakeType::UInt128
        | ducklake_catalog::DuckLakeType::Decimal { .. }
        | ducklake_catalog::DuckLakeType::Date
        | ducklake_catalog::DuckLakeType::Time
        | ducklake_catalog::DuckLakeType::TimeNs
        | ducklake_catalog::DuckLakeType::Timestamp
        | ducklake_catalog::DuckLakeType::TimestampS
        | ducklake_catalog::DuckLakeType::TimestampMs
        | ducklake_catalog::DuckLakeType::TimestampUs
        | ducklake_catalog::DuckLakeType::TimestampNs
        | ducklake_catalog::DuckLakeType::TimestampTz
        | ducklake_catalog::DuckLakeType::TimeTz
        | ducklake_catalog::DuckLakeType::Interval
        | ducklake_catalog::DuckLakeType::Uuid => Ok(Box::new(Option::<String>::None)),
        ducklake_catalog::DuckLakeType::Float32 => Ok(Box::new(Option::<f32>::None)),
        ducklake_catalog::DuckLakeType::Float64 => Ok(Box::new(Option::<f64>::None)),
        ducklake_catalog::DuckLakeType::Blob => Ok(Box::new(Option::<Vec<u8>>::None)),
        ducklake_catalog::DuckLakeType::Varchar | ducklake_catalog::DuckLakeType::Json => {
            Ok(Box::new(Option::<Vec<u8>>::None))
        }
        other => Err(StorageError::Unsupported(format!(
            "Inlined append writes do not support DuckLake type {other}"
        ))),
    }
}

fn postgres_qualified_table(schema: &str, table: &str) -> String {
    format!("{}.{}", quote_ident(schema), quote_ident(table))
}

async fn postgres_global_metadata_value<C>(
    client: &C,
    metadata_schema: &str,
    key: &str,
) -> Result<Option<String>>
where
    C: GenericClient + Sync,
{
    let query = crate::catalog_queries::global_metadata_value(key);
    let rows = postgres_catalog_query_rows(
        client,
        query.render_postgres(metadata_schema),
        query.params().to_vec(),
        crate::catalog_queries::opt_string_value,
    )
    .await?;
    Ok(rows.into_iter().next().flatten())
}

async fn postgres_global_metadata_value_tx(
    tx: &Transaction<'_>,
    schema: &str,
    key: &str,
) -> Result<Option<String>> {
    if !postgres_table_exists(tx, schema, "ducklake_metadata").await? {
        return Ok(None);
    }
    postgres_global_metadata_value(tx, schema, key).await
}

async fn read_inlined_data(
    client: &Client,
    table_name: &str,
    schema: &TableSchema,
    snapshot: SnapshotId,
) -> Result<Vec<RecordBatch>> {
    let projection = crate::inlined_values::inlined_read_projection_for_dialect(
        schema,
        crate::inlined_values::InlinedSqlDialect::Postgres,
        quote_ident,
    )?;
    let sql = crate::inlined_values::read_inlined_data_sql(table_name, &projection, "$1");
    let decoder = TextInlinedValueDecoder::new("PostgreSQL", false, false);

    crate::inlined_values::read_inlined_record_batches_async(
        schema,
        InlinedSqlDialect::Postgres,
        |mut columns| async move {
            let rows = client
                .query(sql.as_str(), &[&snapshot.0])
                .await
                .map_err(pg_err)?;
            for row in rows {
                for (idx, column) in columns.iter_mut().enumerate() {
                    crate::inlined_values::push_decoded_inlined_column(
                        column,
                        row.try_get::<_, Option<String>>(idx).map_err(pg_err)?,
                        &decoder,
                    )?;
                }
            }
            Ok(columns)
        },
    )
    .await
}

async fn read_inlined_data_with_metadata(
    client: &Client,
    table_name: &str,
    schema: &TableSchema,
    snapshot: SnapshotId,
) -> Result<Vec<InlinedDataBatch>> {
    let projection = crate::inlined_values::inlined_read_projection_for_dialect(
        schema,
        crate::inlined_values::InlinedSqlDialect::Postgres,
        quote_ident,
    )?;
    let sql =
        crate::inlined_values::read_inlined_data_with_metadata_sql(table_name, &projection, "$1");
    let decoder = TextInlinedValueDecoder::new("PostgreSQL", false, false);

    crate::inlined_values::read_inlined_data_batches_async(
        schema,
        InlinedSqlDialect::Postgres,
        |mut columns, mut row_ids, mut begin_snapshots| async move {
            let rows = client
                .query(sql.as_str(), &[&snapshot.0])
                .await
                .map_err(pg_err)?;
            for row in rows {
                row_ids.push(row.try_get(0).map_err(pg_err)?);
                begin_snapshots.push(SnapshotId(row.try_get(1).map_err(pg_err)?));
                for (idx, column) in columns.iter_mut().enumerate() {
                    crate::inlined_values::push_decoded_inlined_column(
                        column,
                        row.try_get::<_, Option<String>>(idx + 2).map_err(pg_err)?,
                        &decoder,
                    )?;
                }
            }
            Ok((columns, row_ids, begin_snapshots))
        },
    )
    .await
}

async fn postgres_table_exists<C>(client: &C, schema: &str, table: &str) -> Result<bool>
where
    C: GenericClient + Sync,
{
    let row = client
        .query_one(
            crate::catalog_table_exists::POSTGRES_TABLE_EXISTS_SQL,
            &[&schema, &table],
        )
        .await
        .map_err(pg_err)?;
    get(&row, 0)
}

async fn upsert_postgres_table_column_stats(
    tx: &Transaction<'_>,
    schema: &str,
    stats: &TableColumnStats,
) -> Result<()> {
    crate::stats_engine::upsert_table_column_stats_by_update_async(stats, |query| async move {
        let executor = postgres_catalog_query_executor(tx, schema);
        executor.execute_catalog_command(&query).await
    })
    .await
}

fn get<T>(row: &Row, idx: usize) -> Result<T>
where
    T: FromSqlOwned,
{
    row.try_get(idx).map_err(pg_err)
}

struct PostgresCatalogQueryExecutor<'a, 'tx> {
    tx: &'a Transaction<'tx>,
    metadata_schema: &'a str,
}

fn postgres_catalog_query_executor<'a, 'tx>(
    tx: &'a Transaction<'tx>,
    metadata_schema: &'a str,
) -> PostgresCatalogQueryExecutor<'a, 'tx> {
    PostgresCatalogQueryExecutor {
        tx,
        metadata_schema,
    }
}

#[async_trait]
trait PostgresCatalogQueryAccess {
    fn postgres_catalog_metadata_schema(&self) -> &str;

    async fn query_postgres_catalog<T>(
        &self,
        sql: String,
        params: Vec<crate::catalog_queries::CatalogParam>,
        mapper: for<'row> fn(&'row dyn crate::embedded_rows::EmbeddedRow) -> Result<T>,
    ) -> Result<Vec<T>>
    where
        T: Send;

    async fn execute_postgres_catalog_command(
        &self,
        sql: String,
        params: Vec<crate::catalog_queries::CatalogParam>,
    ) -> Result<usize>;
}

#[async_trait]
impl<C> AsyncCatalogQueryExecutor for C
where
    C: PostgresCatalogQueryAccess + Sync,
{
    fn catalog_query_render(&self) -> crate::catalog_async_query_methods::CatalogQueryRender {
        crate::catalog_async_query_methods::CatalogQueryRender::Postgres
    }

    fn catalog_query_metadata_schema(&self) -> Option<&str> {
        Some(self.postgres_catalog_metadata_schema())
    }

    async fn execute_rendered_catalog_query<T>(
        &self,
        sql: String,
        params: Vec<crate::catalog_queries::CatalogParam>,
        mapper: for<'row> fn(&'row dyn crate::embedded_rows::EmbeddedRow) -> Result<T>,
    ) -> Result<Vec<T>>
    where
        T: Send,
    {
        self.query_postgres_catalog(sql, params, mapper).await
    }

    async fn execute_rendered_catalog_command(
        &self,
        sql: String,
        params: Vec<crate::catalog_queries::CatalogParam>,
    ) -> Result<usize> {
        self.execute_postgres_catalog_command(sql, params).await
    }
}

#[async_trait]
impl PostgresCatalogQueryAccess for PostgresCatalog {
    fn postgres_catalog_metadata_schema(&self) -> &str {
        &self.metadata_schema
    }

    async fn query_postgres_catalog<T>(
        &self,
        sql: String,
        params: Vec<crate::catalog_queries::CatalogParam>,
        mapper: for<'row> fn(&'row dyn crate::embedded_rows::EmbeddedRow) -> Result<T>,
    ) -> Result<Vec<T>>
    where
        T: Send,
    {
        let client = self.client().await?;
        postgres_catalog_query_rows(&client, sql, params, mapper).await
    }

    async fn execute_postgres_catalog_command(
        &self,
        sql: String,
        params: Vec<crate::catalog_queries::CatalogParam>,
    ) -> Result<usize> {
        let client = self.client().await?;
        postgres_catalog_execute_command(&client, sql, params).await
    }
}

#[async_trait]
impl PostgresCatalogQueryAccess for PostgresCatalogQueryExecutor<'_, '_> {
    fn postgres_catalog_metadata_schema(&self) -> &str {
        self.metadata_schema
    }

    async fn query_postgres_catalog<T>(
        &self,
        sql: String,
        params: Vec<crate::catalog_queries::CatalogParam>,
        mapper: for<'row> fn(&'row dyn crate::embedded_rows::EmbeddedRow) -> Result<T>,
    ) -> Result<Vec<T>>
    where
        T: Send,
    {
        postgres_catalog_query_rows(self.tx, sql, params, mapper).await
    }

    async fn execute_postgres_catalog_command(
        &self,
        sql: String,
        params: Vec<crate::catalog_queries::CatalogParam>,
    ) -> Result<usize> {
        postgres_catalog_execute_command(self.tx, sql, params).await
    }
}

#[async_trait]
impl PostgresCatalogQueryAccess for PostgresInlinedMutationOps<'_, '_> {
    fn postgres_catalog_metadata_schema(&self) -> &str {
        self.metadata_schema
    }

    async fn query_postgres_catalog<T>(
        &self,
        sql: String,
        params: Vec<crate::catalog_queries::CatalogParam>,
        mapper: for<'row> fn(&'row dyn crate::embedded_rows::EmbeddedRow) -> Result<T>,
    ) -> Result<Vec<T>>
    where
        T: Send,
    {
        postgres_catalog_query_rows(self.tx, sql, params, mapper).await
    }

    async fn execute_postgres_catalog_command(
        &self,
        sql: String,
        params: Vec<crate::catalog_queries::CatalogParam>,
    ) -> Result<usize> {
        postgres_catalog_execute_command(self.tx, sql, params).await
    }
}

async fn postgres_catalog_query_rows<C, T>(
    client: &C,
    sql: String,
    params: Vec<crate::catalog_queries::CatalogParam>,
    mapper: for<'row> fn(&'row dyn crate::embedded_rows::EmbeddedRow) -> Result<T>,
) -> Result<Vec<T>>
where
    C: GenericClient + Sync,
    T: Send,
{
    let params = postgres_catalog_params(&params);
    let param_refs = params
        .iter()
        .map(|param| param.as_ref() as &(dyn ToSql + Sync))
        .collect::<Vec<_>>();
    let rows = client.query(&sql, &param_refs).await.map_err(pg_err)?;
    crate::embedded_rows::collect_embedded_rows(rows.iter(), mapper)
}

async fn postgres_catalog_execute_command<C>(
    client: &C,
    sql: String,
    params: Vec<crate::catalog_queries::CatalogParam>,
) -> Result<usize>
where
    C: GenericClient + Sync,
{
    let params = postgres_catalog_params(&params);
    let param_refs = params
        .iter()
        .map(|param| param.as_ref() as &(dyn ToSql + Sync))
        .collect::<Vec<_>>();
    let updated = client.execute(&sql, &param_refs).await.map_err(pg_err)?;
    Ok(usize::try_from(updated).unwrap_or(usize::MAX))
}

fn postgres_catalog_params(
    params: &[crate::catalog_queries::CatalogParam],
) -> Vec<Box<dyn ToSql + Sync + Send>> {
    params
        .iter()
        .map(|param| match param {
            crate::catalog_queries::CatalogParam::I64(value) => {
                Box::new(*value) as Box<dyn ToSql + Sync + Send>
            }
            crate::catalog_queries::CatalogParam::String(value) => {
                Box::new(value.clone()) as Box<dyn ToSql + Sync + Send>
            }
            crate::catalog_queries::CatalogParam::OptI64(value) => {
                Box::new(*value) as Box<dyn ToSql + Sync + Send>
            }
            crate::catalog_queries::CatalogParam::OptString(value) => {
                Box::new(value.clone()) as Box<dyn ToSql + Sync + Send>
            }
            crate::catalog_queries::CatalogParam::OptBool(value) => {
                Box::new(*value) as Box<dyn ToSql + Sync + Send>
            }
            crate::catalog_queries::CatalogParam::Bool(value) => {
                Box::new(*value) as Box<dyn ToSql + Sync + Send>
            }
        })
        .collect()
}

fn pg_err(err: tokio_postgres::Error) -> StorageError {
    if let Some(db) = err.as_db_error() {
        let mut message = db.message().to_string();
        if let Some(detail) = db.detail() {
            message.push_str(": ");
            message.push_str(detail);
        }
        if let Some(table) = db.table() {
            message.push_str(" [table ");
            message.push_str(table);
            message.push(']');
        }
        if let Some(constraint) = db.constraint() {
            message.push_str(" [constraint ");
            message.push_str(constraint);
            message.push(']');
        }
        return StorageError::Backend(message);
    }
    StorageError::Backend(err.to_string())
}

#[cfg(test)]
#[path = "postgres/tests.rs"]
mod tests;
