drizzle 0.1.11

//! 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(()) }
//! ```

mod prepared;

use core::marker::PhantomData;

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::fallible_iterator::FallibleIterator;
use postgres::{Client, IsolationLevel, Row};

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;

/// 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,
}

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

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 };
        (drizzle, schema)
    }
}

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

impl<Schema> Drizzle<Schema> {
    /// 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 const fn conn_mut(&mut self) -> &mut Client {
        &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());

        let param_refs = {
            #[cfg(feature = "profiling")]
            drizzle_core::drizzle_profile_scope!("postgres.sync", "drizzle.execute.param_refs");
            let mut param_refs: SmallVec<[&(dyn postgres::types::ToSql + Sync); 8]> =
                SmallVec::with_capacity(params.len());
            param_refs.extend(
                params
                    .iter()
                    .map(|&p| p as &(dyn postgres::types::ToSql + Sync)),
            );
            param_refs
        };

        let mut typed_params: SmallVec<
            [(&(dyn postgres::types::ToSql + Sync), postgres::types::Type); 8],
        > = SmallVec::with_capacity(params.len());
        let mut all_typed = true;
        for p in &params {
            if let Some(ty) = crate::builder::postgres::prepared_common::postgres_sync_param_type(p)
            {
                typed_params.push((*p as &(dyn postgres::types::ToSql + Sync), ty));
            } else {
                all_typed = false;
                break;
            }
        }

        if all_typed {
            #[cfg(feature = "profiling")]
            drizzle_core::drizzle_profile_scope!("postgres.sync", "drizzle.execute.db_typed");
            let mut rows = self.client.query_typed_raw(&sql, typed_params)?;
            while rows.next()?.is_some() {}
            return Ok(rows.rows_affected().unwrap_or(0));
        }

        #[cfg(feature = "profiling")]
        drizzle_core::drizzle_profile_scope!("postgres.sync", "drizzle.execute.db");
        self.client.execute(&sql, &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 mut param_refs: SmallVec<[&(dyn postgres::types::ToSql + Sync); 8]> =
            SmallVec::with_capacity(params.len());
        param_refs.extend(
            params
                .iter()
                .map(|&p| p as &(dyn postgres::types::ToSql + Sync)),
        );

        let rows = self
            .client
            .query(&sql_str, &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 mut param_refs: SmallVec<[&(dyn postgres::types::ToSql + Sync); 8]> =
            SmallVec::with_capacity(params.len());
        param_refs.extend(
            params
                .iter()
                .map(|&p| p as &(dyn postgres::types::ToSql + Sync)),
        );

        let row = self
            .client
            .query_one(&sql_str, &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>,
    {
        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);

        let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| f(&transaction)));

        match result {
            Ok(callback_result) => match callback_result {
                Ok(value) => {
                    drizzle_core::drizzle_trace_tx!("commit", "postgres.sync");
                    transaction.commit()?;
                    Ok(value)
                }
                Err(e) => {
                    drizzle_core::drizzle_trace_tx!("rollback", "postgres.sync");
                    transaction.rollback()?;
                    Err(e)
                }
            },
            Err(panic_payload) => {
                drizzle_core::drizzle_trace_tx!("rollback", "postgres.sync");
                let _ = transaction.rollback();
                std::panic::resume_unwind(panic_payload);
            }
        }
    }
}

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.
    pub fn migrate(
        &mut self,
        migrations: &[drizzle_migrations::Migration],
        tracking: drizzle_migrations::Tracking,
    ) -> 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, &[])?;
        }
        ensure_postgres_migration_table(&mut self.client, &set)?;
        let rows = self.client.query(&set.applied_names_sql(), &[])?;
        let applied_names: Vec<String> = rows.iter().filter_map(|r| r.try_get(0).ok()).collect();
        let pending: Vec<_> = set.pending(&applied_names).collect();

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

        let mut tx = self.client.transaction()?;
        let mut applied = Vec::with_capacity(pending.len());

        for migration in &pending {
            for stmt in migration.statements() {
                if !stmt.trim().is_empty() {
                    tx.execute(stmt, &[])?;
                }
            }
            tx.execute(&set.record_migration_sql(migration), &[])?;
            applied.push(migration.tag().to_string());
        }

        tx.commit()?;

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

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()],
    )?;
    if rows
        .iter()
        .filter_map(|row| row.try_get::<_, String>(0).ok())
        .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 {
        let where_clause = if let Some(id) = row.id {
            format!("\"id\" = {id}")
        } else {
            format!(
                "\"created_at\" = {} AND \"hash\" = '{}'",
                row.created_at,
                row.hash.replace('\'', "''")
            )
        };
        let update_sql = format!(
            "UPDATE {} SET \"name\" = '{}', \"applied_at\" = NULL WHERE {}",
            set.table_ident_sql(),
            row.name.replace('\'', "''"),
            where_clause
        );
        client.execute(&update_sql, &[])?;
    }

    Ok(())
}

fn pg_sync_err(msg: &str, e: &postgres::Error) -> DrizzleError {
    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),
        })
        .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),
            ordinal_position: row.get(12),
        })
        .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),
        })
        .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, pg_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: pg_action_code_to_string(&row.get::<_, String>(7)),
            on_delete: pg_action_code_to_string(&row.get::<_, String>(8)),
        })
        .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),
        })
        .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::introspect::{
            process_check_constraints, process_columns, process_enums, process_foreign_keys,
            process_indexes, process_policies, process_primary_keys, process_roles,
            process_sequences, process_tables, process_unique_constraints, process_views,
        };
        use drizzle_migrations::postgres::{PostgresDDL, ddl::Schema as PgSchema};

        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)?;

        // Process raw → DDL entities
        let mut ddl = PostgresDDL::new();
        for s in schemas {
            ddl.schemas.push(s);
        }
        for e in process_enums(&raw_enums) {
            ddl.enums.push(e);
        }
        for s in process_sequences(&raw_sequences) {
            ddl.sequences.push(s);
        }
        for r in process_roles(&raw_roles) {
            ddl.roles.push(r);
        }
        for p in process_policies(&raw_policies) {
            ddl.policies.push(p);
        }
        for t in process_tables(&raw_tables) {
            ddl.tables.push(t);
        }
        for c in process_columns(&raw_columns) {
            ddl.columns.push(c);
        }
        for i in process_indexes(&raw_indexes) {
            ddl.indexes.push(i);
        }
        for fk in process_foreign_keys(&raw_fks) {
            ddl.fks.push(fk);
        }
        for pk in process_primary_keys(&raw_primary_keys) {
            ddl.pks.push(pk);
        }
        for u in process_unique_constraints(&raw_uniques) {
            ddl.uniques.push(u);
        }
        for c in process_check_constraints(&raw_checks) {
            ddl.checks.push(c);
        }
        for v in process_views(&raw_views) {
            ddl.views.push(v);
        }

        // 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());

        let param_refs = {
            #[cfg(feature = "profiling")]
            drizzle_core::drizzle_profile_scope!("postgres.sync", "builder.execute.param_refs");
            let mut param_refs: SmallVec<[&(dyn postgres::types::ToSql + Sync); 8]> =
                SmallVec::with_capacity(params.len());
            param_refs.extend(
                params
                    .iter()
                    .map(|&p| p as &(dyn postgres::types::ToSql + Sync)),
            );
            param_refs
        };

        let mut typed_params: SmallVec<
            [(&(dyn postgres::types::ToSql + Sync), postgres::types::Type); 8],
        > = SmallVec::with_capacity(params.len());
        let mut all_typed = true;
        for p in &params {
            if let Some(ty) = crate::builder::postgres::prepared_common::postgres_sync_param_type(p)
            {
                typed_params.push((*p as &(dyn postgres::types::ToSql + Sync), ty));
            } else {
                all_typed = false;
                break;
            }
        }

        if all_typed {
            #[cfg(feature = "profiling")]
            drizzle_core::drizzle_profile_scope!("postgres.sync", "builder.execute.db_typed");
            let mut rows = self
                .runner
                .client
                .query_typed_raw(&sql_str, typed_params)
                .with_query(|| QueryContext::new(&sql_str, &params))?;
            while rows
                .next()
                .with_query(|| QueryContext::new(&sql_str, &params))?
                .is_some()
            {}
            return Ok(rows.rows_affected().unwrap_or(0));
        }

        #[cfg(feature = "profiling")]
        drizzle_core::drizzle_profile_scope!("postgres.sync", "builder.execute.db");
        self.runner
            .client
            .execute(&sql_str, &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 mut param_refs: SmallVec<[&(dyn postgres::types::ToSql + Sync); 8]> =
            SmallVec::with_capacity(params.len());
        param_refs.extend(
            params
                .iter()
                .map(|&p| p as &(dyn postgres::types::ToSql + Sync)),
        );

        let rows = self
            .runner
            .client
            .query(&sql_str, &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 mut param_refs: SmallVec<[&(dyn postgres::types::ToSql + Sync); 8]> =
            SmallVec::with_capacity(params.len());
        param_refs.extend(
            params
                .iter()
                .map(|&p| p as &(dyn postgres::types::ToSql + Sync)),
        );

        let rows = self
            .runner
            .client
            .query(&sql_str, &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 mut param_refs: SmallVec<[&(dyn postgres::types::ToSql + Sync); 8]> =
            SmallVec::with_capacity(params.len());
        param_refs.extend(
            params
                .iter()
                .map(|&p| p as &(dyn postgres::types::ToSql + Sync)),
        );

        let row = self
            .runner
            .client
            .query_one(&sql_str, &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<
            drizzle_core::query::QueryRow<
                <T as drizzle_core::query::QueryTable>::Select,
                <Rels as drizzle_core::query::BuildStore>::Store,
            >,
        >,
    >
    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::BuildStore
            + 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_refs: SmallVec<[&(dyn postgres::types::ToSql + Sync); 8]> = bind_params
            .iter()
            .map(|&p| p as &(dyn postgres::types::ToSql + Sync))
            .collect();

        let rows = self
            .runner
            .client
            .query(&sql, &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(drizzle_core::query::QueryRow::new(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<
            drizzle_core::query::QueryRow<
                <T as drizzle_core::query::QueryTable>::Select,
                <Rels as drizzle_core::query::BuildStore>::Store,
            >,
        >,
    >
    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::BuildStore
            + 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<
            drizzle_core::query::QueryRow<
                <T as drizzle_core::query::QueryTable>::PartialSelect,
                <Rels as drizzle_core::query::BuildStore>::Store,
            >,
        >,
    >
    where
        T: drizzle_core::query::QueryTable,
        <T as drizzle_core::query::QueryTable>::PartialSelect: drizzle_core::query::FromJsonObject,
        Rels: drizzle_core::query::BuildStore
            + 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_refs: SmallVec<[&(dyn postgres::types::ToSql + Sync); 8]> = bind_params
            .iter()
            .map(|&p| p as &(dyn postgres::types::ToSql + Sync))
            .collect();

        let rows = self
            .runner
            .client
            .query(&sql, &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(drizzle_core::query::QueryRow::new(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<
            drizzle_core::query::QueryRow<
                <T as drizzle_core::query::QueryTable>::PartialSelect,
                <Rels as drizzle_core::query::BuildStore>::Store,
            >,
        >,
    >
    where
        T: drizzle_core::query::QueryTable,
        <T as drizzle_core::query::QueryTable>::PartialSelect: drizzle_core::query::FromJsonObject,
        Rels: drizzle_core::query::BuildStore
            + 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<
            drizzle_core::query::QueryRow<
                <T as drizzle_core::query::QueryTable>::Select,
                <Rels as drizzle_core::query::BuildStore>::Store,
            >,
        >,
    >
    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::BuildStore,
        <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(drizzle_core::query::QueryRow::new(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<
            drizzle_core::query::QueryRow<
                <T as drizzle_core::query::QueryTable>::Select,
                <Rels as drizzle_core::query::BuildStore>::Store,
            >,
        >,
    >
    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::BuildStore,
        <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<
            drizzle_core::query::QueryRow<
                <T as drizzle_core::query::QueryTable>::PartialSelect,
                <Rels as drizzle_core::query::BuildStore>::Store,
            >,
        >,
    >
    where
        T: drizzle_core::query::QueryTable,
        <T as drizzle_core::query::QueryTable>::PartialSelect: drizzle_core::query::FromJsonObject,
        Rels: drizzle_core::query::BuildStore,
        <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(drizzle_core::query::QueryRow::new(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<
            drizzle_core::query::QueryRow<
                <T as drizzle_core::query::QueryTable>::PartialSelect,
                <Rels as drizzle_core::query::BuildStore>::Store,
            >,
        >,
    >
    where
        T: drizzle_core::query::QueryTable,
        <T as drizzle_core::query::QueryTable>::PartialSelect: drizzle_core::query::FromJsonObject,
        Rels: drizzle_core::query::BuildStore,
        <Rels as drizzle_core::query::BuildStore>::Store: drizzle_core::query::DeserializeStore,
    {
        Ok(self.find_many(client, params)?.into_iter().next())
    }
}