drizzle 0.1.16

//! Synchronous `PostgreSQL` driver using [`postgres`].
//!
//! # Quick start
//!
//! ```no_run
//! use drizzle::postgres::prelude::*;
//! use drizzle::postgres::sync::Drizzle;
//!
//! #[PostgresTable]
//! struct User {
//!     #[column(serial, primary)]
//!     id: i32,
//!     name: String,
//! }
//!
//! #[derive(PostgresSchema)]
//! struct AppSchema {
//!     user: User,
//! }
//!
//! fn main() -> drizzle::Result<()> {
//!     let client = ::postgres::Client::connect("host=localhost user=postgres", ::postgres::NoTls)?;
//!     let (mut db, AppSchema { user }) = Drizzle::new(client, AppSchema::new());
//!     db.create()?;
//!
//!     // Insert
//!     db.insert(user).values([InsertUser::new("Alice")]).execute()?;
//!
//!     // Select
//!     let users: Vec<SelectUser> = db.select(()).from(user).all()?;
//!
//!     Ok(())
//! }
//! ```
//!
//! # Transactions
//!
//! Return `Ok(value)` to commit, `Err(...)` to rollback. Panics also trigger
//! a rollback.
//!
//! ```no_run
//! # use drizzle::postgres::prelude::*;
//! # use drizzle::postgres::sync::Drizzle;
//! # #[PostgresTable] struct User { #[column(serial, primary)] id: i32, name: String }
//! # #[derive(PostgresSchema)] struct S { user: User }
//! # fn main() -> drizzle::Result<()> {
//! # let client = ::postgres::Client::connect("host=localhost user=postgres", ::postgres::NoTls)?;
//! # let (mut db, S { user }) = Drizzle::new(client, S::new());
//! use drizzle::postgres::common::PostgresTransactionType;
//!
//! let count = db.transaction(PostgresTransactionType::ReadCommitted, |tx| {
//!     tx.insert(user).values([InsertUser::new("Alice")]).execute()?;
//!     let users: Vec<SelectUser> = tx.select(()).from(user).all()?;
//!     Ok(users.len())
//! })?;
//! # Ok(()) }
//! ```
//!
//! # Savepoints
//!
//! Savepoints nest inside transactions — a failed savepoint rolls back
//! without aborting the outer transaction.
//!
//! ```no_run
//! # use drizzle::postgres::prelude::*;
//! # use drizzle::postgres::sync::Drizzle;
//! # use drizzle::postgres::common::PostgresTransactionType;
//! # #[PostgresTable] struct User { #[column(serial, primary)] id: i32, name: String }
//! # #[derive(PostgresSchema)] struct S { user: User }
//! # fn main() -> drizzle::Result<()> {
//! # let client = ::postgres::Client::connect("host=localhost user=postgres", ::postgres::NoTls)?;
//! # let (mut db, S { user }) = Drizzle::new(client, S::new());
//! db.transaction(PostgresTransactionType::ReadCommitted, |tx| {
//!     tx.insert(user).values([InsertUser::new("Alice")]).execute()?;
//!
//!     // This savepoint fails — only its changes roll back
//!     let _: Result<(), _> = tx.savepoint(|stx| {
//!         stx.insert(user).values([InsertUser::new("Bad")]).execute()?;
//!         Err(drizzle::error::DrizzleError::Other("oops".into()))
//!     });
//!
//!     let users: Vec<SelectUser> = tx.select(()).from(user).all()?;
//!     assert_eq!(users.len(), 1); // only Alice
//!     Ok(())
//! })?;
//! # Ok(()) }
//! ```
//!
//! # Prepared statements
//!
//! Build a query once and execute it many times with different parameters.
//!
//! ```no_run
//! # use drizzle::postgres::prelude::*;
//! # use drizzle::postgres::sync::Drizzle;
//! # use drizzle::core::expr::eq;
//! # #[PostgresTable] struct User { #[column(serial, primary)] id: i32, name: String }
//! # #[derive(PostgresSchema)] struct S { user: User }
//! # fn main() -> drizzle::Result<()> {
//! # let client = ::postgres::Client::connect("host=localhost user=postgres", ::postgres::NoTls)?;
//! # let (mut db, S { user }) = Drizzle::new(client, S::new());
//!
//! let find_name = user.name.placeholder("find_name");
//!
//! let find_user = db
//!     .select(())
//!     .from(user)
//!     .r#where(eq(user.name, find_name))
//!     .prepare()
//!     .into_owned();
//!
//! let alice: Vec<SelectUser> = find_user
//!     .all(db.conn_mut(), [find_name.bind("Alice")])?;
//! # Ok(()) }
//! ```

pub(crate) mod prepared;

use core::marker::PhantomData;
use std::sync::OnceLock;

use drizzle_core::error::{DrizzleError, QueryContext, ResultExt};
use drizzle_core::prepared::prepare_render;
use drizzle_core::traits::ToSQL;
use drizzle_postgres::builder::{DeleteInitial, InsertInitial, SelectInitial, UpdateInitial};
use drizzle_postgres::traits::PostgresTable;
use postgres::{
    Client, IsolationLevel, Row, Statement,
    types::{ToSql, Type},
};

use drizzle_postgres::builder::{
    self, QueryBuilder, delete::DeleteBuilder, insert::InsertBuilder, select::SelectBuilder,
    update::UpdateBuilder,
};
use drizzle_postgres::common::PostgresTransactionType;
use drizzle_postgres::values::PostgresValue;
use smallvec::SmallVec;

use crate::builder::postgres::common;
use crate::builder::postgres::rows::DecodeRows;
use crate::transaction::savepoint::sync_transaction;

/// Postgres-specific drizzle builder
pub type DrizzleBuilder<'a, Schema, Builder, State> =
    common::DrizzleBuilder<'a, &'a mut Drizzle<Schema>, Schema, Builder, State>;

use crate::transaction::postgres::postgres_sync::Transaction;

#[cfg(feature = "query")]
impl<Schema> common::RelationalPreparedDriver for &mut Drizzle<Schema> {
    type PreparedDriver = Client;
}

crate::drizzle_prepare_impl!();

/// Synchronous `PostgreSQL` database wrapper using [`postgres::Client`].
///
/// Provides query building methods (`select`, `insert`, `update`, `delete`)
/// and execution methods (`execute`, `all`, `get`, `transaction`).
pub struct Drizzle<Schema = ()> {
    client: Client,
    schema: Schema,
    client_id: OnceLock<u64>,
    statement_cache: OnceLock<prepared::StatementCache>,
}

/// Lazy decoded row cursor for postgres sync queries.
pub type Rows<R> = DecodeRows<Row, R>;

pub(crate) fn postgres_sync_materialize_params<'p>(
    params: &[&'p PostgresValue<'_>],
) -> (SmallVec<[Type; 8]>, SmallVec<[&'p (dyn ToSql + Sync); 8]>) {
    let mut param_types = SmallVec::with_capacity(params.len());
    let mut param_refs = SmallVec::with_capacity(params.len());
    let mut collect_types = true;

    for &param in params {
        param_refs.push(param as &(dyn ToSql + Sync));
        if collect_types {
            if let Some(ty) =
                crate::builder::postgres::prepared_common::postgres_sync_param_type(param)
            {
                param_types.push(ty);
            } else {
                param_types.clear();
                collect_types = false;
            }
        }
    }

    (param_types, param_refs)
}

impl Drizzle {
    /// Creates a new `Drizzle` instance.
    ///
    /// Returns a tuple of (Drizzle, Schema) for destructuring.
    #[inline]
    pub const fn new<S: Copy>(client: Client, schema: S) -> (Drizzle<S>, S) {
        let drizzle = Drizzle {
            client,
            schema,
            client_id: OnceLock::new(),
            statement_cache: OnceLock::new(),
        };
        (drizzle, schema)
    }
}

impl<Schema> Drop for Drizzle<Schema> {
    fn drop(&mut self) {
        if let Some(client_id) = self.client_id.get() {
            prepared::unregister_client(*client_id);
        }
    }
}

impl<S> AsRef<Self> for Drizzle<S> {
    #[inline]
    fn as_ref(&self) -> &Self {
        self
    }
}

impl<Schema> Drizzle<Schema> {
    fn client_id(&self) -> u64 {
        *self.client_id.get_or_init(prepared::next_client_id)
    }

    fn statement_cache(&self) -> prepared::StatementCache {
        self.statement_cache
            .get_or_init(prepared::StatementCache::default)
            .clone()
    }

    fn cached_statement(
        &mut self,
        sql: &str,
        param_types: &[Type],
    ) -> Result<Statement, postgres::Error> {
        let client_id = self.client_id();
        prepared::register_client(&self.client, client_id);
        let statement_cache = self.statement_cache();
        statement_cache.statement(&mut self.client, sql, param_types)
    }

    /// Gets a reference to the underlying connection.
    #[inline]
    pub const fn conn(&self) -> &Client {
        &self.client
    }

    /// Gets a mutable reference to the underlying connection.
    #[inline]
    pub fn conn_mut(&mut self) -> &mut Client {
        let client_id = self.client_id();
        prepared::register_client(&self.client, client_id);
        &mut self.client
    }

    /// Gets a reference to the schema.
    #[inline]
    pub const fn schema(&self) -> &Schema {
        &self.schema
    }

    postgres_builder_constructors!(mut);

    /// Execute a statement and return the number of affected rows.
    ///
    /// # Errors
    ///
    /// Returns a [`postgres::Error`] if the database connection fails or the SQL is invalid.
    pub fn execute<'a, T>(&'a mut self, query: T) -> Result<u64, postgres::Error>
    where
        T: ToSQL<'a, PostgresValue<'a>>,
    {
        #[cfg(feature = "profiling")]
        drizzle_core::drizzle_profile_scope!("postgres.sync", "drizzle.execute");
        let query = query.to_sql();
        #[cfg(feature = "profiling")]
        drizzle_core::drizzle_profile_scope!("postgres.sync", "drizzle.execute.build");
        let (sql, params) = query.build();
        drizzle_core::drizzle_trace_query!(&sql, params.len());

        #[cfg(feature = "profiling")]
        drizzle_core::drizzle_profile_scope!("postgres.sync", "drizzle.execute.param_refs");
        let (param_types, param_refs) = postgres_sync_materialize_params(&params);

        #[cfg(feature = "profiling")]
        drizzle_core::drizzle_profile_scope!("postgres.sync", "drizzle.execute.db");
        let statement = self.cached_statement(&sql, &param_types)?;
        self.client.execute(&statement, &param_refs[..])
    }

    /// Runs the query and returns all matching rows (for SELECT queries)
    ///
    /// # Errors
    ///
    /// Returns [`DrizzleError`] if the query fails or row decoding fails.
    pub fn all<'a, T, R, C>(&'a mut self, query: T) -> drizzle_core::error::Result<C>
    where
        R: for<'r> TryFrom<&'r Row>,
        for<'r> <R as TryFrom<&'r Row>>::Error: Into<drizzle_core::error::DrizzleError>,
        T: ToSQL<'a, PostgresValue<'a>>,
        C: std::iter::FromIterator<R>,
    {
        self.rows(query)?
            .collect::<drizzle_core::error::Result<C>>()
    }

    /// Runs the query and returns a lazy row cursor.
    ///
    /// # Errors
    ///
    /// Returns [`DrizzleError`] if the query fails.
    pub fn rows<'a, T, R>(&'a mut self, query: T) -> drizzle_core::error::Result<Rows<R>>
    where
        R: for<'r> TryFrom<&'r Row>,
        for<'r> <R as TryFrom<&'r Row>>::Error: Into<drizzle_core::error::DrizzleError>,
        T: ToSQL<'a, PostgresValue<'a>>,
    {
        #[cfg(feature = "profiling")]
        drizzle_core::drizzle_profile_scope!("postgres.sync", "drizzle.all");
        let sql = query.to_sql();
        #[cfg(feature = "profiling")]
        drizzle_core::drizzle_profile_scope!("postgres.sync", "drizzle.all.build");
        let (sql_str, params) = sql.build();
        drizzle_core::drizzle_trace_query!(&sql_str, params.len());

        #[cfg(feature = "profiling")]
        drizzle_core::drizzle_profile_scope!("postgres.sync", "drizzle.all.param_refs");
        let (param_types, param_refs) = postgres_sync_materialize_params(&params);
        let statement = self
            .cached_statement(&sql_str, &param_types)
            .with_query(|| QueryContext::new(&sql_str, &params))?;

        let rows = self
            .client
            .query(&statement, &param_refs[..])
            .with_query(|| QueryContext::new(&sql_str, &params))?;

        Ok(Rows::new(rows))
    }

    /// Runs the query and returns a single row (for SELECT queries)
    ///
    /// # Errors
    ///
    /// Returns [`DrizzleError`] if the query fails, no rows match, or row decoding fails.
    pub fn get<'a, T, R>(&'a mut self, query: T) -> drizzle_core::error::Result<R>
    where
        R: for<'r> TryFrom<&'r Row>,
        for<'r> <R as TryFrom<&'r Row>>::Error: Into<drizzle_core::error::DrizzleError>,
        T: ToSQL<'a, PostgresValue<'a>>,
    {
        #[cfg(feature = "profiling")]
        drizzle_core::drizzle_profile_scope!("postgres.sync", "drizzle.get");
        let sql = query.to_sql();
        #[cfg(feature = "profiling")]
        drizzle_core::drizzle_profile_scope!("postgres.sync", "drizzle.get.build");
        let (sql_str, params) = sql.build();
        drizzle_core::drizzle_trace_query!(&sql_str, params.len());

        #[cfg(feature = "profiling")]
        drizzle_core::drizzle_profile_scope!("postgres.sync", "drizzle.get.param_refs");
        let (param_types, param_refs) = postgres_sync_materialize_params(&params);
        let statement = self
            .cached_statement(&sql_str, &param_types)
            .with_query(|| QueryContext::new(&sql_str, &params))?;

        let row = self
            .client
            .query_one(&statement, &param_refs[..])
            .with_query(|| QueryContext::new(&sql_str, &params))?;

        R::try_from(&row).map_err(Into::into)
    }

    /// Creates a relational query builder for the given table.
    #[cfg(feature = "query")]
    pub fn query<'a, T>(
        &mut self,
        _table: T,
    ) -> common::DrizzleQueryBuilder<'_, 'a, &mut Self, Schema, T>
    where
        T: drizzle_core::query::QueryTable,
    {
        common::DrizzleQueryBuilder {
            runner: self,
            builder: drizzle_core::query::QueryBuilder::new(),
            _schema: PhantomData,
        }
    }

    /// Executes a transaction with the given callback.
    ///
    /// The transaction is committed when the callback returns `Ok` and
    /// rolled back on `Err` or panic.
    ///
    /// ```no_run
    /// # use drizzle::postgres::prelude::*;
    /// # use drizzle::postgres::sync::Drizzle;
    /// # use drizzle::postgres::common::PostgresTransactionType;
    /// # #[PostgresTable] struct User { #[column(serial, primary)] id: i32, name: String }
    /// # #[derive(PostgresSchema)] struct S { user: User }
    /// # fn main() -> drizzle::Result<()> {
    /// # let client = ::postgres::Client::connect("host=localhost user=postgres", ::postgres::NoTls)?;
    /// # let (mut db, S { user }) = Drizzle::new(client, S::new());
    /// let count = db.transaction(PostgresTransactionType::ReadCommitted, |tx| {
    ///     tx.insert(user).values([InsertUser::new("Alice")]).execute()?;
    ///     let users: Vec<SelectUser> = tx.select(()).from(user).all()?;
    ///     Ok(users.len())
    /// })?;
    /// # Ok(()) }
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`DrizzleError`] if beginning or committing the transaction fails, or if the inner closure returns an error.
    pub fn transaction<F, R>(
        &mut self,
        tx_type: PostgresTransactionType,
        f: F,
    ) -> drizzle_core::error::Result<R>
    where
        Schema: Copy,
        F: FnOnce(&Transaction<Schema>) -> drizzle_core::error::Result<R>,
    {
        // Resolved before the transaction borrows the client mutably; sharing the
        // connection's identity and cache lets statements prepared inside the
        // transaction serve later transactions and the connection runner.
        let client_id = self.client_id();
        let statement_cache = self.statement_cache();

        let builder = self.client.build_transaction();
        let builder = if tx_type == PostgresTransactionType::default() {
            builder
        } else {
            let isolation = match tx_type {
                PostgresTransactionType::ReadUncommitted => IsolationLevel::ReadUncommitted,
                PostgresTransactionType::ReadCommitted => IsolationLevel::ReadCommitted,
                PostgresTransactionType::RepeatableRead => IsolationLevel::RepeatableRead,
                PostgresTransactionType::Serializable => IsolationLevel::Serializable,
            };
            builder.isolation_level(isolation)
        };
        drizzle_core::drizzle_trace_tx!("begin", "postgres.sync");
        let tx = builder.start()?;

        let transaction = Transaction::new(tx, tx_type, self.schema, client_id, statement_cache);
        sync_transaction(
            transaction,
            "postgres.sync",
            || {
                drizzle_core::drizzle_trace_tx!("commit", "postgres.sync");
            },
            || {
                drizzle_core::drizzle_trace_tx!("rollback", "postgres.sync");
            },
            |transaction| f(transaction),
            |transaction| transaction.commit(),
            |transaction| transaction.rollback(),
        )
    }
}

impl<Schema> Drizzle<Schema>
where
    Schema: drizzle_core::traits::SQLSchemaImpl + Default,
{
    /// Create schema objects from `SQLSchemaImpl`.
    ///
    /// # Errors
    ///
    /// Returns [`DrizzleError`] if any CREATE statement fails to execute.
    pub fn create(&mut self) -> drizzle_core::error::Result<()> {
        let schema = Schema::default();
        let statements = schema.create_statements()?;

        for statement in statements {
            self.client.execute(&statement, &[])?;
        }

        Ok(())
    }
}

impl<Schema> Drizzle<Schema> {
    /// Apply pending migrations from an embedded migration slice.
    ///
    /// Creates the drizzle schema if needed and runs pending migrations in a transaction.
    ///
    /// # Errors
    ///
    /// Returns [`DrizzleError`] if the underlying migration fails.
    /// # Two-phase tracking on the CONCURRENTLY path
    ///
    /// A migration containing `CREATE/DROP INDEX CONCURRENTLY` cannot run
    /// inside a transaction, so it executes autocommit statement-by-statement.
    /// That path uses two-phase tracking: the tracking row is inserted with
    /// `applied_at` NULL *before* the statements run (marking the migration
    /// **dirty**) and updated to set `applied_at` after they succeed. A crash
    /// in between leaves a dirty row instead of an untracked partial schema,
    /// and the next `migrate()` says which migration was interrupted rather
    /// than re-running its DDL. See [`Self::migrate_with_repair`].
    ///
    /// Non-concurrent migrations keep the single-transaction flow (statements
    /// and tracking insert commit together), which is already atomic.
    pub fn migrate(
        &mut self,
        migrations: &[drizzle_migrations::Migration],
        tracking: drizzle_migrations::Tracking,
    ) -> drizzle_core::error::Result<drizzle_migrations::MigrateOutcome> {
        self.migrate_inner(migrations, tracking, false)
    }

    /// Apply pending migrations, first reconciling any interrupted migration.
    ///
    /// For each migration marked dirty by the two-phase CONCURRENTLY flow in
    /// [`Self::migrate`], this introspects the live catalogs and classifies
    /// every statement: `CREATE TABLE`, `CREATE [UNIQUE] INDEX [CONCURRENTLY]`,
    /// `CREATE VIEW` and `CREATE TYPE ... AS ENUM` statements whose object
    /// already exists with a matching definition are skipped, and the rest are
    /// executed. Anything that cannot be proven either way aborts with a list
    /// of what needs manual resolution.
    ///
    /// # Errors
    ///
    /// Returns [`DrizzleError`] if repair cannot reconcile every statement, or
    /// if any remaining statement fails.
    pub fn migrate_with_repair(
        &mut self,
        migrations: &[drizzle_migrations::Migration],
        tracking: drizzle_migrations::Tracking,
    ) -> drizzle_core::error::Result<drizzle_migrations::MigrateOutcome> {
        self.migrate_inner(migrations, tracking, true)
    }

    fn migrate_inner(
        &mut self,
        migrations: &[drizzle_migrations::Migration],
        tracking: drizzle_migrations::Tracking,
        repair: bool,
    ) -> drizzle_core::error::Result<drizzle_migrations::MigrateOutcome> {
        let set = drizzle_migrations::Migrations::with_tracking(
            migrations.to_vec(),
            drizzle_types::Dialect::PostgreSQL,
            tracking,
        );

        if let Some(schema_sql) = set.create_schema_sql() {
            self.client.execute(&schema_sql, &[])?;
        }
        let lock_key = set.postgres_advisory_lock_key();
        self.client
            .query_one("SELECT pg_advisory_lock($1)", &[&lock_key])?;

        let result = (|| {
            ensure_postgres_migration_table(&mut self.client, &set)?;
            let mut applied = repair_dirty_migrations(&mut self.client, &set, repair)?;

            let rows = self.client.query(&set.applied_names_sql(), &[])?;
            let applied_names = rows
                .iter()
                .map(|row| row.try_get::<_, String>(0))
                .collect::<Result<Vec<_>, postgres::Error>>()?;
            let pending: Vec<_> = set.pending(&applied_names).collect();

            if pending.is_empty() {
                if applied.is_empty() {
                    return Ok(drizzle_migrations::MigrateOutcome::UpToDate);
                }
                return Ok(drizzle_migrations::MigrateOutcome::Applied { tags: applied });
            }

            for migration in &pending {
                if migration.has_postgres_concurrent_index() {
                    // CREATE/DROP INDEX CONCURRENTLY cannot run inside a
                    // transaction; only this migration runs autocommit, so it
                    // needs the dirty marker to stay recoverable.
                    self.client
                        .execute(&set.record_migration_started_sql(migration), &[])?;

                    let mut executed = 0usize;
                    for statement in migration.statements() {
                        if statement.trim().is_empty() {
                            continue;
                        }
                        if let Err(error) = self.client.execute(statement, &[]) {
                            // Nothing can have been applied yet, so drop the
                            // marker rather than demand a pointless repair.
                            if executed == 0 {
                                let _ = self
                                    .client
                                    .execute(&set.clear_migration_started_sql(migration), &[]);
                            }
                            return Err(error.into());
                        }
                        executed += 1;
                    }

                    self.client
                        .execute(&set.record_migration_finished_sql(migration), &[])?;
                } else {
                    let mut tx = self.client.transaction()?;
                    for statement in migration.statements() {
                        if !statement.trim().is_empty() {
                            tx.execute(statement, &[])?;
                        }
                    }
                    tx.execute(&set.record_migration_sql(migration), &[])?;
                    tx.commit()?;
                }
                applied.push(migration.tag().to_string());
            }

            Ok(drizzle_migrations::MigrateOutcome::Applied { tags: applied })
        })();

        let unlock_result = self
            .client
            .query_one("SELECT pg_advisory_unlock($1)", &[&lock_key])
            .map(|_| ());

        match (result, unlock_result) {
            (Ok(outcome), Ok(())) => Ok(outcome),
            (Err(error), _) => Err(error),
            (Ok(_), Err(error)) => Err(error.into()),
        }
    }
}

fn ensure_postgres_migration_table(
    client: &mut postgres::Client,
    set: &drizzle_migrations::Migrations,
) -> drizzle_core::error::Result<()> {
    client.execute(&set.create_table_sql(), &[])?;

    let schema = set.schema_name().unwrap_or("public");
    let rows = client.query(
        "SELECT column_name FROM information_schema.columns WHERE table_schema = $1 AND table_name = $2",
        &[&schema, &set.table_name()],
    )?;
    let column_names = rows
        .iter()
        .map(|row| row.try_get::<_, String>(0))
        .collect::<Result<Vec<_>, postgres::Error>>()?;
    if column_names.iter().any(|column| column == "name") {
        return Ok(());
    }

    let rows = client.query(
        &format!(
            "SELECT id, hash, created_at FROM {} ORDER BY id ASC",
            set.table_ident_sql()
        ),
        &[],
    )?;
    let applied = rows
        .iter()
        .map(|row| {
            Ok(drizzle_migrations::AppliedMigrationMetadata {
                id: row.try_get::<_, Option<i64>>(0).ok().flatten(),
                hash: row.try_get::<_, String>(1)?,
                created_at: row.try_get::<_, i64>(2)?,
            })
        })
        .collect::<Result<Vec<_>, postgres::Error>>()?;

    let matched = drizzle_migrations::match_applied_migration_metadata(set.all(), &applied)
        .map_err(|e| drizzle_core::error::DrizzleError::Other(e.to_string().into()))?;

    client.execute(
        &format!(
            "ALTER TABLE {} ADD COLUMN \"name\" TEXT",
            set.table_ident_sql()
        ),
        &[],
    )?;
    client.execute(
        &format!(
            "ALTER TABLE {} ADD COLUMN \"applied_at\" TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP",
            set.table_ident_sql()
        ),
        &[],
    )?;

    for row in matched {
        client.execute(&set.backfill_migration_metadata_sql(&row), &[])?;
    }

    Ok(())
}

/// Read the live PostgreSQL catalogs into a repair [`Catalog`](drizzle_migrations::repair::Catalog).
fn introspect_catalog(
    client: &mut postgres::Client,
) -> drizzle_core::error::Result<drizzle_migrations::repair::Catalog> {
    use drizzle_migrations::repair::postgres as pg;

    let two = |client: &mut postgres::Client,
               sql: &str|
     -> Result<Vec<(String, String)>, postgres::Error> {
        Ok(client
            .query(sql, &[])?
            .iter()
            .map(|row| (row.get::<_, String>(0), row.get::<_, String>(1)))
            .collect())
    };
    let three = |client: &mut postgres::Client,
                 sql: &str|
     -> Result<Vec<(String, String, String)>, postgres::Error> {
        Ok(client
            .query(sql, &[])?
            .iter()
            .map(|row| {
                (
                    row.get::<_, String>(0),
                    row.get::<_, String>(1),
                    row.get::<_, String>(2),
                )
            })
            .collect())
    };

    let indexes = client
        .query(pg::INDEXES_QUERY, &[])?
        .iter()
        .map(|row| {
            (
                row.get::<_, String>(0),
                row.get::<_, String>(1),
                row.get::<_, bool>(2),
            )
        })
        .collect::<Vec<_>>();

    let tables = two(client, pg::TABLES_QUERY)?;
    let columns = three(client, pg::COLUMNS_QUERY)?;
    let views = two(client, pg::VIEWS_QUERY)?;
    let index_columns = three(client, pg::INDEX_COLUMNS_QUERY)?;
    let enums = three(client, pg::ENUMS_QUERY)?;

    Ok(pg::catalog(
        &tables,
        &columns,
        &views,
        &indexes,
        &index_columns,
        &enums,
    ))
}

/// Reject or reconcile interrupted migrations. Returns the repaired tags.
fn repair_dirty_migrations(
    client: &mut postgres::Client,
    set: &drizzle_migrations::Migrations,
    repair: bool,
) -> drizzle_core::error::Result<Vec<String>> {
    let dirty = client
        .query(&set.dirty_names_sql(), &[])?
        .iter()
        .map(|row| row.try_get::<_, String>(0))
        .collect::<Result<Vec<_>, postgres::Error>>()?;

    if dirty.is_empty() {
        return Ok(Vec::new());
    }

    let migrator_error =
        |error: drizzle_migrations::MigratorError| DrizzleError::Other(error.to_string().into());

    if !repair {
        return Err(migrator_error(
            set.interrupted_migration_error(&dirty)
                .expect("dirty list is non-empty"),
        ));
    }

    let table_ident = set.table_ident_sql();
    let dirty_migrations = set
        .resolve_dirty_migrations(&dirty)
        .map_err(migrator_error)?
        .into_iter()
        .cloned()
        .collect::<Vec<_>>();

    let mut repaired = Vec::new();
    for migration in &dirty_migrations {
        let catalog = introspect_catalog(client)?;
        let plan = drizzle_migrations::repair::plan(
            drizzle_types::Dialect::PostgreSQL,
            migration,
            &catalog,
        );
        for statement in plan.into_executable(&table_ident).map_err(migrator_error)? {
            client.execute(statement.as_str(), &[])?;
        }
        client.execute(&set.record_migration_finished_sql(migration), &[])?;
        repaired.push(migration.tag().to_string());
    }

    Ok(repaired)
}

fn pg_sync_err(msg: &str, e: &postgres::Error) -> DrizzleError {
    // `postgres::Error`'s Display is just "db error"; the server's actual
    // message lives in the DbError source.
    match e.as_db_error() {
        Some(db) => DrizzleError::Other(format!("{msg}: {db}").into()),
        None => DrizzleError::Other(format!("{msg}: {e}").into()),
    }
}

fn pg_sync_query_schemas(
    client: &mut postgres::Client,
) -> drizzle_core::error::Result<Vec<drizzle_migrations::postgres::ddl::Schema>> {
    use drizzle_migrations::postgres::ddl::Schema as PgSchema;
    use drizzle_migrations::postgres::introspect::queries;

    Ok(client
        .query(queries::SCHEMAS_QUERY, &[])
        .map_err(|e| pg_sync_err("Failed to query schemas", &e))?
        .into_iter()
        .map(|row| PgSchema::new(row.get::<_, String>(0)))
        .collect())
}

fn pg_sync_query_tables(
    client: &mut postgres::Client,
) -> drizzle_core::error::Result<Vec<drizzle_migrations::postgres::introspect::RawTableInfo>> {
    use drizzle_migrations::postgres::introspect::{RawTableInfo, queries};

    Ok(client
        .query(queries::TABLES_QUERY, &[])
        .map_err(|e| pg_sync_err("Failed to query tables", &e))?
        .into_iter()
        .map(|row| RawTableInfo {
            schema: row.get(0),
            name: row.get(1),
            is_rls_enabled: row.get(2),
            is_unlogged: row.get(3),
            is_temporary: row.get(4),
            tablespace: row.get(5),
            comment: row.get(6),
        })
        .collect())
}

fn pg_sync_query_columns(
    client: &mut postgres::Client,
) -> drizzle_core::error::Result<Vec<drizzle_migrations::postgres::introspect::RawColumnInfo>> {
    use drizzle_migrations::postgres::introspect::{RawColumnInfo, queries};

    Ok(client
        .query(queries::COLUMNS_QUERY, &[])
        .map_err(|e| pg_sync_err("Failed to query columns", &e))?
        .into_iter()
        .map(|row| RawColumnInfo {
            schema: row.get(0),
            table: row.get(1),
            name: row.get(2),
            column_type: row.get(3),
            type_schema: row.get(4),
            not_null: row.get(5),
            default_value: row.get(6),
            is_identity: row.get(7),
            identity_type: row.get(8),
            is_generated: row.get(9),
            generated_expression: row.get(10),
            generated_stored: row.get(11),
            dimensions: row.get(12),
            comment: row.get(13),
            ordinal_position: row.get(14),
        })
        .collect())
}

fn pg_sync_query_enums(
    client: &mut postgres::Client,
) -> drizzle_core::error::Result<Vec<drizzle_migrations::postgres::introspect::RawEnumInfo>> {
    use drizzle_migrations::postgres::introspect::{RawEnumInfo, queries};

    Ok(client
        .query(queries::ENUMS_QUERY, &[])
        .map_err(|e| pg_sync_err("Failed to query enums", &e))?
        .into_iter()
        .map(|row| RawEnumInfo {
            schema: row.get(0),
            name: row.get(1),
            values: row.get(2),
        })
        .collect())
}

fn pg_sync_query_sequences(
    client: &mut postgres::Client,
) -> drizzle_core::error::Result<Vec<drizzle_migrations::postgres::introspect::RawSequenceInfo>> {
    use drizzle_migrations::postgres::introspect::{RawSequenceInfo, queries};

    Ok(client
        .query(queries::SEQUENCES_QUERY, &[])
        .map_err(|e| pg_sync_err("Failed to query sequences", &e))?
        .into_iter()
        .map(|row| RawSequenceInfo {
            schema: row.get(0),
            name: row.get(1),
            data_type: row.get(2),
            start_value: row.get(3),
            min_value: row.get(4),
            max_value: row.get(5),
            increment: row.get(6),
            cycle: row.get(7),
            cache_value: row.get(8),
            owned_by: row.get(9),
        })
        .collect())
}

fn pg_sync_query_views(
    client: &mut postgres::Client,
    schema_filter: Option<&[String]>,
) -> drizzle_core::error::Result<Vec<drizzle_migrations::postgres::introspect::RawViewInfo>> {
    use drizzle_migrations::postgres::introspect::{RawViewInfo, queries};

    Ok(client
        .query(queries::VIEWS_QUERY, &[&schema_filter])
        .map_err(|e| pg_sync_err("Failed to query views", &e))?
        .into_iter()
        .map(|row| RawViewInfo {
            schema: row.get(0),
            name: row.get(1),
            definition: row.get(2),
            is_materialized: row.get(3),
        })
        .collect())
}

fn pg_sync_query_indexes(
    client: &mut postgres::Client,
    schema_filter: Option<&[String]>,
) -> drizzle_core::error::Result<Vec<drizzle_migrations::postgres::introspect::RawIndexInfo>> {
    use drizzle_migrations::postgres::introspect::{RawIndexInfo, parse_index_columns, queries};

    let rows = if let Some(schemas) = schema_filter {
        client
            .query(queries::INDEXES_QUERY_FILTERED, &[&schemas])
            .map_err(|e| pg_sync_err("Failed to query indexes", &e))?
    } else {
        client
            .query(queries::INDEXES_QUERY, &[])
            .map_err(|e| pg_sync_err("Failed to query indexes", &e))?
    };
    Ok(rows
        .into_iter()
        .map(|row| RawIndexInfo {
            schema: row.get(0),
            table: row.get(1),
            name: row.get(2),
            is_unique: row.get(3),
            is_primary: row.get(4),
            method: row.get(5),
            columns: parse_index_columns(row.get(6)),
            where_clause: row.get(7),
            concurrent: false,
        })
        .collect())
}

fn pg_sync_query_foreign_keys(
    client: &mut postgres::Client,
) -> drizzle_core::error::Result<Vec<drizzle_migrations::postgres::introspect::RawForeignKeyInfo>> {
    use drizzle_migrations::postgres::introspect::{
        RawForeignKeyInfo, action_code_to_string, queries,
    };

    Ok(client
        .query(queries::FOREIGN_KEYS_QUERY, &[])
        .map_err(|e| pg_sync_err("Failed to query foreign keys", &e))?
        .into_iter()
        .map(|row| RawForeignKeyInfo {
            schema: row.get(0),
            table: row.get(1),
            name: row.get(2),
            columns: row.get(3),
            schema_to: row.get(4),
            table_to: row.get(5),
            columns_to: row.get(6),
            on_update: action_code_to_string(&row.get::<_, String>(7)),
            on_delete: action_code_to_string(&row.get::<_, String>(8)),
            deferrable: row.get(9),
            initially_deferred: row.get(10),
        })
        .collect())
}

fn pg_sync_query_primary_keys(
    client: &mut postgres::Client,
) -> drizzle_core::error::Result<Vec<drizzle_migrations::postgres::introspect::RawPrimaryKeyInfo>> {
    use drizzle_migrations::postgres::introspect::{RawPrimaryKeyInfo, queries};

    Ok(client
        .query(queries::PRIMARY_KEYS_QUERY, &[])
        .map_err(|e| pg_sync_err("Failed to query primary keys", &e))?
        .into_iter()
        .map(|row| RawPrimaryKeyInfo {
            schema: row.get(0),
            table: row.get(1),
            name: row.get(2),
            columns: row.get(3),
        })
        .collect())
}

fn pg_sync_query_uniques(
    client: &mut postgres::Client,
) -> drizzle_core::error::Result<Vec<drizzle_migrations::postgres::introspect::RawUniqueInfo>> {
    use drizzle_migrations::postgres::introspect::{RawUniqueInfo, queries};

    Ok(client
        .query(queries::UNIQUES_QUERY, &[])
        .map_err(|e| pg_sync_err("Failed to query unique constraints", &e))?
        .into_iter()
        .map(|row| RawUniqueInfo {
            schema: row.get(0),
            table: row.get(1),
            name: row.get(2),
            columns: row.get(3),
            nulls_not_distinct: row.get(4),
            deferrable: row.get(5),
            initially_deferred: row.get(6),
        })
        .collect())
}

fn pg_sync_query_checks(
    client: &mut postgres::Client,
    schema_filter: Option<&[String]>,
) -> drizzle_core::error::Result<Vec<drizzle_migrations::postgres::introspect::RawCheckInfo>> {
    use drizzle_migrations::postgres::introspect::{RawCheckInfo, queries};

    let rows = if let Some(schemas) = schema_filter {
        client
            .query(queries::CHECKS_QUERY_FILTERED, &[&schemas])
            .map_err(|e| pg_sync_err("Failed to query check constraints", &e))?
    } else {
        client
            .query(queries::CHECKS_QUERY, &[])
            .map_err(|e| pg_sync_err("Failed to query check constraints", &e))?
    };
    Ok(rows
        .into_iter()
        .map(|row| RawCheckInfo {
            schema: row.get(0),
            table: row.get(1),
            name: row.get(2),
            expression: row.get(3),
        })
        .collect())
}

fn pg_sync_query_roles(
    client: &mut postgres::Client,
) -> drizzle_core::error::Result<Vec<drizzle_migrations::postgres::introspect::RawRoleInfo>> {
    use drizzle_migrations::postgres::introspect::{RawRoleInfo, queries};

    Ok(client
        .query(queries::ROLES_QUERY, &[])
        .map_err(|e| pg_sync_err("Failed to query roles", &e))?
        .into_iter()
        .map(|row| RawRoleInfo {
            name: row.get(0),
            create_db: row.get(1),
            create_role: row.get(2),
            inherit: row.get(3),
        })
        .collect())
}

fn pg_sync_query_policies(
    client: &mut postgres::Client,
) -> drizzle_core::error::Result<Vec<drizzle_migrations::postgres::introspect::RawPolicyInfo>> {
    use drizzle_migrations::postgres::introspect::{RawPolicyInfo, queries};

    Ok(client
        .query(queries::POLICIES_QUERY, &[])
        .map_err(|e| pg_sync_err("Failed to query policies", &e))?
        .into_iter()
        .map(|row| RawPolicyInfo {
            schema: row.get(0),
            table: row.get(1),
            name: row.get(2),
            as_clause: row.get(3),
            for_clause: row.get(4),
            to: row.get(5),
            using: row.get(6),
            with_check: row.get(7),
        })
        .collect())
}

impl<Schema> Drizzle<Schema> {
    /// Introspect the connected `PostgreSQL` database and return a [`Snapshot`](drizzle_migrations::schema::Snapshot).
    ///
    /// Queries the `pg_catalog` and `information_schema` to extract tables, columns,
    /// indexes, foreign keys, primary keys, unique/check constraints, enums, sequences,
    /// views, roles, and policies.
    ///
    /// # Errors
    ///
    /// Returns [`DrizzleError`] if the underlying introspection queries fail.
    pub fn introspect(
        &mut self,
    ) -> drizzle_core::error::Result<drizzle_migrations::schema::Snapshot> {
        self.introspect_impl(None)
    }

    /// Inner introspection with optional schema filter.
    ///
    /// When `schema_filter` is `Some`, queries that use `pg_get_indexdef()` or
    /// `pg_get_expr()` are scoped to those schemas.  These functions call
    /// `relation_open()` which is not MVCC-protected and can fail when
    /// concurrent DDL drops objects in other schemas.
    fn introspect_impl(
        &mut self,
        schema_filter: Option<&[String]>,
    ) -> drizzle_core::error::Result<drizzle_migrations::schema::Snapshot> {
        use drizzle_migrations::postgres::ddl::Schema as PgSchema;
        use drizzle_migrations::postgres::introspect::{RawIntrospection, assemble_ddl};

        let schemas: Vec<PgSchema> = pg_sync_query_schemas(&mut self.client)?;
        let accessible_schema_names = schemas
            .iter()
            .map(|schema| schema.name().to_string())
            .collect::<Vec<_>>();
        let effective_schema_filter = schema_filter.or(Some(accessible_schema_names.as_slice()));
        let raw_tables = pg_sync_query_tables(&mut self.client)?;
        let raw_columns = pg_sync_query_columns(&mut self.client)?;
        let raw_enums = pg_sync_query_enums(&mut self.client)?;
        let raw_sequences = pg_sync_query_sequences(&mut self.client)?;
        let raw_views = pg_sync_query_views(&mut self.client, effective_schema_filter)?;
        let raw_indexes = pg_sync_query_indexes(&mut self.client, effective_schema_filter)?;
        let raw_fks = pg_sync_query_foreign_keys(&mut self.client)?;
        let raw_primary_keys = pg_sync_query_primary_keys(&mut self.client)?;
        let raw_uniques = pg_sync_query_uniques(&mut self.client)?;
        let raw_checks = pg_sync_query_checks(&mut self.client, effective_schema_filter)?;
        let raw_roles = pg_sync_query_roles(&mut self.client)?;
        let raw_policies = pg_sync_query_policies(&mut self.client)?;

        let ddl = assemble_ddl(RawIntrospection {
            schemas,
            tables: raw_tables,
            columns: raw_columns,
            enums: raw_enums,
            sequences: raw_sequences,
            views: raw_views,
            indexes: raw_indexes,
            foreign_keys: raw_fks,
            primary_keys: raw_primary_keys,
            unique_constraints: raw_uniques,
            check_constraints: raw_checks,
            roles: raw_roles,
            policies: raw_policies,
        });

        // Build snapshot
        let mut snap = drizzle_migrations::postgres::PostgresSnapshot::new();
        for entity in ddl.to_entities() {
            snap.add_entity(entity);
        }

        Ok(drizzle_migrations::schema::Snapshot::Postgres(snap))
    }

    /// Introspect the live database, diff against the desired schema, and
    /// execute the SQL statements needed to bring the database in sync.
    ///
    /// This is a no-op if the database already matches.
    ///
    /// # Errors
    ///
    /// Returns [`DrizzleError`] if introspection, diff, or applying statements fails.
    pub fn push<S: drizzle_migrations::Schema>(
        &mut self,
        schema: &S,
    ) -> drizzle_core::error::Result<()> {
        let desired = schema.to_snapshot();
        // Scope introspection to only our schemas. pg_get_indexdef() /
        // pg_get_expr() call relation_open() which is not MVCC-protected
        // and will fail if a concurrent session drops objects.
        let target_schemas: Vec<String> = match &desired {
            drizzle_migrations::schema::Snapshot::Postgres(pg) => pg.schema_names(),
            drizzle_migrations::schema::Snapshot::Sqlite(_) => Vec::new(),
        };
        let live = self.introspect_impl(if target_schemas.is_empty() {
            None
        } else {
            Some(&target_schemas)
        })?;
        let live = match (live, &desired) {
            (
                drizzle_migrations::schema::Snapshot::Postgres(live_pg),
                drizzle_migrations::schema::Snapshot::Postgres(desired_pg),
            ) => {
                drizzle_migrations::schema::Snapshot::Postgres(live_pg.prepare_for_push(desired_pg))
            }
            (other, _) => other,
        };
        let generated = drizzle_migrations::diff(&live, &desired)
            .map_err(|e| DrizzleError::Other(e.to_string().into()))?;
        for stmt in generated.statements {
            if !stmt.trim().is_empty() {
                self.client.execute(&*stmt, &[])?;
            }
        }
        Ok(())
    }
}

impl<S, Schema, State, Table, Mk, Rw, Grouped>
    DrizzleBuilder<'_, S, QueryBuilder<'_, Schema, State, Table, Mk, Rw, Grouped>, State>
where
    State: builder::ExecutableState,
{
    /// Runs the query and returns the number of affected rows
    pub fn execute(self) -> drizzle_core::error::Result<u64> {
        #[cfg(feature = "profiling")]
        drizzle_core::drizzle_profile_scope!("postgres.sync", "builder.execute");
        let (sql_str, params) = self.builder.sql.build();
        drizzle_core::drizzle_trace_query!(&sql_str, params.len());

        #[cfg(feature = "profiling")]
        drizzle_core::drizzle_profile_scope!("postgres.sync", "builder.execute.param_refs");
        let (param_types, param_refs) = postgres_sync_materialize_params(&params);
        let statement = self
            .runner
            .cached_statement(&sql_str, &param_types)
            .with_query(|| QueryContext::new(&sql_str, &params))?;

        #[cfg(feature = "profiling")]
        drizzle_core::drizzle_profile_scope!("postgres.sync", "builder.execute.db");
        self.runner
            .client
            .execute(&statement, &param_refs[..])
            .with_query(|| QueryContext::new(&sql_str, &params))
    }

    /// Runs the query and returns all matching rows using the builder's row type.
    pub fn all<R, Proof, AggProof>(self) -> drizzle_core::error::Result<Vec<R>>
    where
        for<'r> Mk: drizzle_core::row::DecodeSelectedRef<&'r ::postgres::Row, R>
            + drizzle_core::row::MarkerScopeValidFor<Proof>
            + drizzle_core::row::StrictDecodeMarker
            + drizzle_core::row::MarkerColumnCountValid<::postgres::Row, Rw, R>,
        Mk: drizzle_core::row::MarkerAggValidFor<Grouped, AggProof>,
    {
        #[cfg(feature = "profiling")]
        drizzle_core::drizzle_profile_scope!("postgres.sync", "builder.all");
        let (sql_str, params) = self.builder.sql.build();
        drizzle_core::drizzle_trace_query!(&sql_str, params.len());

        #[cfg(feature = "profiling")]
        drizzle_core::drizzle_profile_scope!("postgres.sync", "builder.all.param_refs");
        let (param_types, param_refs) = postgres_sync_materialize_params(&params);
        let statement = self
            .runner
            .cached_statement(&sql_str, &param_types)
            .with_query(|| QueryContext::new(&sql_str, &params))?;

        let rows = self
            .runner
            .client
            .query(&statement, &param_refs[..])
            .with_query(|| QueryContext::new(&sql_str, &params))?;
        let mut decoded = Vec::with_capacity(rows.len());
        for row in &rows {
            decoded.push(<Mk as drizzle_core::row::DecodeSelectedRef<
                &::postgres::Row,
                R,
            >>::decode(row)?);
        }
        Ok(decoded)
    }

    /// Runs the query and returns a lazy row cursor using the builder's row type.
    pub fn rows(self) -> drizzle_core::error::Result<Rows<Rw>>
    where
        Rw: for<'r> TryFrom<&'r Row>,
        for<'r> <Rw as TryFrom<&'r Row>>::Error: Into<drizzle_core::error::DrizzleError>,
    {
        #[cfg(feature = "profiling")]
        drizzle_core::drizzle_profile_scope!("postgres.sync", "builder.rows");
        let (sql_str, params) = self.builder.sql.build();
        drizzle_core::drizzle_trace_query!(&sql_str, params.len());

        #[cfg(feature = "profiling")]
        drizzle_core::drizzle_profile_scope!("postgres.sync", "builder.rows.param_refs");
        let (param_types, param_refs) = postgres_sync_materialize_params(&params);
        let statement = self
            .runner
            .cached_statement(&sql_str, &param_types)
            .with_query(|| QueryContext::new(&sql_str, &params))?;

        let rows = self
            .runner
            .client
            .query(&statement, &param_refs[..])
            .with_query(|| QueryContext::new(&sql_str, &params))?;

        Ok(Rows::new(rows))
    }

    /// Runs the query and returns a single row using the builder's row type.
    pub fn get<R, Proof, AggProof>(self) -> drizzle_core::error::Result<R>
    where
        for<'r> Mk: drizzle_core::row::DecodeSelectedRef<&'r ::postgres::Row, R>
            + drizzle_core::row::MarkerScopeValidFor<Proof>
            + drizzle_core::row::StrictDecodeMarker
            + drizzle_core::row::MarkerColumnCountValid<::postgres::Row, Rw, R>,
        Mk: drizzle_core::row::MarkerAggValidFor<Grouped, AggProof>,
    {
        #[cfg(feature = "profiling")]
        drizzle_core::drizzle_profile_scope!("postgres.sync", "builder.get");
        let (sql_str, params) = self.builder.sql.build();
        drizzle_core::drizzle_trace_query!(&sql_str, params.len());

        #[cfg(feature = "profiling")]
        drizzle_core::drizzle_profile_scope!("postgres.sync", "builder.get.param_refs");
        let (param_types, param_refs) = postgres_sync_materialize_params(&params);
        let statement = self
            .runner
            .cached_statement(&sql_str, &param_types)
            .with_query(|| QueryContext::new(&sql_str, &params))?;

        let row = self
            .runner
            .client
            .query_one(&statement, &param_refs[..])
            .with_query(|| QueryContext::new(&sql_str, &params))?;
        <Mk as drizzle_core::row::DecodeSelectedRef<&::postgres::Row, R>>::decode(&row)
    }
}

// =============================================================================
// Relational Query API
// =============================================================================

#[cfg(feature = "query")]
use drizzle_core::query::DeserializeStore;
#[cfg(feature = "query")]
use drizzle_core::query::FromJsonObject as _;

// AllColumns: read base from individual row columns via TryFrom<Row>
#[cfg(feature = "query")]
impl<'db, 'a, Schema, T, Rels, Cl>
    common::DrizzleQueryBuilder<
        'db,
        'a,
        &'db mut Drizzle<Schema>,
        Schema,
        T,
        Rels,
        drizzle_core::query::AllColumns,
        Cl,
    >
{
    /// Executes the query and returns all matching rows with their relations.
    pub fn find_many(
        self,
    ) -> drizzle_core::error::Result<
        Vec<
            <Rels as drizzle_core::query::BuildRow<
                <T as drizzle_core::query::QueryTable>::Select,
            >>::Row,
        >,
    >
    where
        T: drizzle_core::query::QueryTable,
        <T as drizzle_core::query::QueryTable>::Select: for<'r> TryFrom<&'r Row>,
        for<'r> <<T as drizzle_core::query::QueryTable>::Select as TryFrom<&'r Row>>::Error:
            Into<drizzle_core::error::DrizzleError>,
        Rels: drizzle_core::query::BuildRow<<T as drizzle_core::query::QueryTable>::Select>
            + drizzle_core::query::RenderRelations<'a, PostgresValue<'a>>,
        <Rels as drizzle_core::query::BuildStore>::Store: drizzle_core::query::DeserializeStore,
    {
        let num_base_cols = T::COLUMN_NAMES.len();

        let builder = self.builder;
        let mut rendered = Vec::new();
        builder.relations.render_into(&mut rendered);
        let query_sql = drizzle_core::query::build_query_sql(
            T::TABLE_NAME,
            T::COLUMN_NAMES,
            T::BLOB_COLUMNS,
            rendered,
            builder.where_sql,
            builder.order_by_sql,
            builder.limit,
            builder.offset,
            false,
        );
        let (sql, bind_params) = query_sql.build();

        drizzle_core::drizzle_trace_query!(&sql, bind_params.len());

        let (param_types, param_refs) = postgres_sync_materialize_params(&bind_params);
        let statement = self
            .runner
            .cached_statement(&sql, &param_types)
            .with_query(|| QueryContext::new(&sql, &bind_params))?;

        let rows = self
            .runner
            .client
            .query(&statement, &param_refs[..])
            .with_query(|| QueryContext::new(&sql, &bind_params))?;
        let mut results = Vec::with_capacity(rows.len());

        for row in &rows {
            let base = <T as drizzle_core::query::QueryTable>::Select::try_from(row)
                .map_err(Into::into)?;

            let mut rel_col = num_base_cols;
            let mut next_rel = || {
                let json: Option<String> = row.get(rel_col);
                rel_col += 1;
                Ok(json)
            };
            let store =
                <Rels as drizzle_core::query::BuildStore>::Store::from_json_columns(&mut next_rel)?;

            results.push(<Rels as drizzle_core::query::BuildRow<_>>::assemble(
                base, store,
            ));
        }

        Ok(results)
    }
}

// AllColumns find_first: requires no LIMIT set yet (internally adds LIMIT 1)
#[cfg(feature = "query")]
impl<'db, 'a, Schema, T, Rels, W, Ord>
    common::DrizzleQueryBuilder<
        'db,
        'a,
        &'db mut Drizzle<Schema>,
        Schema,
        T,
        Rels,
        drizzle_core::query::AllColumns,
        drizzle_core::query::Clauses<W, Ord, drizzle_core::query::NoLimit>,
    >
{
    /// Executes the query and returns the first matching row, or `None`.
    pub fn find_first(
        self,
    ) -> drizzle_core::error::Result<
        Option<
            <Rels as drizzle_core::query::BuildRow<
                <T as drizzle_core::query::QueryTable>::Select,
            >>::Row,
        >,
    >
    where
        T: drizzle_core::query::QueryTable,
        <T as drizzle_core::query::QueryTable>::Select: for<'r> TryFrom<&'r Row>,
        for<'r> <<T as drizzle_core::query::QueryTable>::Select as TryFrom<&'r Row>>::Error:
            Into<drizzle_core::error::DrizzleError>,
        Rels: drizzle_core::query::BuildRow<<T as drizzle_core::query::QueryTable>::Select>
            + drizzle_core::query::RenderRelations<'a, PostgresValue<'a>>,
        <Rels as drizzle_core::query::BuildStore>::Store: drizzle_core::query::DeserializeStore,
    {
        Ok(self.limit(1).find_many()?.into_iter().next())
    }
}

// PartialColumns: read base from a single JSON "__base" column via FromJsonObject
#[cfg(feature = "query")]
impl<'db, 'a, Schema, T, Rels, Cl>
    common::DrizzleQueryBuilder<
        'db,
        'a,
        &'db mut Drizzle<Schema>,
        Schema,
        T,
        Rels,
        drizzle_core::query::PartialColumns,
        Cl,
    >
{
    /// Executes the query and returns all matching rows with their relations.
    ///
    /// Base columns are deserialized from a JSON `"__base"` column.
    pub fn find_many(
        self,
    ) -> drizzle_core::error::Result<
        Vec<
            <Rels as drizzle_core::query::BuildRow<
                <T as drizzle_core::query::QueryTable>::PartialSelect,
            >>::Row,
        >,
    >
    where
        T: drizzle_core::query::QueryTable,
        <T as drizzle_core::query::QueryTable>::PartialSelect: drizzle_core::query::FromJsonObject,
        Rels: drizzle_core::query::BuildRow<<T as drizzle_core::query::QueryTable>::PartialSelect>
            + drizzle_core::query::RenderRelations<'a, PostgresValue<'a>>,
        <Rels as drizzle_core::query::BuildStore>::Store: drizzle_core::query::DeserializeStore,
    {
        let builder = self.builder;
        let column_names = &builder.cols.columns;
        let mut rendered = Vec::new();
        builder.relations.render_into(&mut rendered);
        let col_refs: Vec<&str> = column_names.clone();
        let query_sql = drizzle_core::query::build_query_sql(
            T::TABLE_NAME,
            &col_refs,
            T::BLOB_COLUMNS,
            rendered,
            builder.where_sql,
            builder.order_by_sql,
            builder.limit,
            builder.offset,
            true,
        );
        let (sql, bind_params) = query_sql.build();

        drizzle_core::drizzle_trace_query!(&sql, bind_params.len());

        let (param_types, param_refs) = postgres_sync_materialize_params(&bind_params);
        let statement = self
            .runner
            .cached_statement(&sql, &param_types)
            .with_query(|| QueryContext::new(&sql, &bind_params))?;

        let rows = self
            .runner
            .client
            .query(&statement, &param_refs[..])
            .with_query(|| QueryContext::new(&sql, &bind_params))?;
        let mut results = Vec::with_capacity(rows.len());

        for row in &rows {
            // Column 0 is the JSON "__base" object
            let base_json: String = row.get(0);
            let base = <T as drizzle_core::query::QueryTable>::PartialSelect::from_json_str(
                &base_json, "base",
            )?;

            let mut rel_col = 1usize;
            let mut next_rel = || {
                let json: Option<String> = row.get(rel_col);
                rel_col += 1;
                Ok(json)
            };
            let store =
                <Rels as drizzle_core::query::BuildStore>::Store::from_json_columns(&mut next_rel)?;

            results.push(<Rels as drizzle_core::query::BuildRow<_>>::assemble(
                base, store,
            ));
        }

        Ok(results)
    }
}

// PartialColumns find_first: requires no LIMIT set yet
#[cfg(feature = "query")]
impl<'db, 'a, Schema, T, Rels, W, Ord>
    common::DrizzleQueryBuilder<
        'db,
        'a,
        &'db mut Drizzle<Schema>,
        Schema,
        T,
        Rels,
        drizzle_core::query::PartialColumns,
        drizzle_core::query::Clauses<W, Ord, drizzle_core::query::NoLimit>,
    >
{
    /// Executes the query and returns the first matching row, or `None`.
    pub fn find_first(
        self,
    ) -> drizzle_core::error::Result<
        Option<
            <Rels as drizzle_core::query::BuildRow<
                <T as drizzle_core::query::QueryTable>::PartialSelect,
            >>::Row,
        >,
    >
    where
        T: drizzle_core::query::QueryTable,
        <T as drizzle_core::query::QueryTable>::PartialSelect: drizzle_core::query::FromJsonObject,
        Rels: drizzle_core::query::BuildRow<<T as drizzle_core::query::QueryTable>::PartialSelect>
            + drizzle_core::query::RenderRelations<'a, PostgresValue<'a>>,
        <Rels as drizzle_core::query::BuildStore>::Store: drizzle_core::query::DeserializeStore,
    {
        Ok(self.limit(1).find_many()?.into_iter().next())
    }
}

#[cfg(feature = "query")]
impl<'a, T, Rels>
    common::DrizzlePreparedQuery<'a, Client, T, Rels, drizzle_core::query::AllColumns>
{
    /// Executes the prepared relational query and returns all matching rows.
    pub fn find_many<const N: usize>(
        &self,
        client: &mut Client,
        params: [drizzle_core::param::ParamBind<'a, PostgresValue<'a>>; N],
    ) -> drizzle_core::error::Result<
        Vec<
            <Rels as drizzle_core::query::BuildRow<
                <T as drizzle_core::query::QueryTable>::Select,
            >>::Row,
        >,
    >
    where
        T: drizzle_core::query::QueryTable,
        <T as drizzle_core::query::QueryTable>::Select: for<'r> TryFrom<&'r Row>,
        for<'r> <<T as drizzle_core::query::QueryTable>::Select as TryFrom<&'r Row>>::Error:
            Into<drizzle_core::error::DrizzleError>,
        Rels: drizzle_core::query::BuildRow<<T as drizzle_core::query::QueryTable>::Select>,
        <Rels as drizzle_core::query::BuildStore>::Store: drizzle_core::query::DeserializeStore,
    {
        debug_assert_eq!(
            N,
            self.inner.external_param_count(),
            "parameter count mismatch: expected {} params but got {}",
            self.inner.external_param_count(),
            N
        );

        let num_base_cols = T::COLUMN_NAMES.len();
        let (sql_str, bound_params) = self.inner.bind(params)?;
        let (lower, upper) = bound_params.size_hint();
        let mut params_vec: SmallVec<[PostgresValue<'a>; 8]> =
            SmallVec::with_capacity(upper.unwrap_or(lower));
        params_vec.extend(bound_params);
        let mut param_refs: SmallVec<[&(dyn postgres::types::ToSql + Sync); 8]> =
            SmallVec::with_capacity(params_vec.len());
        for param in &params_vec {
            param_refs.push(param as &(dyn postgres::types::ToSql + Sync));
        }
        let param_types =
            crate::builder::postgres::prepared_common::postgres_sync_param_types(&params_vec);

        let statement = client.prepare_typed(sql_str, &param_types)?;
        let rows = client.query(&statement, &param_refs)?;
        let mut results = Vec::with_capacity(rows.len());

        for row in rows {
            let base = <T as drizzle_core::query::QueryTable>::Select::try_from(&row)
                .map_err(Into::into)?;

            let mut rel_col = num_base_cols;
            let mut next_rel = || {
                let json: Option<String> = row.get(rel_col);
                rel_col += 1;
                Ok(json)
            };
            let store =
                <Rels as drizzle_core::query::BuildStore>::Store::from_json_columns(&mut next_rel)?;

            results.push(<Rels as drizzle_core::query::BuildRow<_>>::assemble(
                base, store,
            ));
        }

        Ok(results)
    }

    /// Executes the prepared relational query and returns the first row, if any.
    ///
    /// To apply `LIMIT 1` in SQL, call `.limit(1)` before `.prepare()`.
    pub fn find_first<const N: usize>(
        &self,
        client: &mut Client,
        params: [drizzle_core::param::ParamBind<'a, PostgresValue<'a>>; N],
    ) -> drizzle_core::error::Result<
        Option<
            <Rels as drizzle_core::query::BuildRow<
                <T as drizzle_core::query::QueryTable>::Select,
            >>::Row,
        >,
    >
    where
        T: drizzle_core::query::QueryTable,
        <T as drizzle_core::query::QueryTable>::Select: for<'r> TryFrom<&'r Row>,
        for<'r> <<T as drizzle_core::query::QueryTable>::Select as TryFrom<&'r Row>>::Error:
            Into<drizzle_core::error::DrizzleError>,
        Rels: drizzle_core::query::BuildRow<<T as drizzle_core::query::QueryTable>::Select>,
        <Rels as drizzle_core::query::BuildStore>::Store: drizzle_core::query::DeserializeStore,
    {
        Ok(self.find_many(client, params)?.into_iter().next())
    }
}

#[cfg(feature = "query")]
impl<'a, T, Rels>
    common::DrizzlePreparedQuery<'a, Client, T, Rels, drizzle_core::query::PartialColumns>
{
    /// Executes the prepared relational query and returns all matching rows.
    pub fn find_many<const N: usize>(
        &self,
        client: &mut Client,
        params: [drizzle_core::param::ParamBind<'a, PostgresValue<'a>>; N],
    ) -> drizzle_core::error::Result<
        Vec<
            <Rels as drizzle_core::query::BuildRow<
                <T as drizzle_core::query::QueryTable>::PartialSelect,
            >>::Row,
        >,
    >
    where
        T: drizzle_core::query::QueryTable,
        <T as drizzle_core::query::QueryTable>::PartialSelect: drizzle_core::query::FromJsonObject,
        Rels: drizzle_core::query::BuildRow<<T as drizzle_core::query::QueryTable>::PartialSelect>,
        <Rels as drizzle_core::query::BuildStore>::Store: drizzle_core::query::DeserializeStore,
    {
        debug_assert_eq!(
            N,
            self.inner.external_param_count(),
            "parameter count mismatch: expected {} params but got {}",
            self.inner.external_param_count(),
            N
        );

        let (sql_str, bound_params) = self.inner.bind(params)?;
        let (lower, upper) = bound_params.size_hint();
        let mut params_vec: SmallVec<[PostgresValue<'a>; 8]> =
            SmallVec::with_capacity(upper.unwrap_or(lower));
        params_vec.extend(bound_params);
        let mut param_refs: SmallVec<[&(dyn postgres::types::ToSql + Sync); 8]> =
            SmallVec::with_capacity(params_vec.len());
        for param in &params_vec {
            param_refs.push(param as &(dyn postgres::types::ToSql + Sync));
        }
        let param_types =
            crate::builder::postgres::prepared_common::postgres_sync_param_types(&params_vec);

        let statement = client.prepare_typed(sql_str, &param_types)?;
        let rows = client.query(&statement, &param_refs)?;
        let mut results = Vec::with_capacity(rows.len());

        for row in rows {
            let base_json: String = row.get(0);
            let base = <T as drizzle_core::query::QueryTable>::PartialSelect::from_json_str(
                &base_json, "base",
            )?;

            let mut rel_col = 1usize;
            let mut next_rel = || {
                let json: Option<String> = row.get(rel_col);
                rel_col += 1;
                Ok(json)
            };
            let store =
                <Rels as drizzle_core::query::BuildStore>::Store::from_json_columns(&mut next_rel)?;

            results.push(<Rels as drizzle_core::query::BuildRow<_>>::assemble(
                base, store,
            ));
        }

        Ok(results)
    }

    /// Executes the prepared relational query and returns the first row, if any.
    ///
    /// To apply `LIMIT 1` in SQL, call `.limit(1)` before `.prepare()`.
    pub fn find_first<const N: usize>(
        &self,
        client: &mut Client,
        params: [drizzle_core::param::ParamBind<'a, PostgresValue<'a>>; N],
    ) -> drizzle_core::error::Result<
        Option<
            <Rels as drizzle_core::query::BuildRow<
                <T as drizzle_core::query::QueryTable>::PartialSelect,
            >>::Row,
        >,
    >
    where
        T: drizzle_core::query::QueryTable,
        <T as drizzle_core::query::QueryTable>::PartialSelect: drizzle_core::query::FromJsonObject,
        Rels: drizzle_core::query::BuildRow<<T as drizzle_core::query::QueryTable>::PartialSelect>,
        <Rels as drizzle_core::query::BuildStore>::Store: drizzle_core::query::DeserializeStore,
    {
        Ok(self.find_many(client, params)?.into_iter().next())
    }
}