rustango 0.61.13

//! `.values()` and `.values_list()`: fetch chosen columns as plain
//! data instead of decoding whole models.
//!
//! [`super::decode`] turns cells into `SqlValue`s; the `fetch_values_*`
//! functions shape the rows into dicts, tuples or a flat list.

#[cfg(feature = "postgres")]
use sqlx::postgres::PgArguments;
#[cfg(feature = "postgres")]
use sqlx::query::Query;

use super::{run_on, ExecError, RowRef, SqlExecutor};
use crate::core::{AggregateQuery, SelectQuery, SqlValue};
use crate::sql::Pool;
use crate::sql::{
    MaybeMyFromRow, MaybeMyLoadRelated, MaybePgFromRow, MaybeSqliteFromRow, MaybeSqliteLoadRelated,
    UpdaterPool as _,
};

#[cfg(feature = "postgres")]
use super::bind_query;
#[cfg(feature = "mysql")]
use super::bind_query_my;
#[cfg(feature = "sqlite")]
use super::bind_query_sqlite;

use super::decode::{RowReader, TypedCol};
use crate::core::joins::joined_key;

/// Cells of a value row: the projection, then each join's columns. Any
/// after them are DISTINCT's ORDER BY helpers, which the row drops.
fn values_width(query: &SelectQuery) -> usize {
    let own = query
        .projection
        .as_ref()
        .map_or_else(|| query.model.scalar_fields().count(), Vec::len);
    own + query.joins.iter().map(|j| j.project.len()).sum::<usize>()
}

/// The model columns a SELECT emits first: its projection, else every
/// field, then each join's projected `alias__col`.
fn select_model_cols(query: &SelectQuery) -> Vec<TypedCol> {
    let model = query.model;
    // A projected `alias.col` reads back as `alias__col`, typed by its join.
    let typed = |c: &'static str| match c.split_once('.') {
        None => (c.into(), model.field_by_column(c).map(|f| f.ty)),
        Some((alias, col)) => {
            let target = query.joins.iter().find(|j| j.alias == alias);
            let ty = target
                .and_then(|j| j.target.field_by_column(col))
                .map(|f| f.ty);
            (joined_key(alias, col).into(), ty)
        }
    };
    let mut cols: Vec<TypedCol> = match &query.projection {
        Some(cols) => cols.iter().copied().map(typed).collect(),
        None => model.scalar_fields().map(|f| typed(f.column)).collect(),
    };
    for j in &query.joins {
        cols.extend(j.project.iter().map(|&c| {
            let ty = j.target.field_by_column(c).map(|f| f.ty);
            (joined_key(j.alias, c).into(), ty)
        }));
    }
    cols
}

/// An aggregate's group columns, then its aggregates, as projected. A
/// joined `alias.col` comes back as `alias__col`, typed by the joined
/// model (#2322). MAX / MIN / ANY_VALUE take their column's type, as PG
/// and MySQL report it (dialects-001); other aggregates stay untyped.
fn aggregate_cols(query: &AggregateQuery) -> Vec<TypedCol> {
    let model = query.model;
    let ty = |col: &str| match col.split_once('.') {
        None => model.field_by_column(col).map(|f| f.ty),
        Some((alias, c)) => {
            let target = if alias == model.table {
                Some(model)
            } else {
                let derived = query.source.iter().flat_map(|s| s.joins.iter());
                query
                    .joins
                    .iter()
                    .chain(derived)
                    .find(|j| j.alias == alias)
                    .map(|j| j.target)
            };
            target.and_then(|m| m.field_by_column(c)).map(|f| f.ty)
        }
    };
    let groups = query.group_by.iter().map(|&col| {
        let label = match col.split_once('.') {
            None => col.into(),
            Some((alias, c)) => joined_key(alias, c).into(),
        };
        (label, ty(col))
    });
    let aggregates = query.aggregates.iter().map(|(alias, expr)| {
        use crate::core::AggregateExpr;
        let declared = match expr {
            AggregateExpr::Max(c) | AggregateExpr::Min(c) | AggregateExpr::AnyValue(c) => ty(c),
            _ => None,
        };
        (alias.clone(), declared)
    });
    groups.chain(aggregates).collect()
}

/// Run a [`SelectQuery`] with a projection and return each row as a
/// map from column name to value. Backs
/// [`crate::query::ValuesQuerySet::fetch`].
///
/// # Errors
/// SQL compilation or driver failure.
pub async fn fetch_values_dict(
    pool: &Pool,
    query: &SelectQuery,
) -> Result<Vec<std::collections::HashMap<String, SqlValue>>, ExecError> {
    fetch_values_dict_on(pool, query).await
}

async fn fetch_values_dict_on<'c, E: SqlExecutor<'c>>(
    exec: E,
    query: &SelectQuery,
) -> Result<Vec<std::collections::HashMap<String, SqlValue>>, ExecError> {
    let stmt = exec.dialect().compile_select(query)?;
    let mut reader = RowReader::new(|| select_model_cols(query), values_width(query));
    run_on(
        exec,
        &stmt.sql,
        stmt.params,
        move |row| Ok(reader.dict(row)),
    )
    .await
}

/// [`fetch_values_dict`] for an [`AggregateQuery`]: each row comes
/// back keyed by output-column alias. Backs
/// [`crate::query::AggregateBuilder::fetch`].
///
/// It goes through [`Pool`], so `.annotate(…)` works on every
/// backend. [`crate::query::AggregateBuilder::fetch_on`] does the same
/// on any [`SqlExecutor`].
///
/// # Errors
/// SQL compilation or driver failure.
pub async fn fetch_aggregate_dict(
    pool: &Pool,
    query: &AggregateQuery,
) -> Result<Vec<std::collections::HashMap<String, SqlValue>>, ExecError> {
    fetch_aggregate_dict_on(pool, query).await
}

pub(super) async fn fetch_aggregate_dict_on<'c, E: SqlExecutor<'c>>(
    exec: E,
    query: &AggregateQuery,
) -> Result<Vec<std::collections::HashMap<String, SqlValue>>, ExecError> {
    let stmt = exec.dialect().compile_aggregate(query)?;
    let mut reader = RowReader::new(|| aggregate_cols(query), usize::MAX);
    run_on(
        exec,
        &stmt.sql,
        stmt.params,
        move |row| Ok(reader.dict(row)),
    )
    .await
}

/// Run a [`SelectQuery`] with a projection and return each row as a
/// `Vec<SqlValue>` in the projection's column order. Backs
/// [`crate::query::ValuesListQuerySet::fetch`].
///
/// # Errors
/// SQL compilation or driver failure.
pub async fn fetch_values_list(
    pool: &Pool,
    query: &SelectQuery,
) -> Result<Vec<Vec<SqlValue>>, ExecError> {
    fetch_values_list_on(pool, query).await
}

pub(crate) async fn fetch_values_list_on<'c, E: SqlExecutor<'c>>(
    exec: E,
    query: &SelectQuery,
) -> Result<Vec<Vec<SqlValue>>, ExecError> {
    let stmt = exec.dialect().compile_select(query)?;
    let mut reader = RowReader::new(|| select_model_cols(query), values_width(query));
    run_on(
        exec,
        &stmt.sql,
        stmt.params,
        move |row| Ok(reader.list(row)),
    )
    .await
}

/// The bound `.values_list_flat::<U>(…)` needs on Postgres. With the
/// feature off it is an empty blanket impl, so such builds compile.
/// See [`super::MaybePgFromRow`].
#[cfg(feature = "postgres")]
pub trait MaybePgScalar:
    for<'r> sqlx::Decode<'r, sqlx::Postgres> + sqlx::Type<sqlx::Postgres>
{
}
#[cfg(feature = "postgres")]
impl<T> MaybePgScalar for T where
    T: for<'r> sqlx::Decode<'r, sqlx::Postgres> + sqlx::Type<sqlx::Postgres>
{
}
#[cfg(not(feature = "postgres"))]
pub trait MaybePgScalar {}
#[cfg(not(feature = "postgres"))]
impl<T> MaybePgScalar for T {}

#[cfg(feature = "mysql")]
pub trait MaybeMyScalar: for<'r> sqlx::Decode<'r, sqlx::MySql> + sqlx::Type<sqlx::MySql> {}
#[cfg(feature = "mysql")]
impl<T> MaybeMyScalar for T where T: for<'r> sqlx::Decode<'r, sqlx::MySql> + sqlx::Type<sqlx::MySql> {}
#[cfg(not(feature = "mysql"))]
pub trait MaybeMyScalar {}
#[cfg(not(feature = "mysql"))]
impl<T> MaybeMyScalar for T {}

#[cfg(feature = "sqlite")]
pub trait MaybeSqliteScalar:
    for<'r> sqlx::Decode<'r, sqlx::Sqlite> + sqlx::Type<sqlx::Sqlite>
{
}
#[cfg(feature = "sqlite")]
impl<T> MaybeSqliteScalar for T where
    T: for<'r> sqlx::Decode<'r, sqlx::Sqlite> + sqlx::Type<sqlx::Sqlite>
{
}
#[cfg(not(feature = "sqlite"))]
pub trait MaybeSqliteScalar {}
#[cfg(not(feature = "sqlite"))]
impl<T> MaybeSqliteScalar for T {}

mod flat_sealed {
    pub trait Sealed {}
}

/// A type the flat `values_list` / `pluck` / `value` decode can return.
/// Sealed; `Option<T>` reads NULL as `None`, a bare `T` errors on NULL.
/// `u8`–`u64` need a build without `postgres`.
pub trait FlatScalar: flat_sealed::Sealed + Send + Unpin + Sized {
    #[doc(hidden)]
    type Cell: MaybePgScalar + MaybeMyScalar + MaybeSqliteScalar + Send + Unpin;
    #[doc(hidden)]
    fn from_cell(cell: Self::Cell) -> Self;
    #[doc(hidden)]
    fn from_null(column: &str) -> Result<Self, sqlx::Error>;
}

/// The error PG and MySQL raise themselves, so all three agree.
fn unexpected_null(column: &str) -> sqlx::Error {
    sqlx::Error::ColumnDecode {
        index: format!("{column:?}"),
        source: Box::new(sqlx::error::UnexpectedNullError),
    }
}

macro_rules! flat_scalar {
    ($($t:ty),* $(,)?) => {$(
        impl flat_sealed::Sealed for $t {}
        impl flat_sealed::Sealed for Option<$t> {}
        impl FlatScalar for $t {
            type Cell = $t;
            fn from_cell(cell: $t) -> Self {
                cell
            }
            fn from_null(column: &str) -> Result<Self, sqlx::Error> {
                Err(unexpected_null(column))
            }
        }
        impl FlatScalar for Option<$t> {
            type Cell = $t;
            fn from_cell(cell: $t) -> Self {
                Some(cell)
            }
            fn from_null(_: &str) -> Result<Self, sqlx::Error> {
                Ok(None)
            }
        }
    )*};
}

flat_scalar!(
    i8,
    i16,
    i32,
    i64,
    f32,
    f64,
    bool,
    String,
    Vec<u8>,
    serde_json::Value,
    chrono::DateTime<chrono::Utc>,
    chrono::NaiveDateTime,
    chrono::NaiveDate,
    chrono::NaiveTime,
);
impl flat_sealed::Sealed for rust_decimal::Decimal {}
impl flat_sealed::Sealed for Option<rust_decimal::Decimal> {}
impl FlatScalar for rust_decimal::Decimal {
    type Cell = DecimalCell;
    fn from_cell(cell: DecimalCell) -> Self {
        cell.0
    }
    fn from_null(column: &str) -> Result<Self, sqlx::Error> {
        Err(unexpected_null(column))
    }
}
impl FlatScalar for Option<rust_decimal::Decimal> {
    type Cell = DecimalCell;
    fn from_cell(cell: DecimalCell) -> Self {
        Some(cell.0)
    }
    fn from_null(_: &str) -> Result<Self, sqlx::Error> {
        Ok(None)
    }
}
// sqlx-postgres has no unsigned decode (MySQL reads them from UNSIGNED columns).
#[cfg(not(feature = "postgres"))]
flat_scalar!(u8, u16, u32, u64);

/// The one place a UUID cell is decoded. MySQL keeps it as `CHAR(36)` text,
/// and sqlx's `Uuid` there wants 16 raw bytes (#1733).
#[doc(hidden)]
pub struct UuidCell(uuid::Uuid);

#[cfg(feature = "postgres")]
impl sqlx::Type<sqlx::Postgres> for UuidCell {
    fn type_info() -> sqlx::postgres::PgTypeInfo {
        <uuid::Uuid as sqlx::Type<sqlx::Postgres>>::type_info()
    }
    fn compatible(ty: &sqlx::postgres::PgTypeInfo) -> bool {
        <uuid::Uuid as sqlx::Type<sqlx::Postgres>>::compatible(ty)
    }
}
#[cfg(feature = "postgres")]
impl<'r> sqlx::Decode<'r, sqlx::Postgres> for UuidCell {
    fn decode(v: sqlx::postgres::PgValueRef<'r>) -> Result<Self, sqlx::error::BoxDynError> {
        <uuid::Uuid as sqlx::Decode<sqlx::Postgres>>::decode(v).map(Self)
    }
}
#[cfg(feature = "mysql")]
impl sqlx::Type<sqlx::MySql> for UuidCell {
    fn type_info() -> sqlx::mysql::MySqlTypeInfo {
        <uuid::fmt::Hyphenated as sqlx::Type<sqlx::MySql>>::type_info()
    }
    fn compatible(ty: &sqlx::mysql::MySqlTypeInfo) -> bool {
        <uuid::fmt::Hyphenated as sqlx::Type<sqlx::MySql>>::compatible(ty)
    }
}
#[cfg(feature = "mysql")]
impl<'r> sqlx::Decode<'r, sqlx::MySql> for UuidCell {
    fn decode(v: sqlx::mysql::MySqlValueRef<'r>) -> Result<Self, sqlx::error::BoxDynError> {
        <uuid::fmt::Hyphenated as sqlx::Decode<sqlx::MySql>>::decode(v).map(|h| Self(h.into_uuid()))
    }
}
#[cfg(feature = "sqlite")]
impl sqlx::Type<sqlx::Sqlite> for UuidCell {
    fn type_info() -> sqlx::sqlite::SqliteTypeInfo {
        <uuid::Uuid as sqlx::Type<sqlx::Sqlite>>::type_info()
    }
    fn compatible(ty: &sqlx::sqlite::SqliteTypeInfo) -> bool {
        <uuid::Uuid as sqlx::Type<sqlx::Sqlite>>::compatible(ty)
    }
}
#[cfg(feature = "sqlite")]
impl<'r> sqlx::Decode<'r, sqlx::Sqlite> for UuidCell {
    fn decode(v: sqlx::sqlite::SqliteValueRef<'r>) -> Result<Self, sqlx::error::BoxDynError> {
        // sqlx binds 16 raw bytes, but a column default or raw SQL may store text.
        let bytes = <&[u8] as sqlx::Decode<sqlx::Sqlite>>::decode(v)?;
        let u = match std::str::from_utf8(bytes) {
            Ok(text) if bytes.len() != 16 => uuid::Uuid::parse_str(text)?,
            _ => uuid::Uuid::from_slice(bytes)?,
        };
        Ok(Self(u))
    }
}

/// The one place a `Decimal` cell is decoded. sqlx-sqlite has no `Decimal`,
/// and a SQLite `NUMERIC` cell holds INTEGER, REAL or TEXT.
#[doc(hidden)]
pub struct DecimalCell(rust_decimal::Decimal);

#[cfg(feature = "postgres")]
impl sqlx::Type<sqlx::Postgres> for DecimalCell {
    fn type_info() -> sqlx::postgres::PgTypeInfo {
        <rust_decimal::Decimal as sqlx::Type<sqlx::Postgres>>::type_info()
    }
    fn compatible(ty: &sqlx::postgres::PgTypeInfo) -> bool {
        <rust_decimal::Decimal as sqlx::Type<sqlx::Postgres>>::compatible(ty)
    }
}
#[cfg(feature = "postgres")]
impl<'r> sqlx::Decode<'r, sqlx::Postgres> for DecimalCell {
    fn decode(v: sqlx::postgres::PgValueRef<'r>) -> Result<Self, sqlx::error::BoxDynError> {
        <rust_decimal::Decimal as sqlx::Decode<sqlx::Postgres>>::decode(v).map(Self)
    }
}
#[cfg(feature = "mysql")]
impl sqlx::Type<sqlx::MySql> for DecimalCell {
    fn type_info() -> sqlx::mysql::MySqlTypeInfo {
        <rust_decimal::Decimal as sqlx::Type<sqlx::MySql>>::type_info()
    }
    fn compatible(ty: &sqlx::mysql::MySqlTypeInfo) -> bool {
        <rust_decimal::Decimal as sqlx::Type<sqlx::MySql>>::compatible(ty)
    }
}
#[cfg(feature = "mysql")]
impl<'r> sqlx::Decode<'r, sqlx::MySql> for DecimalCell {
    fn decode(v: sqlx::mysql::MySqlValueRef<'r>) -> Result<Self, sqlx::error::BoxDynError> {
        <rust_decimal::Decimal as sqlx::Decode<sqlx::MySql>>::decode(v).map(Self)
    }
}
#[cfg(feature = "sqlite")]
impl sqlx::Type<sqlx::Sqlite> for DecimalCell {
    fn type_info() -> sqlx::sqlite::SqliteTypeInfo {
        <f64 as sqlx::Type<sqlx::Sqlite>>::type_info()
    }
    fn compatible(_: &sqlx::sqlite::SqliteTypeInfo) -> bool {
        true
    }
}
#[cfg(feature = "sqlite")]
impl<'r> sqlx::Decode<'r, sqlx::Sqlite> for DecimalCell {
    fn decode(v: sqlx::sqlite::SqliteValueRef<'r>) -> Result<Self, sqlx::error::BoxDynError> {
        use rust_decimal::Decimal;
        use sqlx::{TypeInfo as _, ValueRef as _};
        let d = match v.type_info().name() {
            "INTEGER" => Decimal::from(<i64 as sqlx::Decode<sqlx::Sqlite>>::decode(v)?),
            "REAL" => Decimal::try_from(<f64 as sqlx::Decode<sqlx::Sqlite>>::decode(v)?)?,
            _ => {
                let s = <&str as sqlx::Decode<sqlx::Sqlite>>::decode(v)?.trim();
                s.parse::<Decimal>()
                    .or_else(|_| Decimal::from_scientific(s))?
            }
        };
        Ok(Self(d))
    }
}

impl flat_sealed::Sealed for uuid::Uuid {}
impl flat_sealed::Sealed for Option<uuid::Uuid> {}
impl FlatScalar for uuid::Uuid {
    type Cell = UuidCell;
    fn from_cell(cell: UuidCell) -> Self {
        cell.0
    }
    fn from_null(column: &str) -> Result<Self, sqlx::Error> {
        Err(unexpected_null(column))
    }
}
impl FlatScalar for Option<uuid::Uuid> {
    type Cell = UuidCell;
    fn from_cell(cell: UuidCell) -> Self {
        Some(cell.0)
    }
    fn from_null(_: &str) -> Result<Self, sqlx::Error> {
        Ok(None)
    }
}

impl<T> flat_sealed::Sealed for sqlx::types::Json<T> {}
impl<T> flat_sealed::Sealed for Option<sqlx::types::Json<T>> {}
impl<T> FlatScalar for sqlx::types::Json<T>
where
    T: serde::de::DeserializeOwned + Send + Unpin + 'static,
{
    type Cell = Self;
    fn from_cell(cell: Self) -> Self {
        cell
    }
    fn from_null(column: &str) -> Result<Self, sqlx::Error> {
        Err(unexpected_null(column))
    }
}
impl<T> FlatScalar for Option<sqlx::types::Json<T>>
where
    T: serde::de::DeserializeOwned + Send + Unpin + 'static,
{
    type Cell = sqlx::types::Json<T>;
    fn from_cell(cell: Self::Cell) -> Self {
        Some(cell)
    }
    fn from_null(_: &str) -> Result<Self, sqlx::Error> {
        Ok(None)
    }
}

#[cfg(test)]
mod flat_scalar_tests {
    fn is_flat<U: super::FlatScalar>() {}

    /// The types `pluck_pairs` decoded before `FlatScalar` narrowed it.
    #[test]
    fn small_ints_and_json_are_flat_scalars() {
        is_flat::<i8>();
        is_flat::<Option<i8>>();
        is_flat::<sqlx::types::Json<Vec<String>>>();
        is_flat::<Option<sqlx::types::Json<Vec<String>>>>();
    }

    #[cfg(not(feature = "postgres"))]
    #[test]
    fn unsigned_ints_are_flat_scalars_without_postgres() {
        is_flat::<u8>();
        is_flat::<u16>();
        is_flat::<u32>();
        is_flat::<Option<u64>>();
    }

    /// `pluck::<Decimal>` compiled before `FlatScalar`; keep it that way.
    #[test]
    fn decimal_is_a_flat_scalar() {
        is_flat::<rust_decimal::Decimal>();
        is_flat::<Option<rust_decimal::Decimal>>();
    }
}

/// Decode column 0, checking NULL first: sqlx-sqlite reads NULL as `0` (#1773).
#[cfg(any(feature = "postgres", feature = "mysql", feature = "sqlite"))]
fn decode_flat<R, U>(row: &R) -> Result<U, sqlx::Error>
where
    R: sqlx::Row,
    usize: sqlx::ColumnIndex<R>,
    U: FlatScalar,
    U::Cell: for<'r> sqlx::Decode<'r, R::Database> + sqlx::Type<R::Database>,
{
    decode_cell(row, 0)
}

/// Decode column `index` the same NULL-checked way.
#[cfg(any(feature = "postgres", feature = "mysql", feature = "sqlite"))]
fn decode_cell<R, U>(row: &R, index: usize) -> Result<U, sqlx::Error>
where
    R: sqlx::Row,
    usize: sqlx::ColumnIndex<R>,
    U: FlatScalar,
    U::Cell: for<'r> sqlx::Decode<'r, R::Database> + sqlx::Type<R::Database>,
{
    use sqlx::Column as _;
    if super::row_to_json::cell_is_null(row, index) {
        let name = row.columns().get(index).map(|c| c.name().to_owned());
        return U::from_null(name.as_deref().unwrap_or("?"));
    }
    row.try_get::<U::Cell, _>(index).map(U::from_cell)
}

/// Decode the named column of a MySQL row through [`FlatScalar`], for
/// macro-emitted code; a missing column is an error, not NULL.
///
/// # Errors
/// The driver's decode error, or NULL for a bare `U`.
#[cfg(feature = "mysql")]
#[doc(hidden)]
pub fn try_get_flat_my<U: FlatScalar>(
    row: &crate::sql::MyReturningRow,
    name: &str,
) -> Result<U, sqlx::Error> {
    use sqlx::{Row as _, ValueRef as _};
    if row.try_get_raw(name)?.is_null() {
        return U::from_null(name);
    }
    row.try_get::<U::Cell, _>(name).map(U::from_cell)
}

#[cfg(not(feature = "mysql"))]
#[doc(hidden)]
#[allow(clippy::missing_errors_doc)]
pub fn try_get_flat_my<U>(row: &crate::sql::MyReturningRow, _name: &str) -> Result<U, sqlx::Error> {
    match *row {}
}

/// Decode the named column of any backend's row through [`FlatScalar`], for
/// macro-emitted code (a `Decimal` field, which sqlx-sqlite cannot decode).
///
/// # Errors
/// The driver's decode error, or NULL for a bare `U`.
#[doc(hidden)]
pub fn try_get_flat<R, U>(row: &R, name: &str) -> Result<U, sqlx::Error>
where
    R: sqlx::Row,
    for<'n> &'n str: sqlx::ColumnIndex<R>,
    U: FlatScalar,
    U::Cell: for<'r> sqlx::Decode<'r, R::Database> + sqlx::Type<R::Database>,
{
    use sqlx::ValueRef as _;
    if row.try_get_raw(name)?.is_null() {
        return U::from_null(name);
    }
    row.try_get::<U::Cell, _>(name).map(U::from_cell)
}

/// SQLite twin of [`try_get_flat_my`]: a UUID a column default wrote is TEXT.
///
/// # Errors
/// The driver's decode error, or NULL for a bare `U`.
#[cfg(feature = "sqlite")]
#[doc(hidden)]
pub fn try_get_flat_sqlite<U: FlatScalar>(
    row: &crate::sql::SqliteReturningRow,
    name: &str,
) -> Result<U, sqlx::Error> {
    use sqlx::{Row as _, ValueRef as _};
    if row.try_get_raw(name)?.is_null() {
        return U::from_null(name);
    }
    row.try_get::<U::Cell, _>(name).map(U::from_cell)
}

#[cfg(not(feature = "sqlite"))]
#[doc(hidden)]
#[allow(clippy::missing_errors_doc)]
pub fn try_get_flat_sqlite<U>(
    row: &crate::sql::SqliteReturningRow,
    _name: &str,
) -> Result<U, sqlx::Error> {
    match *row {}
}

/// Decode a `(K, V)` row, each cell NULL-checked (#1808).
#[cfg(any(feature = "postgres", feature = "mysql", feature = "sqlite"))]
fn decode_pair<R, K, V>(row: &R) -> Result<(K, V), sqlx::Error>
where
    R: sqlx::Row,
    usize: sqlx::ColumnIndex<R>,
    K: FlatScalar,
    V: FlatScalar,
    K::Cell: for<'r> sqlx::Decode<'r, R::Database> + sqlx::Type<R::Database>,
    V::Cell: for<'r> sqlx::Decode<'r, R::Database> + sqlx::Type<R::Database>,
{
    Ok((decode_cell(row, 0)?, decode_cell(row, 1)?))
}

/// Run a one-column [`SelectQuery`] and decode each row's only cell
/// into `U` — the flat form of `.values_list()`.
///
/// # Errors
/// SQL compilation or driver failure, including a decode error when
/// `U` does not match the column's type or a bare `U` meets NULL.
pub async fn fetch_values_flat<U: FlatScalar>(
    pool: &Pool,
    query: &SelectQuery,
) -> Result<Vec<U>, ExecError> {
    fetch_values_flat_on(pool, query).await
}

pub(crate) async fn fetch_values_flat_on<'c, U: FlatScalar, E: SqlExecutor<'c>>(
    exec: E,
    query: &SelectQuery,
) -> Result<Vec<U>, ExecError> {
    let stmt = exec.dialect().compile_select(query)?;
    run_on(exec, &stmt.sql, stmt.params, |row| match row {
        #[cfg(feature = "postgres")]
        RowRef::Postgres(r) => decode_flat(r),
        #[cfg(feature = "mysql")]
        RowRef::Mysql(r) => decode_flat(r),
        #[cfg(feature = "sqlite")]
        RowRef::Sqlite(r) => decode_flat(r),
    })
    .await
}

/// Run a two-column [`SelectQuery`] and decode each row into
/// `(K, V)`. Backs [`crate::query::QuerySet::pluck_pairs`].
///
/// # Errors
/// SQL compilation or driver failure, including a decode error when
/// `K` or `V` does not match its column's type.
pub async fn fetch_values_pairs<K, V>(
    pool: &Pool,
    query: &SelectQuery,
) -> Result<Vec<(K, V)>, ExecError>
where
    K: FlatScalar,
    V: FlatScalar,
{
    let stmt = pool.dialect().compile_select(query)?;
    crate::test_assertions::query_counter::bump();
    match pool {
        #[cfg(feature = "postgres")]
        Pool::Postgres(pg) => {
            let mut q: Query<'_, sqlx::Postgres, PgArguments> = sqlx::query(&stmt.sql);
            for v in stmt.params {
                q = bind_query(q, v);
            }
            let rows = q.fetch_all(pg).await?;
            Ok(rows.iter().map(decode_pair).collect::<Result<_, _>>()?)
        }
        #[cfg(feature = "mysql")]
        Pool::Mysql(my) => {
            let mut q: sqlx::query::Query<'_, sqlx::MySql, sqlx::mysql::MySqlArguments> =
                sqlx::query(&stmt.sql);
            for v in stmt.params {
                q = bind_query_my(q, v);
            }
            let rows = q.fetch_all(my).await?;
            Ok(rows.iter().map(decode_pair).collect::<Result<_, _>>()?)
        }
        #[cfg(feature = "sqlite")]
        Pool::Sqlite(sq) => {
            let mut q: sqlx::query::Query<'_, sqlx::Sqlite, sqlx::sqlite::SqliteArguments<'_>> =
                sqlx::query(&stmt.sql);
            for v in stmt.params {
                q = bind_query_sqlite(q, v);
            }
            let rows = q.fetch_all(sq).await?;
            Ok(rows.iter().map(decode_pair).collect::<Result<_, _>>()?)
        }
    }
}

// Bridge methods on the values builders so callers chain `.fetch(&pool)`.

impl<T: crate::core::Model> crate::query::ValuesQuerySet<T> {
    /// Run the projection and return one map per row.
    ///
    /// # Errors
    /// - [`ExecError::Query`] when the SQL fails to compile, for
    ///   example on a misspelled column.
    /// - [`ExecError::Driver`] for driver, network or decode failures.
    pub async fn fetch(
        self,
        pool: &Pool,
    ) -> Result<Vec<std::collections::HashMap<String, SqlValue>>, ExecError> {
        let q = self.compile()?;
        fetch_values_dict(pool, &q).await
    }
}

impl<T: crate::core::Model> crate::query::ValuesQuerySet<T> {
    /// [`Self::fetch`] on a [`SqlExecutor`], e.g. a held [`crate::sql::PoolConn`]
    /// or a schema-mode tenant's connection, which holds its `search_path`.
    ///
    /// # Errors
    /// As [`Self::fetch`].
    pub async fn fetch_on<'c, E: SqlExecutor<'c>>(
        self,
        exec: E,
    ) -> Result<Vec<std::collections::HashMap<String, SqlValue>>, ExecError> {
        let q = self.compile()?;
        fetch_values_dict_on(exec, &q).await
    }
}

impl<T: crate::core::Model> crate::query::AggregateBuilder<T> {
    /// Run the aggregate query and return one map per row, keyed by
    /// output-column name.
    ///
    /// This backs every `.aggregate()` and `.annotate(…)` chain,
    /// including [`crate::query::QuerySet::annotate_count`] and its
    /// siblings, where each row carries the parent's own columns plus
    /// the derived one.
    ///
    /// # Errors
    /// - [`ExecError::Query`] when the SQL fails to compile, for
    ///   example on an unknown relation or a bad GROUP BY.
    /// - [`ExecError::Driver`] for driver, network or decode failures.
    pub async fn fetch(
        self,
        pool: &Pool,
    ) -> Result<Vec<std::collections::HashMap<String, SqlValue>>, ExecError> {
        let q = self.compile()?;
        fetch_aggregate_dict(pool, &q).await
    }

    /// [`Self::fetch`] on a [`SqlExecutor`]. See
    /// [`crate::query::QuerySet::fetch_on`].
    ///
    /// ```ignore
    /// SetLog::objects()
    ///     .values(&["member_id"])
    ///     .annotate("total", sum("volume").into())
    ///     .fetch_on(&mut conn)
    ///     .await?;
    /// ```
    ///
    /// # Errors
    /// As [`Self::fetch`].
    pub async fn fetch_on<'c, E: SqlExecutor<'c>>(
        self,
        exec: E,
    ) -> Result<Vec<std::collections::HashMap<String, SqlValue>>, ExecError> {
        let q = self.compile()?;
        fetch_aggregate_dict_on(exec, &q).await
    }
}

impl<T: crate::core::Model> crate::query::ValuesListQuerySet<T> {
    /// Execute the projection and return rows as `Vec<Vec<SqlValue>>`.
    ///
    /// # Errors
    /// As [`crate::query::ValuesQuerySet::fetch`].
    pub async fn fetch(self, pool: &Pool) -> Result<Vec<Vec<SqlValue>>, ExecError> {
        let q = self.compile()?;
        fetch_values_list(pool, &q).await
    }

    /// [`Self::fetch`] on a [`SqlExecutor`].
    ///
    /// # Errors
    /// As [`Self::fetch`].
    pub async fn fetch_on<'c, E: SqlExecutor<'c>>(
        self,
        exec: E,
    ) -> Result<Vec<Vec<SqlValue>>, ExecError> {
        let q = self.compile()?;
        fetch_values_list_on(exec, &q).await
    }
}

impl<T: crate::core::Model> crate::query::QuerySet<T> {
    /// Read one column from the matching rows into a `Vec<U>`.
    ///
    /// ```ignore
    /// let titles: Vec<String> = Post::objects()
    ///     .filter("published", true)
    ///     .pluck::<String>("title", &pool).await?;
    /// ```
    ///
    /// `U` is a [`FlatScalar`]: `i64`, `String`, `bool`, `f64` and the
    /// like, or `Option<_>` of one for a nullable column. Unlike `Model::pluck`, this respects the queryset's
    /// filters, ordering and limits.
    ///
    /// # Errors
    /// As [`crate::query::ValuesFlatQuerySet::fetch`].
    pub async fn pluck<U>(self, col: &'static str, pool: &Pool) -> Result<Vec<U>, ExecError>
    where
        U: FlatScalar,
    {
        self.values_list_flat(col).fetch::<U>(pool).await
    }

    /// [`Self::pluck`] of the primary key, whose name comes from the
    /// model's schema so the call site need not repeat it.
    ///
    /// ```ignore
    /// let ids: Vec<i64> = Post::objects()
    ///     .filter("published", true)
    ///     .pks::<i64>(&pool).await?;
    /// ```
    ///
    /// # Errors
    /// [`ExecError::Query(QueryError::UnknownField)`] when the model
    /// has no primary key, otherwise as [`Self::pluck`].
    pub async fn pks<K>(self, pool: &Pool) -> Result<Vec<K>, ExecError>
    where
        K: FlatScalar,
    {
        let pk_col = T::SCHEMA.primary_key().map(|f| f.column).ok_or_else(|| {
            ExecError::Query(crate::core::QueryError::UnknownField {
                model: T::SCHEMA.name,
                field: "<pk>".to_string(),
            })
        })?;
        self.pluck::<K>(pk_col, pool).await
    }

    /// Read two columns from the matching rows into `Vec<(K, V)>`.
    /// Collect it into a map at the call site if you want lookups.
    /// The key column comes first, matching the tuple.
    ///
    /// ```ignore
    /// let pairs: Vec<(i64, String)> = Post::objects()
    ///     .filter("published", true)
    ///     .pluck_pairs::<i64, String>("id", "title", &pool).await?;
    /// let map: std::collections::BTreeMap<_, _> = pairs.into_iter().collect();
    /// ```
    ///
    /// # Errors
    /// As [`crate::sql::FetcherPool::fetch`], plus a decode error
    /// when `K` or `V` does not match its column's type.
    pub async fn pluck_pairs<K, V>(
        self,
        key_col: &'static str,
        value_col: &'static str,
        pool: &Pool,
    ) -> Result<Vec<(K, V)>, ExecError>
    where
        K: FlatScalar,
        V: FlatScalar,
    {
        let q = self.values_list(&[key_col, value_col]).compile()?;
        fetch_values_pairs::<K, V>(pool, &q).await
    }

    /// Read one column from the first matching row, or `None` when
    /// nothing matches. The query carries `LIMIT 1`, so the database
    /// does not build rows you will not read.
    ///
    /// ```ignore
    /// let email: Option<String> = User::objects()
    ///     .filter("id", 1_i64)
    ///     .value::<String>("email", &pool).await?;
    /// ```
    ///
    /// Unlike `Model::value`, this respects the queryset's filters
    /// and ordering when picking the row.
    ///
    /// # Errors
    /// As [`crate::query::ValuesFlatQuerySet::first`].
    pub async fn value<U>(self, col: &'static str, pool: &Pool) -> Result<Option<U>, ExecError>
    where
        U: FlatScalar,
    {
        self.values_list_flat(col).first::<U>(pool).await
    }

    /// Render this queryset to SQL in the pool's dialect, without
    /// running it. Handy for debugging, logging and snapshot tests.
    ///
    /// The string holds placeholders, not the bound values. Use
    /// [`Self::to_compiled`] when you need those too.
    ///
    /// ```ignore
    /// let sql = Post::objects()
    ///     .filter("published", true)
    ///     .order_by(&[("created_at", true)])
    ///     .limit(10)
    ///     .to_sql(&pool)?;
    /// // -> "SELECT … FROM \"post\" WHERE \"published\" = $1 …"
    /// ```
    ///
    /// # Errors
    /// [`ExecError::Query`] when the queryset fails to compile, and
    /// [`ExecError::Sql`] from the dialect's writer.
    pub fn to_sql(self, pool: &Pool) -> Result<String, ExecError> {
        let q = self.compile()?;
        let stmt = pool.dialect().compile_select(&q)?;
        Ok(stmt.sql)
    }

    /// `SELECT SUM(col)` over the matching rows, or `Ok(None)` when
    /// none match. `Model::sum` is this over the default queryset.
    ///
    /// # Errors
    /// As [`crate::sql::fetch_aggregate_pool`], plus
    /// [`ExecError::Query(QueryError::UnknownField)`] when the model
    /// has no such column.
    pub async fn sum<U>(self, col: &str, pool: &Pool) -> Result<Option<U>, ExecError>
    where
        (Option<U>,): crate::sql::MaybePgFromRow
            + crate::sql::MaybeMyFromRow
            + crate::sql::MaybeSqliteFromRow
            + Send
            + Unpin,
    {
        self.queryset_aggregate_one::<U>(col, crate::core::AggregateExpr::Sum, pool)
            .await
    }

    /// Fetch one page of rows and the matching total, as
    /// `(rows, total)`. Unlike `Model::paginate`, the total counts
    /// only rows the queryset's filters match.
    ///
    /// It runs two queries, a `COUNT(*)` and the page itself, both
    /// with the same WHERE clause.
    ///
    /// `page` starts at 1, so `paginate(1, 10)` gives the first ten
    /// rows and `paginate(2, 10)` the next ten. An unordered queryset
    /// pages in PK order.
    ///
    /// # Errors
    /// As [`crate::sql::CounterPool::count`] and
    /// [`crate::sql::FetcherPool::fetch`].
    pub async fn paginate(
        self,
        page: i64,
        per_page: i64,
        pool: &Pool,
    ) -> Result<(Vec<T>, i64), ExecError>
    where
        T: crate::core::Model
            + crate::sql::MaybePgFromRow
            + crate::sql::MaybeMyFromRow
            + crate::sql::MaybeSqliteFromRow
            + crate::sql::LoadRelated
            + crate::sql::MaybeMyLoadRelated
            + crate::sql::MaybeSqliteLoadRelated
            + Send
            + Unpin,
    {
        use crate::sql::{CounterPool as _, FetcherPool as _};
        let total = <crate::query::QuerySet<T> as ::core::clone::Clone>::clone(&self)
            .count(pool)
            .await?;
        let offset = crate::list_params::page_offset(page, per_page);
        let rows = self
            .ordered_or_by_pk()
            .limit(per_page)
            .offset(offset)
            .fetch(pool)
            .await?;
        Ok((rows, total))
    }

    /// [`Self::sum`] as an average.
    ///
    /// # Errors
    /// As [`Self::sum`].
    pub async fn avg<U>(self, col: &str, pool: &Pool) -> Result<Option<U>, ExecError>
    where
        (Option<U>,): crate::sql::MaybePgFromRow
            + crate::sql::MaybeMyFromRow
            + crate::sql::MaybeSqliteFromRow
            + Send
            + Unpin,
    {
        self.queryset_aggregate_one::<U>(col, crate::core::AggregateExpr::Avg, pool)
            .await
    }

    /// [`Self::sum`] as a minimum.
    ///
    /// # Errors
    /// As [`Self::sum`].
    pub async fn min<U>(self, col: &str, pool: &Pool) -> Result<Option<U>, ExecError>
    where
        (Option<U>,): crate::sql::MaybePgFromRow
            + crate::sql::MaybeMyFromRow
            + crate::sql::MaybeSqliteFromRow
            + Send
            + Unpin,
    {
        self.queryset_aggregate_one::<U>(col, crate::core::AggregateExpr::Min, pool)
            .await
    }

    /// [`Self::sum`] as a maximum.
    ///
    /// # Errors
    /// As [`Self::sum`].
    pub async fn max<U>(self, col: &str, pool: &Pool) -> Result<Option<U>, ExecError>
    where
        (Option<U>,): crate::sql::MaybePgFromRow
            + crate::sql::MaybeMyFromRow
            + crate::sql::MaybeSqliteFromRow
            + Send
            + Unpin,
    {
        self.queryset_aggregate_one::<U>(col, crate::core::AggregateExpr::Max, pool)
            .await
    }

    /// Shared body of `sum`, `avg`, `min` and `max`. It checks the
    /// column against the model, moves the queryset's filters into
    /// the aggregate's WHERE clause, and returns the one value.
    async fn queryset_aggregate_one<U>(
        self,
        col: &str,
        build: fn(&'static str) -> crate::core::AggregateExpr,
        pool: &Pool,
    ) -> Result<Option<U>, ExecError>
    where
        (Option<U>,): crate::sql::MaybePgFromRow
            + crate::sql::MaybeMyFromRow
            + crate::sql::MaybeSqliteFromRow
            + Send
            + Unpin,
    {
        let col_static = crate::sql::model_shortcuts::resolve_col::<T>(col)?;
        let Some(select_q) = self.compile_unless_none()? else {
            return Ok(None);
        };
        // One aggregate column, so the tuple decode lines up.
        let aggregate_q = crate::core::AggregateQuery::over_select(
            select_q,
            vec![("v".into(), build(col_static))],
        );
        let rows: Vec<(Option<U>,)> = crate::sql::fetch_aggregate_pool(pool, &aggregate_q).await?;
        Ok(rows.into_iter().next().and_then(|t| t.0))
    }

    /// [`Self::to_sql`] plus the bound parameters, as a
    /// [`crate::sql::CompiledStatement`].
    ///
    /// # Errors
    /// As [`Self::to_sql`].
    pub fn to_compiled(self, pool: &Pool) -> Result<crate::sql::CompiledStatement, ExecError> {
        let q = self.compile()?;
        let stmt = pool.dialect().compile_select(&q)?;
        Ok(stmt)
    }
}

impl<T> crate::query::QuerySet<T>
where
    T: crate::core::Model
        + Send
        + Unpin
        + MaybePgFromRow
        + MaybeMyFromRow
        + MaybeSqliteFromRow
        + crate::sql::LoadRelated
        + MaybeMyLoadRelated
        + MaybeSqliteLoadRelated,
{
    /// `UPDATE … SET col = col + by` over the matching rows.
    /// Returns the number of rows affected. Unlike
    /// `Model::increment_each`, this respects the queryset's
    /// filters, so you can bump a counter on some rows only.
    ///
    /// A negative `by` subtracts; [`Self::decrement`] reads better
    /// at the call site.
    ///
    /// ```ignore
    /// Post::objects()
    ///     .filter("published", true)
    ///     .increment("views", 1, &pool)
    ///     .await?;
    /// ```
    ///
    /// # Errors
    /// As [`UpdaterPool::execute_pool`](crate::sql::UpdaterPool::execute_pool),
    /// plus [`ExecError::Query`] with `QueryError::UnknownField` when
    /// `col` is not a declared field on `T`.
    pub async fn increment(self, col: &str, by: i64, pool: &Pool) -> Result<u64, ExecError> {
        let col_static = crate::sql::model_shortcuts::resolve_col::<T>(col)?;
        self.update()
            .set_expr(
                col,
                crate::sql::model_shortcuts::add_signed_expr(col_static, by),
            )
            .execute_pool(pool)
            .await
    }

    /// [`Self::increment`] the other way, subtracting `by`.
    ///
    /// ```ignore
    /// User::objects()
    ///     .filter("vip", true)
    ///     .decrement("credits", 10, &pool)
    ///     .await?;
    /// ```
    ///
    /// # Errors
    /// As [`Self::increment`].
    pub async fn decrement(self, col: &str, by: i64, pool: &Pool) -> Result<u64, ExecError> {
        self.increment(col, -by, pool).await
    }
}

impl<T: crate::core::Model> crate::query::ValuesFlatQuerySet<T> {
    /// Run the one-column projection and decode each cell into `U`.
    ///
    /// # Errors
    /// As [`crate::query::ValuesQuerySet::fetch`], plus a decode
    /// error when `U` does not match the column's type.
    pub async fn fetch<U>(self, pool: &Pool) -> Result<Vec<U>, ExecError>
    where
        U: FlatScalar,
    {
        let q = self.compile()?;
        fetch_values_flat::<U>(pool, &q).await
    }

    /// Return the first row's cell, or `None` when nothing matches.
    /// It adds `LIMIT 1`, so unlike [`Self::fetch`] the database
    /// does not build rows you will not read.
    ///
    /// ```ignore
    /// let name: Option<String> = User::query()
    ///     .filter("id", 1_i64)
    ///     .values_list_flat("name")
    ///     .first::<String>(&pool).await?;
    /// ```
    ///
    /// # Errors
    /// As [`Self::fetch`].
    pub async fn first<U>(self, pool: &Pool) -> Result<Option<U>, ExecError>
    where
        U: FlatScalar,
    {
        self.first_on(pool).await
    }

    /// [`Self::fetch`] on a [`SqlExecutor`].
    ///
    /// # Errors
    /// As [`Self::fetch`].
    pub async fn fetch_on<'c, U: FlatScalar>(
        self,
        exec: impl SqlExecutor<'c>,
    ) -> Result<Vec<U>, ExecError> {
        let q = self.compile()?;
        fetch_values_flat_on(exec, &q).await
    }

    /// [`Self::first`] on a [`SqlExecutor`].
    ///
    /// # Errors
    /// As [`Self::fetch`].
    pub async fn first_on<'c, U: FlatScalar>(
        self,
        exec: impl SqlExecutor<'c>,
    ) -> Result<Option<U>, ExecError> {
        // Rebuild with `LIMIT 1`. `compile()` consumes `self.qs`, so
        // the limit has to go on through the builder.
        let col = self.col;
        let qs = self.qs.limit(1);
        let q = crate::query::ValuesFlatQuerySet { qs, col }.compile()?;
        let rows: Vec<U> = fetch_values_flat_on(exec, &q).await?;
        Ok(rows.into_iter().next())
    }
}