sqlmodel-query 0.4.3

//! SELECT query builder.

use crate::clause::{Limit, Offset, OrderBy, Where};
use crate::eager::{
    EagerLoader, IncludePath, build_aliased_column_parts, build_join_clause, find_relationship,
};
use crate::expr::{Dialect, Expr};
use crate::join::Join;
use crate::subquery::SelectQuery;
use asupersync::{Cx, Outcome};
use sqlmodel_core::{Connection, Model, RelationshipKind, Row, Value};
use std::collections::{HashMap, HashSet};
use std::marker::PhantomData;

type ParentFieldsFn = fn() -> &'static [sqlmodel_core::FieldInfo];

/// The implicit `table.discriminator = value` predicate for a single-table
/// inheritance child model, or `None` for base/non-STI models. Applied by
/// SELECT, UPDATE, and DELETE builders so a child model never reads or writes
/// rows of a sibling kind.
pub(crate) fn sti_discriminator_filter<M: Model>() -> Option<Expr> {
    let inh = M::inheritance();
    match (inh.discriminator_column, inh.discriminator_value) {
        (Some(col), Some(val)) => Some(Expr::qualified(M::TABLE_NAME, col).eq(val)),
        _ => None,
    }
}

fn joined_inheritance_parent<M: Model>() -> Option<(&'static str, ParentFieldsFn)> {
    let inh = M::inheritance();
    if inh.strategy != sqlmodel_core::InheritanceStrategy::Joined {
        return None;
    }
    let parent = inh.parent?;
    let parent_fields_fn = inh.parent_fields_fn?;
    Some((parent, parent_fields_fn))
}

fn joined_inheritance_join<M: Model>() -> Option<Join> {
    let (parent_table, _parent_fields_fn) = joined_inheritance_parent::<M>()?;

    // Join child's PK columns to the parent's PK columns (same names).
    let pks = M::PRIMARY_KEY;
    if pks.is_empty() {
        return None;
    }

    let mut on = Expr::qualified(M::TABLE_NAME, pks[0]).eq(Expr::qualified(parent_table, pks[0]));
    for pk in &pks[1..] {
        on = on.and(Expr::qualified(M::TABLE_NAME, *pk).eq(Expr::qualified(parent_table, *pk)));
    }

    Some(Join::inner(parent_table, on))
}

fn joined_inheritance_select_columns<M: Model>(dialect: Dialect) -> Option<Vec<String>> {
    let (parent_table, parent_fields_fn) = joined_inheritance_parent::<M>()?;

    let child_cols: Vec<&str> = M::fields().iter().map(|f| f.column_name).collect();
    let parent_cols: Vec<&str> = parent_fields_fn().iter().map(|f| f.column_name).collect();

    let mut parts = Vec::new();
    parts.extend(build_aliased_column_parts(
        dialect,
        M::TABLE_NAME,
        &child_cols,
    ));
    parts.extend(build_aliased_column_parts(
        dialect,
        parent_table,
        &parent_cols,
    ));
    Some(parts)
}

/// `(table, column)` pairs for every column of `T`, rendered as quoted
/// `table.column AS table__column` projections once the dialect is known.
fn table_columns<T: Model>() -> Vec<(&'static str, &'static str)> {
    T::fields()
        .iter()
        .map(|f| (T::TABLE_NAME, f.column_name))
        .collect()
}

fn render_aliased_projection(dialect: Dialect, pairs: &[(&'static str, &'static str)]) -> String {
    pairs
        .iter()
        .flat_map(|(table, col)| build_aliased_column_parts(dialect, table, &[col]))
        .collect::<Vec<_>>()
        .join(", ")
}

/// Information about a JOIN for eager loading.
///
/// Used internally to track which relationships are being eagerly loaded
/// and how to hydrate them from the query results.
#[derive(Debug, Clone)]
#[allow(dead_code)] // `kind` and `nested` are diagnostics; nested paths are not joined yet.
struct EagerJoinInfo {
    /// Name of the relationship field.
    relationship_name: &'static str,
    /// Table name of the related model.
    related_table: &'static str,
    /// Primary-key columns of the related model, used to drop the no-match
    /// rows of a LEFT JOIN and to deduplicate a to-one row repeated by
    /// another include's fan-out.
    related_pk: Vec<&'static str>,
    /// Kind of relationship.
    kind: RelationshipKind,
    /// Nested relationships to load.
    nested: Vec<IncludePath>,
}

/// A SELECT query builder.
///
/// Provides a fluent API for building SELECT queries with
/// type-safe column references and conditions.
#[derive(Debug, Clone)]
pub struct Select<M: Model> {
    /// Columns to select (empty = all)
    columns: Vec<String>,
    /// `(table, column)` pairs projected as quoted `table.column AS table__column`
    /// at build time (polymorphic joined selects); empty otherwise.
    aliased_projection: Vec<(&'static str, &'static str)>,
    /// WHERE clause conditions
    where_clause: Option<Where>,
    /// ORDER BY clauses
    order_by: Vec<OrderBy>,
    /// JOIN clauses
    joins: Vec<Join>,
    /// LIMIT clause
    limit: Option<Limit>,
    /// OFFSET clause
    offset: Option<Offset>,
    /// GROUP BY columns
    group_by: Vec<String>,
    /// HAVING clause
    having: Option<Where>,
    /// DISTINCT flag
    distinct: bool,
    /// FOR UPDATE flag
    for_update: bool,
    /// Eager loading configuration
    eager_loader: Option<EagerLoader<M>>,
    /// Model type marker
    _marker: PhantomData<M>,
}

/// Generate one `Select::<Base>::polymorphic_joinedN` entry method: project
/// full base + child columns and `LEFT JOIN` every child table on its primary
/// key, which is what deterministic prefixed hydration requires.
macro_rules! polymorphic_joined_entry {
    ($method_name:ident, $select_name:ident, $op:expr; $( ($child:ident) ),+ $(,)?) => {
        #[doc = concat!(
            "Convert this `Select<M>` into a joined-table inheritance polymorphic query (",
            stringify!($op),
            ")."
        )]
        ///
        /// This LEFT JOINs every child table and returns the matching
        #[doc = concat!("`", stringify!($select_name), "`.")]
        ///
        /// Notes:
        /// - Requires `M` to be a joined-inheritance base model (`inheritance="joined"`).
        /// - Requires every child to be a joined-inheritance child with `inherits="M"`.
        /// - This always projects full base + child columns (ignores custom `columns(...)`),
        ///   since hydration depends on a complete prefixed projection.
        #[must_use]
        pub fn $method_name<$($child: Model),+>(mut self) -> $select_name<M, $($child),+> {
            let inh_base = M::inheritance();
            tracing::debug!(
                target: "sqlmodel_query::inheritance",
                model = %M::TABLE_NAME,
                strategy = ?inh_base.strategy,
                parent = ?inh_base.parent,
                discriminator = ?inh_base.discriminator_column,
                "resolved inheritance mapping"
            );
            $(
                let inh_child = $child::inheritance();
                tracing::debug!(
                    target: "sqlmodel_query::inheritance",
                    model = %$child::TABLE_NAME,
                    strategy = ?inh_child.strategy,
                    parent = ?inh_child.parent,
                    discriminator = ?inh_child.discriminator_column,
                    "resolved inheritance mapping"
                );
            )+
            self.aliased_projection = [table_columns::<M>(), $(table_columns::<$child>()),+].concat();
            $(
                if let Some(join) = polymorphic_joined_left_join::<M, $child>() {
                    self.joins.push(join);
                }
            )+

            $select_name {
                select: self,
                _marker: PhantomData,
            }
        }
    };
}

impl<M: Model> Select<M> {
    /// Create a new SELECT query for the model's table.
    pub fn new() -> Self {
        let inh = M::inheritance();
        if inh.strategy != sqlmodel_core::InheritanceStrategy::None {
            tracing::debug!(
                target: "sqlmodel_query::inheritance",
                model = %M::TABLE_NAME,
                strategy = ?inh.strategy,
                parent = ?inh.parent,
                discriminator = ?inh.discriminator_column,
                "resolved inheritance mapping"
            );
        }
        Self {
            columns: Vec::new(),
            aliased_projection: Vec::new(),
            where_clause: None,
            order_by: Vec::new(),
            joins: Vec::new(),
            limit: None,
            offset: None,
            group_by: Vec::new(),
            having: None,
            distinct: false,
            for_update: false,
            eager_loader: None,
            _marker: PhantomData,
        }
    }

    /// Select specific columns.
    pub fn columns(mut self, cols: &[&str]) -> Self {
        self.columns = cols.iter().map(|&s| s.to_string()).collect();
        self
    }

    /// Add a WHERE condition.
    pub fn filter(mut self, expr: Expr) -> Self {
        self.where_clause = Some(match self.where_clause {
            Some(existing) => existing.and(expr),
            None => Where::new(expr),
        });
        self
    }

    /// Add an OR WHERE condition.
    pub fn or_filter(mut self, expr: Expr) -> Self {
        self.where_clause = Some(match self.where_clause {
            Some(existing) => existing.or(expr),
            None => Where::new(expr),
        });
        self
    }

    /// Add ORDER BY clause.
    pub fn order_by(mut self, order: OrderBy) -> Self {
        self.order_by.push(order);
        self
    }

    /// Add a JOIN clause.
    pub fn join(mut self, join: Join) -> Self {
        self.joins.push(join);
        self
    }

    /// Set LIMIT.
    pub fn limit(mut self, n: u64) -> Self {
        self.limit = Some(Limit(n));
        self
    }

    /// Set OFFSET.
    pub fn offset(mut self, n: u64) -> Self {
        self.offset = Some(Offset(n));
        self
    }

    /// Add GROUP BY columns.
    pub fn group_by(mut self, cols: &[&str]) -> Self {
        self.group_by.extend(cols.iter().map(|&s| s.to_string()));
        self
    }

    /// Add HAVING condition.
    pub fn having(mut self, expr: Expr) -> Self {
        self.having = Some(match self.having {
            Some(existing) => existing.and(expr),
            None => Where::new(expr),
        });
        self
    }

    /// Make this a DISTINCT query.
    pub fn distinct(mut self) -> Self {
        self.distinct = true;
        self
    }

    /// Add FOR UPDATE lock.
    pub fn for_update(mut self) -> Self {
        self.for_update = true;
        self
    }

    /// Configure eager loading for relationships.
    ///
    /// # Example
    ///
    /// ```ignore
    /// let heroes = select!(Hero)
    ///     .eager(EagerLoader::new().include("team"))
    ///     .all_eager(cx, &conn)
    ///     .await?;
    /// ```
    pub fn eager(mut self, loader: EagerLoader<M>) -> Self {
        self.eager_loader = Some(loader);
        self
    }

    polymorphic_joined_entry!(polymorphic_joined, PolymorphicJoinedSelect, "polymorphic_joined"; (Child),);

    polymorphic_joined_entry!(polymorphic_joined2, PolymorphicJoinedSelect2, "polymorphic_joined2"; (C1), (C2));

    polymorphic_joined_entry!(polymorphic_joined3, PolymorphicJoinedSelect3, "polymorphic_joined3"; (C1), (C2), (C3));

    polymorphic_joined_entry!(polymorphic_joined4, PolymorphicJoinedSelect4, "polymorphic_joined4"; (C1), (C2), (C3), (C4));

    polymorphic_joined_entry!(polymorphic_joined5, PolymorphicJoinedSelect5, "polymorphic_joined5"; (C1), (C2), (C3), (C4), (C5));

    polymorphic_joined_entry!(polymorphic_joined6, PolymorphicJoinedSelect6, "polymorphic_joined6"; (C1), (C2), (C3), (C4), (C5), (C6));

    polymorphic_joined_entry!(polymorphic_joined7, PolymorphicJoinedSelect7, "polymorphic_joined7"; (C1), (C2), (C3), (C4), (C5), (C6), (C7));

    polymorphic_joined_entry!(polymorphic_joined8, PolymorphicJoinedSelect8, "polymorphic_joined8"; (C1), (C2), (C3), (C4), (C5), (C6), (C7), (C8));

    /// Convert this `Select<M>` into a concrete-table inheritance polymorphic
    /// query over one child table.
    ///
    /// The child tables are combined with `UNION ALL`: every branch projects
    /// the unified column set (typed `CAST(NULL AS ..)` fillers where a branch
    /// lacks a column) plus a `__type` tag naming the branch's table, so each
    /// row hydrates to the right child model in one round trip. Filters apply
    /// per branch; ORDER BY and LIMIT/OFFSET apply to the union.
    #[must_use]
    pub fn polymorphic_concrete<Child: Model>(self) -> PolymorphicConcreteSelect<M, Child> {
        PolymorphicConcreteSelect {
            select: self,
            _marker: PhantomData,
        }
    }

    /// Concrete-table inheritance polymorphic query over two child tables.
    ///
    /// Returns `PolymorphicConcrete2<M, C1, C2>` rows; see
    /// [`Select::polymorphic_concrete`] for the shape of the generated SQL.
    #[must_use]
    pub fn polymorphic_concrete2<C1: Model, C2: Model>(
        self,
    ) -> PolymorphicConcreteSelect2<M, C1, C2> {
        PolymorphicConcreteSelect2 {
            select: self,
            _marker: PhantomData,
        }
    }

    /// Concrete-table inheritance polymorphic query over three child tables.
    ///
    /// Returns `PolymorphicConcrete3<M, C1, C2, C3>` rows; see
    /// [`Select::polymorphic_concrete`] for the shape of the generated SQL.
    #[must_use]
    pub fn polymorphic_concrete3<C1: Model, C2: Model, C3: Model>(
        self,
    ) -> PolymorphicConcreteSelect3<M, C1, C2, C3> {
        PolymorphicConcreteSelect3 {
            select: self,
            _marker: PhantomData,
        }
    }

    /// Build SQL for eager loading with JOINs using a specific dialect.
    ///
    /// Generates SELECT with aliased columns and LEFT JOINs for included relationships.
    #[tracing::instrument(level = "trace", skip(self))]
    fn build_eager_with_dialect(
        &self,
        dialect: Dialect,
    ) -> (String, Vec<Value>, Vec<EagerJoinInfo>) {
        let mut sql = String::new();
        let mut params = Vec::new();
        let mut join_info = Vec::new();
        let mut where_clause = self.where_clause.clone();
        let mut joins = self.joins.clone();

        // Single-table inheritance child models should be implicitly filtered by their discriminator.
        if let Some(expr) = sti_discriminator_filter::<M>() {
            where_clause = Some(match where_clause {
                Some(existing) => existing.and(expr),
                None => Where::new(expr),
            });
        }

        if let Some(join) = joined_inheritance_join::<M>() {
            joins.insert(0, join);
        }

        // Collect parent table columns (database column names).
        let parent_cols: Vec<&str> = M::fields().iter().map(|f| f.column_name).collect();

        // Start with SELECT DISTINCT to avoid duplicates from JOINs
        sql.push_str("SELECT ");
        if self.distinct {
            sql.push_str("DISTINCT ");
        }

        // Build column list with model's table aliased
        let mut col_parts = build_aliased_column_parts(dialect, M::TABLE_NAME, &parent_cols);

        // For joined inheritance, also project parent-table columns so `#[sqlmodel(parent)]`
        // hydration can build the embedded parent model from `row.subset_by_prefix(parent_table)`.
        if let Some((parent_table, parent_fields_fn)) = joined_inheritance_parent::<M>() {
            let parent_cols: Vec<&str> = parent_fields_fn().iter().map(|f| f.column_name).collect();
            col_parts.extend(build_aliased_column_parts(
                dialect,
                parent_table,
                &parent_cols,
            ));
        }

        // Add columns for each eagerly loaded relationship
        if let Some(loader) = &self.eager_loader {
            for include in loader.includes() {
                if let Some(rel) = find_relationship::<M>(include.relationship) {
                    join_info.push(EagerJoinInfo {
                        relationship_name: include.relationship,
                        related_table: rel.related_table,
                        related_pk: (rel.related_fields_fn)()
                            .iter()
                            .filter(|f| f.primary_key)
                            .map(|f| f.column_name)
                            .collect(),
                        kind: rel.kind,
                        nested: include.nested.clone(),
                    });

                    // Add aliased columns for related table so callers can use
                    // `row.subset_by_prefix(rel.related_table)` deterministically.
                    let related_cols: Vec<&str> = (rel.related_fields_fn)()
                        .iter()
                        .map(|f| f.column_name)
                        .collect();
                    col_parts.extend(build_aliased_column_parts(
                        dialect,
                        rel.related_table,
                        &related_cols,
                    ));
                }
            }
        }

        sql.push_str(&col_parts.join(", "));

        // FROM
        sql.push_str(" FROM ");
        sql.push_str(&dialect.quote_table(M::TABLE_NAME));

        // Add JOINs for eager loading
        if let Some(loader) = &self.eager_loader {
            for include in loader.includes() {
                if let Some(rel) = find_relationship::<M>(include.relationship) {
                    let (join_sql, join_params) =
                        build_join_clause(dialect, M::TABLE_NAME, rel, params.len());
                    sql.push_str(&join_sql);
                    params.extend(join_params);
                }
            }
        }

        // Additional explicit JOINs (plus joined-inheritance join if present)
        for join in &joins {
            sql.push_str(&join.build_with_dialect(dialect, &mut params, 0));
        }

        // WHERE
        if let Some(where_clause) = &where_clause {
            let (where_sql, where_params) = where_clause.build_with_dialect(dialect, params.len());
            sql.push_str(" WHERE ");
            sql.push_str(&where_sql);
            params.extend(where_params);
        }

        // GROUP BY
        if !self.group_by.is_empty() {
            sql.push_str(" GROUP BY ");
            sql.push_str(&self.group_by.join(", "));
        }

        // HAVING
        if let Some(having) = &self.having {
            let (having_sql, having_params) = having.build_with_dialect(dialect, params.len());
            sql.push_str(" HAVING ");
            sql.push_str(&having_sql);
            params.extend(having_params);
        }

        // ORDER BY
        if !self.order_by.is_empty() {
            sql.push_str(" ORDER BY ");
            let order_strs: Vec<_> = self
                .order_by
                .iter()
                .map(|o| o.build(dialect, &mut params, 0))
                .collect();
            sql.push_str(&order_strs.join(", "));
        }

        // LIMIT
        if let Some(Limit(n)) = self.limit {
            sql.push_str(&format!(" LIMIT {}", n));
        }

        // OFFSET
        if let Some(Offset(n)) = self.offset {
            sql.push_str(&format!(" OFFSET {}", n));
        }

        (sql, params, join_info)
    }

    /// Execute the query with eager loading and return hydrated models.
    ///
    /// This method fetches the parent models along with their eagerly loaded
    /// relationships in a single query using JOINs. Results are deduplicated
    /// by primary key to handle one-to-many JOINs.
    ///
    /// # Note
    ///
    /// Currently, this method parses parent models from aliased columns and
    /// deduplicates by primary key. Full hydration of `Related<T>` and
    /// `RelatedMany<T>` fields requires macro support and is tracked
    /// separately. The JOIN query is still valuable as it:
    /// - Fetches all data in a single query (avoiding N+1)
    /// - Returns related data that can be accessed via `row.subset_by_prefix()`
    ///
    /// # Example
    ///
    /// ```ignore
    /// let heroes = select!(Hero)
    ///     .eager(EagerLoader::new().include("team"))
    ///     .all_eager(cx, &conn)
    ///     .await?;
    /// ```
    #[tracing::instrument(level = "debug", skip(self, cx, conn))]
    pub async fn all_eager<C: Connection>(
        self,
        cx: &Cx,
        conn: &C,
    ) -> Outcome<Vec<M>, sqlmodel_core::Error> {
        // If no eager loading configured, fall back to regular all()
        if !self.eager_loader.as_ref().is_some_and(|e| e.has_includes()) {
            tracing::trace!("No eager loading configured, falling back to regular all()");
            return self.all(cx, conn).await;
        }

        let (sql, params, join_info) = self.build_eager_with_dialect(conn.dialect());

        tracing::debug!(
            table = M::TABLE_NAME,
            includes = join_info.len(),
            "Executing eager loading query"
        );
        tracing::trace!(sql = %sql, "Eager SQL");

        let rows = match conn.query(cx, &sql, &params).await {
            Outcome::Ok(rows) => rows,
            Outcome::Err(e) => return Outcome::Err(e),
            Outcome::Cancelled(r) => return Outcome::Cancelled(r),
            Outcome::Panicked(p) => return Outcome::Panicked(p),
        };
        tracing::debug!(row_count = rows.len(), "Processing eager query results");

        // A JOIN repeats the parent row once per related row (and a second
        // to-many include multiplies them). Group the related rows under one
        // parent per primary key, in first-seen order, then hand each group to
        // the model so it can load its relationship fields.
        struct Pending<M> {
            model: M,
            related: Vec<Vec<Row>>,
            seen: Vec<HashSet<String>>,
        }
        let mut pending: Vec<Pending<M>> = Vec::new();
        let mut index: HashMap<String, usize> = HashMap::new();

        for row in &rows {
            let parent_row = row.subset_by_prefix(M::TABLE_NAME);
            // A row without the parent prefix can only come from a hand-built
            // query; parse it as-is so such callers keep working.
            let parsed = if parent_row.is_empty() {
                tracing::warn!(
                    table = M::TABLE_NAME,
                    "Row has no columns with parent table prefix"
                );
                M::from_row(row)
            } else {
                M::from_row(&parent_row)
            };
            let model = match parsed {
                Ok(model) => model,
                Err(e) => {
                    tracing::debug!(error = %e, "Failed to parse model from eager row");
                    return Outcome::Err(e);
                }
            };
            let key = format!("{:?}", model.primary_key_value());
            let slot = if let Some(slot) = index.get(&key) {
                *slot
            } else {
                pending.push(Pending {
                    model,
                    related: vec![Vec::new(); join_info.len()],
                    seen: vec![HashSet::new(); join_info.len()],
                });
                index.insert(key, pending.len() - 1);
                pending.len() - 1
            };

            for (i, info) in join_info.iter().enumerate() {
                if !row.has_prefix(info.related_table) || row.prefix_is_all_null(info.related_table)
                {
                    continue;
                }
                let related = row.subset_by_prefix(info.related_table);
                let related_key = if info.related_pk.is_empty() {
                    format!("{:?}", related.values().collect::<Vec<_>>())
                } else {
                    let pk: Vec<Option<&Value>> = info
                        .related_pk
                        .iter()
                        .map(|c| related.get_by_name(c))
                        .collect();
                    if pk.iter().all(|v| v.is_none_or(Value::is_null)) {
                        // LEFT JOIN with no match for this include.
                        continue;
                    }
                    format!("{pk:?}")
                };
                let group = &mut pending[slot];
                if group.seen[i].insert(related_key) {
                    group.related[i].push(related);
                }
            }
        }

        let mut models = Vec::with_capacity(pending.len());
        for group in pending {
            let mut model = group.model;
            for (i, info) in join_info.iter().enumerate() {
                if let Err(e) =
                    model.hydrate_relationship(info.relationship_name, &group.related[i])
                {
                    return Outcome::Err(e);
                }
            }
            models.push(model);
        }

        tracing::debug!(
            unique_models = models.len(),
            "Eager loading complete (deduplicated and hydrated)"
        );
        Outcome::Ok(models)
    }

    /// Build the SQL query and parameters.
    pub fn build(&self) -> (String, Vec<Value>) {
        self.build_with_dialect(Dialect::default())
    }

    /// The statement [`Self::all_eager`] runs for `dialect`: every column of
    /// this model and of each included relationship aliased as
    /// `table__column`, joined with `LEFT JOIN`s. Falls back to
    /// [`Self::build_with_dialect`] when nothing is included.
    pub fn build_eager_sql_with_dialect(&self, dialect: Dialect) -> (String, Vec<Value>) {
        if !self.eager_loader.as_ref().is_some_and(|e| e.has_includes()) {
            return self.build_with_dialect(dialect);
        }
        let (sql, params, _) = self.build_eager_with_dialect(dialect);
        (sql, params)
    }

    /// Build the SQL query and parameters with a specific dialect.
    pub fn build_with_dialect(&self, dialect: Dialect) -> (String, Vec<Value>) {
        let mut sql = String::new();
        let mut params = Vec::new();
        let mut where_clause = self.where_clause.clone();
        let mut joins = self.joins.clone();

        // Single-table inheritance child models should be implicitly filtered by their discriminator.
        if let Some(expr) = sti_discriminator_filter::<M>() {
            where_clause = Some(match where_clause {
                Some(existing) => existing.and(expr),
                None => Where::new(expr),
            });
        }

        if let Some(join) = joined_inheritance_join::<M>() {
            joins.insert(0, join);
        }

        // SELECT
        sql.push_str("SELECT ");
        if self.distinct {
            sql.push_str("DISTINCT ");
        }

        if let Some(cols) = joined_inheritance_select_columns::<M>(dialect) {
            sql.push_str(&cols.join(", "));
        } else if !self.aliased_projection.is_empty() {
            // Polymorphic joined selects: full base + child projections, quoted
            // for the dialect now that it is known.
            sql.push_str(&render_aliased_projection(
                dialect,
                &self.aliased_projection,
            ));
        } else if self.columns.is_empty() {
            if joins.is_empty() {
                sql.push('*');
            } else {
                // With a JOIN, `*` also returns the joined table's columns, and
                // a same-named column (`id`, `name`) from the joined table wins
                // the by-name lookup: the model was silently mapped from the
                // wrong table. Project this model's own columns instead.
                sql.push_str(&dialect.quote_table(M::TABLE_NAME));
                sql.push_str(".*");
            }
        } else {
            sql.push_str(&self.columns.join(", "));
        }

        // FROM
        sql.push_str(" FROM ");
        sql.push_str(&dialect.quote_table(M::TABLE_NAME));

        // JOINs
        for join in &joins {
            sql.push_str(&join.build_with_dialect(dialect, &mut params, 0));
        }

        // WHERE
        if let Some(where_clause) = &where_clause {
            let (where_sql, where_params) = where_clause.build_with_dialect(dialect, params.len());
            sql.push_str(" WHERE ");
            sql.push_str(&where_sql);
            params.extend(where_params);
        }

        // GROUP BY
        if !self.group_by.is_empty() {
            sql.push_str(" GROUP BY ");
            sql.push_str(&self.group_by.join(", "));
        }

        // HAVING
        if let Some(having) = &self.having {
            let (having_sql, having_params) = having.build_with_dialect(dialect, params.len());
            sql.push_str(" HAVING ");
            sql.push_str(&having_sql);
            params.extend(having_params);
        }

        // ORDER BY
        if !self.order_by.is_empty() {
            sql.push_str(" ORDER BY ");
            let order_strs: Vec<_> = self
                .order_by
                .iter()
                .map(|o| o.build(dialect, &mut params, 0))
                .collect();
            sql.push_str(&order_strs.join(", "));
        }

        // LIMIT
        if let Some(Limit(n)) = self.limit {
            sql.push_str(&format!(" LIMIT {}", n));
        }

        // OFFSET
        if let Some(Offset(n)) = self.offset {
            sql.push_str(&format!(" OFFSET {}", n));
        }

        // FOR UPDATE
        if self.for_update {
            sql.push_str(" FOR UPDATE");
        }

        (sql, params)
    }

    /// Convert this SELECT query to an EXISTS expression.
    ///
    /// Creates an `Expr::Exists` that can be used in WHERE clauses of other queries.
    /// For performance, the SELECT is automatically optimized to `SELECT 1` when
    /// generating the EXISTS subquery.
    ///
    /// # Example
    ///
    /// ```ignore
    /// // Find customers who have at least one order
    /// let has_orders = Select::<Order>::new()
    ///     .filter(Expr::raw("orders.customer_id = customers.id"))
    ///     .into_exists();
    ///
    /// let customers = Select::<Customer>::new()
    ///     .filter(has_orders)
    ///     .all(cx, &conn)
    ///     .await?;
    ///
    /// // Generates: SELECT * FROM customers WHERE EXISTS (SELECT 1 FROM orders WHERE orders.customer_id = customers.id)
    /// ```
    pub fn into_exists(self) -> Expr {
        Expr::exists_query(self.into_query())
    }

    /// Convert this SELECT query to an EXISTS expression using a specific dialect.
    ///
    /// Use this when embedding the EXISTS in a query for a non-default dialect.
    pub fn into_exists_with_dialect(self, dialect: Dialect) -> Expr {
        let (sql, params) = self.build_exists_subquery_with_dialect(dialect);
        Expr::exists(sql, params)
    }

    /// Convert this SELECT query to a NOT EXISTS expression.
    ///
    /// Creates an `Expr::Exists` (negated) that can be used in WHERE clauses.
    /// For performance, the SELECT is automatically optimized to `SELECT 1`.
    ///
    /// # Example
    ///
    /// ```ignore
    /// // Find customers with no orders
    /// let has_no_orders = Select::<Order>::new()
    ///     .filter(Expr::raw("orders.customer_id = customers.id"))
    ///     .into_not_exists();
    ///
    /// let customers = Select::<Customer>::new()
    ///     .filter(has_no_orders)
    ///     .all(cx, &conn)
    ///     .await?;
    ///
    /// // Generates: SELECT * FROM customers WHERE NOT EXISTS (SELECT 1 FROM orders WHERE orders.customer_id = customers.id)
    /// ```
    pub fn into_not_exists(self) -> Expr {
        Expr::not_exists_query(self.into_query())
    }

    /// Convert this SELECT query to a NOT EXISTS expression using a specific dialect.
    pub fn into_not_exists_with_dialect(self, dialect: Dialect) -> Expr {
        let (sql, params) = self.build_exists_subquery_with_dialect(dialect);
        Expr::not_exists(sql, params)
    }

    /// Convert this SELECT into a LATERAL JOIN.
    ///
    /// Creates a `Join` with `lateral: true` that can be added to another query.
    /// The subquery can reference columns from the outer query.
    ///
    /// Supported by PostgreSQL (9.3+) and MySQL (8.0.14+). Not supported by SQLite.
    ///
    /// # Arguments
    ///
    /// * `alias` - Required alias for the lateral subquery
    /// * `join_type` - The join type (typically `Left` or `Inner`)
    /// * `on` - ON condition (use `Expr::raw("TRUE")` for implicit TRUE)
    ///
    /// # Example
    ///
    /// ```ignore
    /// // Get top 3 recent orders per customer
    /// let recent_orders = Select::<Order>::new()
    ///     .filter(Expr::raw("orders.customer_id = customers.id"))
    ///     .order_by(OrderBy::desc("date"))
    ///     .limit(3)
    ///     .into_lateral_join("recent_orders", JoinType::Left, Expr::raw("TRUE"));
    ///
    /// let query = Select::<Customer>::new().join(recent_orders);
    /// ```
    pub fn into_lateral_join(
        self,
        alias: impl Into<String>,
        join_type: crate::JoinType,
        on: Expr,
    ) -> crate::Join {
        crate::Join::lateral_query(join_type, self.into_query(), alias, on)
    }

    /// Convert this SELECT into a LATERAL JOIN using a specific dialect.
    pub fn into_lateral_join_with_dialect(
        self,
        alias: impl Into<String>,
        join_type: crate::JoinType,
        on: Expr,
        dialect: Dialect,
    ) -> crate::Join {
        let (sql, params) = self.into_query().build_with_dialect(dialect);
        crate::Join::lateral(join_type, sql, alias, on, params)
    }

    /// Convert this SELECT into a dialect-independent subquery for embedding
    /// in another statement (`Expr::in_query`, `Expr::exists_query`, lateral
    /// joins). The subquery is rendered when the enclosing statement is.
    pub fn into_query(self) -> SelectQuery {
        let Select {
            columns,
            // Polymorphic projections are not meaningful inside a subquery.
            aliased_projection: _,
            where_clause,
            order_by,
            joins,
            limit,
            offset,
            group_by,
            having,
            distinct,
            for_update,
            eager_loader: _,
            _marker: _,
        } = self;

        let mut where_clause = where_clause;
        if let Some(expr) = sti_discriminator_filter::<M>() {
            where_clause = Some(match where_clause {
                Some(existing) => existing.and(expr),
                None => Where::new(expr),
            });
        }

        let mut joins = joins;
        if let Some(join) = joined_inheritance_join::<M>() {
            joins.insert(0, join);
        }

        SelectQuery {
            table: M::TABLE_NAME.to_string(),
            columns,
            where_clause,
            order_by,
            joins,
            limit,
            offset,
            group_by,
            having,
            distinct,
            for_update,
        }
    }

    fn build_exists_subquery_with_dialect(&self, dialect: Dialect) -> (String, Vec<Value>) {
        let mut sql = String::new();
        let mut params = Vec::new();
        let mut where_clause = self.where_clause.clone();
        let mut joins = self.joins.clone();

        if let Some(expr) = sti_discriminator_filter::<M>() {
            where_clause = Some(match where_clause {
                Some(existing) => existing.and(expr),
                None => Where::new(expr),
            });
        }

        if let Some(join) = joined_inheritance_join::<M>() {
            joins.insert(0, join);
        }

        // SELECT 1 for optimal EXISTS performance
        sql.push_str("SELECT 1 FROM ");
        sql.push_str(&dialect.quote_table(M::TABLE_NAME));

        // JOINs (if any)
        for join in &joins {
            sql.push_str(&join.build_with_dialect(dialect, &mut params, 0));
        }

        // WHERE
        if let Some(where_clause) = &where_clause {
            let (where_sql, where_params) = where_clause.build_with_dialect(dialect, params.len());
            sql.push_str(" WHERE ");
            sql.push_str(&where_sql);
            params.extend(where_params);
        }

        // GROUP BY (rare in EXISTS but supported)
        if !self.group_by.is_empty() {
            sql.push_str(" GROUP BY ");
            sql.push_str(&self.group_by.join(", "));
        }

        // HAVING (rare in EXISTS but supported)
        if let Some(having) = &self.having {
            let (having_sql, having_params) = having.build_with_dialect(dialect, params.len());
            sql.push_str(" HAVING ");
            sql.push_str(&having_sql);
            params.extend(having_params);
        }

        // Note: ORDER BY, LIMIT, OFFSET are omitted in EXISTS subquery as they have no effect

        (sql, params)
    }

    /// Execute the query and return all matching rows as models.
    pub async fn all<C: Connection>(
        self,
        cx: &Cx,
        conn: &C,
    ) -> Outcome<Vec<M>, sqlmodel_core::Error> {
        let (sql, params) = self.build_with_dialect(conn.dialect());
        let rows = conn.query(cx, &sql, &params).await;

        rows.and_then(|rows| {
            let mut models = Vec::with_capacity(rows.len());
            for row in &rows {
                match M::from_row(row) {
                    Ok(model) => models.push(model),
                    Err(e) => return Outcome::Err(e),
                }
            }
            Outcome::Ok(models)
        })
    }

    /// Execute the query and return the first matching row.
    pub async fn first<C: Connection>(
        self,
        cx: &Cx,
        conn: &C,
    ) -> Outcome<Option<M>, sqlmodel_core::Error> {
        let query = self.limit(1);
        let (sql, params) = query.build_with_dialect(conn.dialect());
        let row = conn.query_one(cx, &sql, &params).await;

        row.and_then(|opt_row| match opt_row {
            Some(row) => match M::from_row(&row) {
                Ok(model) => Outcome::Ok(Some(model)),
                Err(e) => Outcome::Err(e),
            },
            None => Outcome::Ok(None),
        })
    }

    /// Execute the query and return exactly one row, or error.
    pub async fn one<C: Connection>(self, cx: &Cx, conn: &C) -> Outcome<M, sqlmodel_core::Error> {
        match self.one_or_none(cx, conn).await {
            Outcome::Ok(Some(model)) => Outcome::Ok(model),
            Outcome::Ok(None) => Outcome::Err(sqlmodel_core::Error::Custom(
                "Expected one row, found none".to_string(),
            )),
            Outcome::Err(e) => Outcome::Err(e),
            Outcome::Cancelled(r) => Outcome::Cancelled(r),
            Outcome::Panicked(p) => Outcome::Panicked(p),
        }
    }

    /// Execute the query and return zero or one row, or error on multiple rows.
    pub async fn one_or_none<C: Connection>(
        self,
        cx: &Cx,
        conn: &C,
    ) -> Outcome<Option<M>, sqlmodel_core::Error> {
        // Fetch up to two rows so we can enforce exact-one semantics without
        // scanning the full result set.
        let mut query = self;
        query.limit = Some(Limit(2));
        let (sql, params) = query.build_with_dialect(conn.dialect());
        let rows = conn.query(cx, &sql, &params).await;

        rows.and_then(|rows| match rows.len() {
            0 => Outcome::Ok(None),
            1 => match M::from_row(&rows[0]) {
                Ok(model) => Outcome::Ok(Some(model)),
                Err(e) => Outcome::Err(e),
            },
            n => Outcome::Err(sqlmodel_core::Error::Custom(format!(
                "Expected zero or one row, found {n}"
            ))),
        })
    }

    /// Execute the query and return the count of matching rows.
    pub async fn count<C: Connection>(
        self,
        cx: &Cx,
        conn: &C,
    ) -> Outcome<u64, sqlmodel_core::Error> {
        let mut count_query = self;
        count_query.columns = vec!["COUNT(*) as count".to_string()];
        count_query.order_by.clear();
        count_query.limit = None;
        count_query.offset = None;

        let (sql, params) = count_query.build_with_dialect(conn.dialect());
        let row = conn.query_one(cx, &sql, &params).await;

        row.and_then(|opt_row| match opt_row {
            Some(row) => match row.get_named::<i64>("count") {
                Ok(count) => Outcome::Ok(count as u64),
                Err(e) => Outcome::Err(e),
            },
            None => Outcome::Ok(0),
        })
    }

    /// Check if any rows match the query.
    pub async fn exists<C: Connection>(
        self,
        cx: &Cx,
        conn: &C,
    ) -> Outcome<bool, sqlmodel_core::Error> {
        let count = self.count(cx, conn).await;
        count.map(|n| n > 0)
    }
}

impl<M: Model> Default for Select<M> {
    fn default() -> Self {
        Self::new()
    }
}

fn polymorphic_joined_left_join<Base: Model, Child: Model>() -> Option<Join> {
    // Must have a PK to join on.
    let pks = Base::PRIMARY_KEY;
    if pks.is_empty() {
        return None;
    }

    let mut on =
        Expr::qualified(Base::TABLE_NAME, pks[0]).eq(Expr::qualified(Child::TABLE_NAME, pks[0]));
    for pk in &pks[1..] {
        on = on.and(
            Expr::qualified(Base::TABLE_NAME, *pk).eq(Expr::qualified(Child::TABLE_NAME, *pk)),
        );
    }

    Some(Join::left(Child::TABLE_NAME, on))
}

// ==================== Joined-table inheritance polymorphic selects ====================
//
// One definition generates the arity family (N = 1..8). The public names of
// the original hand-written arities (`PolymorphicJoined{,2,3}` /
// `PolymorphicJoinedSelect{,2,3}` / `Select::polymorphic_joined{,2,3}`) are
// kept as generated instances.

/// The ambiguity guard shared by every arity: errors when more than one
/// child prefix is non-NULL. Returning cleanly means hydration may proceed
/// with "first non-NULL child, else base" semantics.
#[allow(clippy::result_large_err)]
fn joined_row_guard(
    row: &Row,
    child_tables: &[&'static str],
    op: &str,
) -> Result<(), sqlmodel_core::Error> {
    let non_null: Vec<&'static str> = child_tables
        .iter()
        .filter(|table| !row.prefix_is_all_null(table))
        .copied()
        .collect();
    if non_null.len() > 1 {
        tracing::warn!(
            target: "sqlmodel_query::inheritance",
            op = op,
            prefixes = ?non_null,
            "ambiguous row in polymorphic hydration: multiple child prefixes are non-NULL"
        );
        return Err(sqlmodel_core::Error::Custom(format!(
            "{op} ambiguous row: multiple child prefixes are non-NULL: {}",
            non_null.join(", ")
        )));
    }
    Ok(())
}

#[allow(clippy::result_large_err)]
fn joined_base_check<Base: Model>(op: &str) -> Result<(), sqlmodel_core::Error> {
    let inh_base = Base::inheritance();
    if inh_base.strategy != sqlmodel_core::InheritanceStrategy::Joined || inh_base.parent.is_some()
    {
        return Err(sqlmodel_core::Error::Custom(format!(
            "{op} requires a joined-inheritance base model; got strategy={:?}, parent={:?} for {}",
            inh_base.strategy,
            inh_base.parent,
            Base::TABLE_NAME
        )));
    }
    Ok(())
}

#[allow(clippy::result_large_err)]
fn joined_child_check<Base: Model, Child: Model>(op: &str) -> Result<(), sqlmodel_core::Error> {
    let inh_child = Child::inheritance();
    if inh_child.strategy != sqlmodel_core::InheritanceStrategy::Joined
        || inh_child.parent != Some(Base::TABLE_NAME)
    {
        return Err(sqlmodel_core::Error::Custom(format!(
            "{op} requires a joined-inheritance child with parent={}; got strategy={:?}, parent={:?} for {}",
            Base::TABLE_NAME,
            inh_child.strategy,
            inh_child.parent,
            Child::TABLE_NAME
        )));
    }
    Ok(())
}

/// Generate one joined-table polymorphic arity: the result enum, the select
/// wrapper, and its `all()` execution. The `(variant, child)` pairs pair each
/// enum variant name with the child model type.
macro_rules! define_polymorphic_joined {
    (
        $(#[$enum_meta:meta])*
        enum $enum_name:ident, select $select_name:ident,
        method $method_name:ident = $op:expr,
        base $base:ident, children [ $( ($variant:ident, $child:ident) ),+ $(,)? ]
    ) => {
        $(#[$enum_meta])*
        #[derive(Debug, Clone, PartialEq)]
        pub enum $enum_name<Base: Model, $($child: Model),+> {
            Base(Base),
            $($variant($child),)+
        }

        /// A polymorphic SELECT for joined-table inheritance: the base plus
        /// every child type, one `LEFT JOIN` per child.
        #[derive(Debug, Clone)]
        pub struct $select_name<Base: Model, $($child: Model),+> {
            select: Select<Base>,
            _marker: PhantomData<($($child,)+)>,
        }

        impl<Base: Model, $($child: Model),+> $select_name<Base, $($child),+> {
            /// Add a WHERE condition (delegates to the underlying base select).
            #[must_use]
            pub fn filter(mut self, expr: Expr) -> Self {
                self.select = self.select.filter(expr);
                self
            }

            /// Add ORDER BY clause (delegates to the underlying base select).
            #[must_use]
            pub fn order_by(mut self, order: OrderBy) -> Self {
                self.select = self.select.order_by(order);
                self
            }

            /// Set LIMIT (delegates to the underlying base select).
            #[must_use]
            pub fn limit(mut self, n: u64) -> Self {
                self.select = self.select.limit(n);
                self
            }

            /// Set OFFSET (delegates to the underlying base select).
            #[must_use]
            pub fn offset(mut self, n: u64) -> Self {
                self.select = self.select.offset(n);
                self
            }

            /// Build the SQL query and parameters.
            pub fn build_with_dialect(&self, dialect: Dialect) -> (String, Vec<Value>) {
                let (sql, params) = self.select.build_with_dialect(dialect);
                let prefixes: &[&str] = &[$($child::TABLE_NAME),+];
                tracing::trace!(
                    target: "sqlmodel_query::polymorphic",
                    dialect = ?dialect,
                    sql = %sql,
                    params = ?params,
                    children = ?prefixes,
                    "polymorphic select"
                );
                (sql, params)
            }

            /// Execute the polymorphic query and hydrate the base or a child
            /// per row: exactly one non-NULL child prefix selects the child,
            /// all-NULL selects the base, and more than one is an error.
            #[tracing::instrument(level = "debug", skip(self, cx, conn))]
            pub async fn all<C: Connection>(
                self,
                cx: &Cx,
                conn: &C,
            ) -> Outcome<Vec<$enum_name<Base, $($child),+>>, sqlmodel_core::Error> {
                if let Err(e) = joined_base_check::<Base>($op) {
                    return Outcome::Err(e);
                }
                $(
                    if let Err(e) = joined_child_check::<Base, $child>($op) {
                        return Outcome::Err(e);
                    }
                )+

                if Base::PRIMARY_KEY.is_empty() {
                    return Outcome::Err(sqlmodel_core::Error::Custom(format!(
                        "{} requires base model {} to have a primary key",
                        $op,
                        Base::TABLE_NAME
                    )));
                }

                let (sql, params) = self.build_with_dialect(conn.dialect());
                tracing::debug!(
                    sql = %sql,
                    base = Base::TABLE_NAME,
                    $($child = $child::TABLE_NAME,)+
                    "Executing polymorphic joined SELECT"
                );

                let rows = conn.query(cx, &sql, &params).await;
                rows.and_then(|rows| {
                    let mut out = Vec::with_capacity(rows.len());
                    for row in rows {
                        if let Err(e) =
                            joined_row_guard(&row, &[$($child::TABLE_NAME),+], $op)
                        {
                            return Outcome::Err(e);
                        }
                        let mut hydrated: Option<
                            sqlmodel_core::Result<$enum_name<Base, $($child),+>>,
                        > = None;
                        $(
                            if hydrated.is_none()
                                && !row.prefix_is_all_null($child::TABLE_NAME)
                            {
                                hydrated = Some(
                                    $child::from_row(&row).map(|c| $enum_name::$variant(c))
                                );
                            }
                        )+
                        let hydrated = match hydrated {
                            Some(h) => h,
                            // No child prefix matched: a base-only row.
                            None => Base::from_row(&row).map($enum_name::Base),
                        };
                        out.push(match hydrated {
                            Ok(v) => v,
                            Err(e) => return Outcome::Err(e),
                        });
                    }
                    Outcome::Ok(out)
                })
            }
        }
    };
}

define_polymorphic_joined!(
    /// Output of a joined-table inheritance polymorphic query with a single child type.
    enum PolymorphicJoined, select PolymorphicJoinedSelect,
    method polymorphic_joined = "polymorphic_joined",
    base Base, children [(Child, Child)]
);

define_polymorphic_joined!(
    /// Output of a joined-table inheritance polymorphic query with two child types.
    enum PolymorphicJoined2, select PolymorphicJoinedSelect2,
    method polymorphic_joined2 = "polymorphic_joined2",
    base Base, children [(C1, C1), (C2, C2)]
);

define_polymorphic_joined!(
    /// Output of a joined-table inheritance polymorphic query with three child types.
    enum PolymorphicJoined3, select PolymorphicJoinedSelect3,
    method polymorphic_joined3 = "polymorphic_joined3",
    base Base, children [(C1, C1), (C2, C2), (C3, C3)]
);

define_polymorphic_joined!(
    /// Output of a joined-table inheritance polymorphic query with four child types.
    enum PolymorphicJoined4, select PolymorphicJoinedSelect4,
    method polymorphic_joined4 = "polymorphic_joined4",
    base Base, children [(C1, C1), (C2, C2), (C3, C3), (C4, C4)]
);

define_polymorphic_joined!(
    /// Output of a joined-table inheritance polymorphic query with five child types.
    enum PolymorphicJoined5, select PolymorphicJoinedSelect5,
    method polymorphic_joined5 = "polymorphic_joined5",
    base Base, children [(C1, C1), (C2, C2), (C3, C3), (C4, C4), (C5, C5)]
);

define_polymorphic_joined!(
    /// Output of a joined-table inheritance polymorphic query with six child types.
    enum PolymorphicJoined6, select PolymorphicJoinedSelect6,
    method polymorphic_joined6 = "polymorphic_joined6",
    base Base, children [(C1, C1), (C2, C2), (C3, C3), (C4, C4), (C5, C5), (C6, C6)]
);

define_polymorphic_joined!(
    /// Output of a joined-table inheritance polymorphic query with seven child types.
    enum PolymorphicJoined7, select PolymorphicJoinedSelect7,
    method polymorphic_joined7 = "polymorphic_joined7",
    base Base, children [(C1, C1), (C2, C2), (C3, C3), (C4, C4), (C5, C5), (C6, C6), (C7, C7)]
);

define_polymorphic_joined!(
    /// Output of a joined-table inheritance polymorphic query with eight child types.
    enum PolymorphicJoined8, select PolymorphicJoinedSelect8,
    method polymorphic_joined8 = "polymorphic_joined8",
    base Base,
    children [(C1, C1), (C2, C2), (C3, C3), (C4, C4), (C5, C5), (C6, C6), (C7, C7), (C8, C8)]
);

// ==================== Concrete-table inheritance polymorphic selects ====================

/// The variant tag column appended to every branch of a concrete-table union.
/// Its value is the branch child's `TABLE_NAME`, so hydration is
/// self-describing and independent of branch order.
const CONCRETE_TYPE_COLUMN: &str = "__type";

/// Ordered union of the children's columns. The first child declaring a name
/// fixes the unified column's position and type; later children sharing the
/// name must be type-compatible (the union's branch types must unify).
fn concrete_union_columns(
    children: &[&'static [sqlmodel_core::FieldInfo]],
) -> Vec<(&'static str, &'static sqlmodel_core::FieldInfo)> {
    let mut unified: Vec<(&'static str, &'static sqlmodel_core::FieldInfo)> = Vec::new();
    for fields in children {
        for field in *fields {
            if !unified.iter().any(|(name, _)| *name == field.column_name) {
                unified.push((field.column_name, field));
            }
        }
    }
    unified
}

/// The `CAST(NULL AS T)` filler for a unified column a branch does not have.
///
/// SQLite and PostgreSQL accept the dialect's full type names. MySQL's CAST
/// only supports a small target set (`SIGNED`, `DOUBLE`, `DECIMAL(p, s)`,
/// `CHAR(n)`, `DATE`, `TIME`, `DATETIME`, `BINARY`, `JSON`), so those are
/// mapped explicitly.
fn concrete_null_filler(field: &sqlmodel_core::FieldInfo, dialect: Dialect) -> String {
    use sqlmodel_core::SqlType;
    match dialect {
        Dialect::Mysql => {
            let target = match field.sql_type {
                SqlType::TinyInt
                | SqlType::SmallInt
                | SqlType::Integer
                | SqlType::BigInt
                | SqlType::Boolean => "SIGNED".to_string(),
                SqlType::Real | SqlType::Double => "DOUBLE".to_string(),
                SqlType::Date => "DATE".to_string(),
                SqlType::Time => "TIME".to_string(),
                SqlType::DateTime | SqlType::Timestamp | SqlType::TimestampTz => {
                    "DATETIME".to_string()
                }
                SqlType::Json | SqlType::JsonB | SqlType::Array(_) => "JSON".to_string(),
                SqlType::Binary(len) | SqlType::VarBinary(len) => format!("BINARY({len})"),
                SqlType::Blob | SqlType::Uuid => "BINARY".to_string(),
                SqlType::Char(len) | SqlType::VarChar(len) => format!("CHAR({len})"),
                SqlType::Text | SqlType::Enum(_) | SqlType::Custom(_) => "CHAR".to_string(),
                // DECIMAL(p, s) / NUMERIC(p, s) pass straight through.
                SqlType::Numeric { .. } | SqlType::Decimal { .. } => {
                    return format!("CAST(NULL AS {})", field.effective_sql_type_for(dialect));
                }
            };
            format!("CAST(NULL AS {target})")
        }
        _ => format!("CAST(NULL AS {})", field.effective_sql_type_for(dialect)),
    }
}

/// The invariant checks shared by every concrete-table polymorphic entry
/// point: the base is a concrete-table base, each child is a concrete-table
/// child of that base, and the base has a primary key.
#[allow(clippy::result_large_err)]
fn concrete_hierarchy_check<Base: Model>(
    op: &str,
    child_tables: &[&'static str],
) -> Result<(), sqlmodel_core::Error> {
    let inh_base = Base::inheritance();
    if inh_base.strategy != sqlmodel_core::InheritanceStrategy::Concrete
        || inh_base.parent.is_some()
    {
        return Err(sqlmodel_core::Error::Custom(format!(
            "{op} requires a concrete-inheritance base model; got strategy={:?}, parent={:?} for {}",
            inh_base.strategy,
            inh_base.parent,
            Base::TABLE_NAME
        )));
    }
    for table in child_tables {
        if Base::PRIMARY_KEY.is_empty() {
            return Err(sqlmodel_core::Error::Custom(format!(
                "{op} requires base model {} to have a primary key",
                Base::TABLE_NAME
            )));
        }
        let _ = table;
    }
    Ok(())
}

/// The per-child half of [`concrete_hierarchy_check`].
#[allow(clippy::result_large_err)]
fn concrete_child_check<Base: Model, Child: Model>(op: &str) -> Result<(), sqlmodel_core::Error> {
    let inh_child = Child::inheritance();
    if inh_child.strategy != sqlmodel_core::InheritanceStrategy::Concrete
        || inh_child.parent != Some(Base::TABLE_NAME)
    {
        return Err(sqlmodel_core::Error::Custom(format!(
            "{op} requires a concrete-inheritance child with parent={}; got strategy={:?}, parent={:?} for {}",
            Base::TABLE_NAME,
            inh_child.strategy,
            inh_child.parent,
            Child::TABLE_NAME
        )));
    }
    Ok(())
}

/// Build the concrete-table UNION ALL for `branches` (table name + fields),
/// applying the select's filters per branch and its ORDER BY / LIMIT / OFFSET
/// to the union as a whole. Placeholders stay dense across branches.
fn build_concrete_union_sql<Base: Model>(
    select: &Select<Base>,
    dialect: Dialect,
    branches: &[(&'static str, &'static [sqlmodel_core::FieldInfo])],
) -> (String, Vec<Value>) {
    let unified = concrete_union_columns(&branches.iter().map(|(_, f)| *f).collect::<Vec<_>>());
    let tag_alias = dialect.quote_identifier(CONCRETE_TYPE_COLUMN);

    let mut sql = String::new();
    let mut params: Vec<Value> = Vec::new();

    for (index, (table, fields)) in branches.iter().enumerate() {
        if index > 0 {
            sql.push_str(" UNION ALL ");
        }
        sql.push_str("SELECT ");
        let mut projections: Vec<String> = Vec::with_capacity(unified.len() + 1);
        for (name, first_field) in &unified {
            let alias = dialect.quote_identifier(name);
            let projection = match fields.iter().find(|f| f.column_name == *name) {
                Some(_) => format!(
                    "{}.{} AS {alias}",
                    dialect.quote_identifier(table),
                    dialect.quote_identifier(name)
                ),
                // A column this branch lacks: typed NULL keeps every branch's
                // projection shape identical without materializing values.
                None => format!("{} AS {alias}", concrete_null_filler(first_field, dialect)),
            };
            projections.push(projection);
        }
        projections.push(format!("'{}' AS {tag_alias}", table));
        sql.push_str(&projections.join(", "));

        sql.push_str(" FROM ");
        sql.push_str(&dialect.quote_identifier(table));

        if let Some(where_clause) = &select.where_clause {
            let (where_sql, where_params) = where_clause.build_with_dialect(dialect, params.len());
            sql.push_str(" WHERE ");
            sql.push_str(&where_sql);
            params.extend(where_params);
        }
    }

    // ORDER BY / LIMIT / OFFSET written after the last branch scope the whole
    // union on all three dialects.
    if !select.order_by.is_empty() {
        sql.push_str(" ORDER BY ");
        let param_offset = params.len();
        let order_strs: Vec<_> = select
            .order_by
            .iter()
            .map(|o| o.build(dialect, &mut params, param_offset))
            .collect();
        sql.push_str(&order_strs.join(", "));
    }
    if let Some(Limit(n)) = select.limit {
        sql.push_str(&format!(" LIMIT {n}"));
    }
    if let Some(Offset(n)) = select.offset {
        sql.push_str(&format!(" OFFSET {n}"));
    }

    (sql, params)
}

/// Hydrate one union row by its variant tag.
#[allow(clippy::result_large_err)]
fn concrete_tag(row: &Row) -> Result<String, sqlmodel_core::Error> {
    row.get_named(CONCRETE_TYPE_COLUMN)
}

/// Output of a concrete-table inheritance polymorphic query with a single child type.
#[derive(Debug, Clone, PartialEq)]
pub enum PolymorphicConcrete<Child: Model> {
    Child(Child),
}

/// A concrete-table polymorphic SELECT over base + one child table.
///
/// Construct via `select!(Base).polymorphic_concrete::<Child>()`.
#[derive(Debug, Clone)]
pub struct PolymorphicConcreteSelect<Base: Model, Child: Model> {
    select: Select<Base>,
    _marker: PhantomData<Child>,
}

impl<Base: Model, Child: Model> PolymorphicConcreteSelect<Base, Child> {
    /// Add a WHERE condition (applied to this child's branch).
    #[must_use]
    pub fn filter(mut self, expr: Expr) -> Self {
        self.select = self.select.filter(expr);
        self
    }

    /// Add ORDER BY (applied to the union).
    #[must_use]
    pub fn order_by(mut self, order: OrderBy) -> Self {
        self.select = self.select.order_by(order);
        self
    }

    /// Set LIMIT (applied to the union).
    #[must_use]
    pub fn limit(mut self, n: u64) -> Self {
        self.select = self.select.limit(n);
        self
    }

    /// Set OFFSET (applied to the union).
    #[must_use]
    pub fn offset(mut self, n: u64) -> Self {
        self.select = self.select.offset(n);
        self
    }

    /// Build the SQL query and parameters.
    pub fn build_with_dialect(&self, dialect: Dialect) -> (String, Vec<Value>) {
        let (sql, params) = build_concrete_union_sql(
            &self.select,
            dialect,
            &[(Child::TABLE_NAME, Child::fields())],
        );
        let prefixes: &[&str] = &[Child::TABLE_NAME];
        tracing::trace!(
            target: "sqlmodel_query::polymorphic",
            dialect = ?dialect,
            sql = %sql,
            params = ?params,
            children = ?prefixes,
            "polymorphic select"
        );
        (sql, params)
    }

    /// Execute the query and hydrate every row as `Child`.
    #[tracing::instrument(level = "debug", skip(self, cx, conn))]
    pub async fn all<C: Connection>(
        self,
        cx: &Cx,
        conn: &C,
    ) -> Outcome<Vec<PolymorphicConcrete<Child>>, sqlmodel_core::Error> {
        if let Err(e) =
            concrete_hierarchy_check::<Base>("polymorphic_concrete", &[Child::TABLE_NAME])
        {
            return Outcome::Err(e);
        }
        if let Err(e) = concrete_child_check::<Base, Child>("polymorphic_concrete") {
            return Outcome::Err(e);
        }

        let (sql, params) = self.build_with_dialect(conn.dialect());
        let rows = conn.query(cx, &sql, &params).await;
        rows.and_then(|rows| {
            let mut out = Vec::with_capacity(rows.len());
            for row in rows {
                let tag = match concrete_tag(&row) {
                    Ok(t) => t,
                    Err(e) => return Outcome::Err(e),
                };
                if tag != Child::TABLE_NAME {
                    return Outcome::Err(sqlmodel_core::Error::Custom(format!(
                        "polymorphic_concrete: unknown {} tag {tag:?} for base {}",
                        CONCRETE_TYPE_COLUMN,
                        Base::TABLE_NAME
                    )));
                }
                match Child::from_row(&row) {
                    Ok(c) => out.push(PolymorphicConcrete::Child(c)),
                    Err(e) => return Outcome::Err(e),
                }
            }
            Outcome::Ok(out)
        })
    }
}

/// Output of a concrete-table inheritance polymorphic query with two child types.
#[derive(Debug, Clone, PartialEq)]
pub enum PolymorphicConcrete2<C1: Model, C2: Model> {
    C1(C1),
    C2(C2),
}

/// A concrete-table polymorphic SELECT over two child tables.
///
/// Construct via `select!(Base).polymorphic_concrete2::<C1, C2>()`.
#[derive(Debug, Clone)]
pub struct PolymorphicConcreteSelect2<Base: Model, C1: Model, C2: Model> {
    select: Select<Base>,
    _marker: PhantomData<(C1, C2)>,
}

impl<Base: Model, C1: Model, C2: Model> PolymorphicConcreteSelect2<Base, C1, C2> {
    /// Add a WHERE condition (applied to every branch).
    #[must_use]
    pub fn filter(mut self, expr: Expr) -> Self {
        self.select = self.select.filter(expr);
        self
    }

    /// Add ORDER BY (applied to the union).
    #[must_use]
    pub fn order_by(mut self, order: OrderBy) -> Self {
        self.select = self.select.order_by(order);
        self
    }

    /// Set LIMIT (applied to the union).
    #[must_use]
    pub fn limit(mut self, n: u64) -> Self {
        self.select = self.select.limit(n);
        self
    }

    /// Set OFFSET (applied to the union).
    #[must_use]
    pub fn offset(mut self, n: u64) -> Self {
        self.select = self.select.offset(n);
        self
    }

    /// Build the SQL query and parameters.
    pub fn build_with_dialect(&self, dialect: Dialect) -> (String, Vec<Value>) {
        let (sql, params) = build_concrete_union_sql(
            &self.select,
            dialect,
            &[
                (C1::TABLE_NAME, C1::fields()),
                (C2::TABLE_NAME, C2::fields()),
            ],
        );
        let prefixes: &[&str] = &[C1::TABLE_NAME, C2::TABLE_NAME];
        tracing::trace!(
            target: "sqlmodel_query::polymorphic",
            dialect = ?dialect,
            sql = %sql,
            params = ?params,
            children = ?prefixes,
            "polymorphic select"
        );
        (sql, params)
    }

    /// Execute the query and hydrate `C1` or `C2` per row via the variant tag.
    #[tracing::instrument(level = "debug", skip(self, cx, conn))]
    pub async fn all<C: Connection>(
        self,
        cx: &Cx,
        conn: &C,
    ) -> Outcome<Vec<PolymorphicConcrete2<C1, C2>>, sqlmodel_core::Error> {
        if let Err(e) = concrete_hierarchy_check::<Base>(
            "polymorphic_concrete2",
            &[C1::TABLE_NAME, C2::TABLE_NAME],
        ) {
            return Outcome::Err(e);
        }
        if let Err(e) = concrete_child_check::<Base, C1>("polymorphic_concrete2") {
            return Outcome::Err(e);
        }
        if let Err(e) = concrete_child_check::<Base, C2>("polymorphic_concrete2") {
            return Outcome::Err(e);
        }

        let (sql, params) = self.build_with_dialect(conn.dialect());
        let rows = conn.query(cx, &sql, &params).await;
        rows.and_then(|rows| {
            let mut out = Vec::with_capacity(rows.len());
            for row in rows {
                let tag = match concrete_tag(&row) {
                    Ok(t) => t,
                    Err(e) => return Outcome::Err(e),
                };
                let hydrated = if tag == C1::TABLE_NAME {
                    C1::from_row(&row).map(PolymorphicConcrete2::C1)
                } else if tag == C2::TABLE_NAME {
                    C2::from_row(&row).map(PolymorphicConcrete2::C2)
                } else {
                    return Outcome::Err(sqlmodel_core::Error::Custom(format!(
                        "polymorphic_concrete2: unknown {} tag {tag:?} for base {}",
                        CONCRETE_TYPE_COLUMN,
                        Base::TABLE_NAME
                    )));
                };
                out.push(match hydrated {
                    Ok(v) => v,
                    Err(e) => return Outcome::Err(e),
                });
            }
            Outcome::Ok(out)
        })
    }
}

/// Output of a concrete-table inheritance polymorphic query with three child types.
#[derive(Debug, Clone, PartialEq)]
pub enum PolymorphicConcrete3<C1: Model, C2: Model, C3: Model> {
    C1(C1),
    C2(C2),
    C3(C3),
}

/// A concrete-table polymorphic SELECT over three child tables.
///
/// Construct via `select!(Base).polymorphic_concrete3::<C1, C2, C3>()`.
#[derive(Debug, Clone)]
pub struct PolymorphicConcreteSelect3<Base: Model, C1: Model, C2: Model, C3: Model> {
    select: Select<Base>,
    _marker: PhantomData<(C1, C2, C3)>,
}

impl<Base: Model, C1: Model, C2: Model, C3: Model> PolymorphicConcreteSelect3<Base, C1, C2, C3> {
    /// Add a WHERE condition (applied to every branch).
    #[must_use]
    pub fn filter(mut self, expr: Expr) -> Self {
        self.select = self.select.filter(expr);
        self
    }

    /// Add ORDER BY (applied to the union).
    #[must_use]
    pub fn order_by(mut self, order: OrderBy) -> Self {
        self.select = self.select.order_by(order);
        self
    }

    /// Set LIMIT (applied to the union).
    #[must_use]
    pub fn limit(mut self, n: u64) -> Self {
        self.select = self.select.limit(n);
        self
    }

    /// Set OFFSET (applied to the union).
    #[must_use]
    pub fn offset(mut self, n: u64) -> Self {
        self.select = self.select.offset(n);
        self
    }

    /// Build the SQL query and parameters.
    pub fn build_with_dialect(&self, dialect: Dialect) -> (String, Vec<Value>) {
        let (sql, params) = build_concrete_union_sql(
            &self.select,
            dialect,
            &[
                (C1::TABLE_NAME, C1::fields()),
                (C2::TABLE_NAME, C2::fields()),
                (C3::TABLE_NAME, C3::fields()),
            ],
        );
        let prefixes: &[&str] = &[C1::TABLE_NAME, C2::TABLE_NAME, C3::TABLE_NAME];
        tracing::trace!(
            target: "sqlmodel_query::polymorphic",
            dialect = ?dialect,
            sql = %sql,
            params = ?params,
            children = ?prefixes,
            "polymorphic select"
        );
        (sql, params)
    }

    /// Execute the query and hydrate `C1`/`C2`/`C3` per row via the variant tag.
    #[tracing::instrument(level = "debug", skip(self, cx, conn))]
    pub async fn all<C: Connection>(
        self,
        cx: &Cx,
        conn: &C,
    ) -> Outcome<Vec<PolymorphicConcrete3<C1, C2, C3>>, sqlmodel_core::Error> {
        if let Err(e) = concrete_hierarchy_check::<Base>(
            "polymorphic_concrete3",
            &[C1::TABLE_NAME, C2::TABLE_NAME, C3::TABLE_NAME],
        ) {
            return Outcome::Err(e);
        }
        if let Err(e) = concrete_child_check::<Base, C1>("polymorphic_concrete3") {
            return Outcome::Err(e);
        }
        if let Err(e) = concrete_child_check::<Base, C2>("polymorphic_concrete3") {
            return Outcome::Err(e);
        }
        if let Err(e) = concrete_child_check::<Base, C3>("polymorphic_concrete3") {
            return Outcome::Err(e);
        }

        let (sql, params) = self.build_with_dialect(conn.dialect());
        let rows = conn.query(cx, &sql, &params).await;
        rows.and_then(|rows| {
            let mut out = Vec::with_capacity(rows.len());
            for row in rows {
                let tag = match concrete_tag(&row) {
                    Ok(t) => t,
                    Err(e) => return Outcome::Err(e),
                };
                let hydrated = if tag == C1::TABLE_NAME {
                    C1::from_row(&row).map(PolymorphicConcrete3::C1)
                } else if tag == C2::TABLE_NAME {
                    C2::from_row(&row).map(PolymorphicConcrete3::C2)
                } else if tag == C3::TABLE_NAME {
                    C3::from_row(&row).map(PolymorphicConcrete3::C3)
                } else {
                    return Outcome::Err(sqlmodel_core::Error::Custom(format!(
                        "polymorphic_concrete3: unknown {} tag {tag:?} for base {}",
                        CONCRETE_TYPE_COLUMN,
                        Base::TABLE_NAME
                    )));
                };
                out.push(match hydrated {
                    Ok(v) => v,
                    Err(e) => return Outcome::Err(e),
                });
            }
            Outcome::Ok(out)
        })
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::JoinType;
    use sqlmodel_core::{
        Error, FieldInfo, InheritanceInfo, InheritanceStrategy, Result, Row, Value,
    };

    #[derive(Debug, Clone)]
    struct Hero;

    impl Model for Hero {
        const TABLE_NAME: &'static str = "heroes";
        const PRIMARY_KEY: &'static [&'static str] = &["id"];

        fn fields() -> &'static [FieldInfo] {
            &[]
        }

        fn to_row(&self) -> Vec<(&'static str, Value)> {
            Vec::new()
        }

        fn from_row(_row: &Row) -> Result<Self> {
            Err(Error::Custom("not used in tests".to_string()))
        }

        fn primary_key_value(&self) -> Vec<Value> {
            Vec::new()
        }

        fn is_new(&self) -> bool {
            true
        }
    }

    #[derive(Debug, Clone)]
    struct StiManager;

    impl Model for StiManager {
        // STI child shares the physical table.
        const TABLE_NAME: &'static str = "employees";
        const PRIMARY_KEY: &'static [&'static str] = &["id"];

        fn fields() -> &'static [FieldInfo] {
            &[]
        }

        fn to_row(&self) -> Vec<(&'static str, Value)> {
            Vec::new()
        }

        fn from_row(_row: &Row) -> Result<Self> {
            Err(Error::Custom("not used in tests".to_string()))
        }

        fn primary_key_value(&self) -> Vec<Value> {
            Vec::new()
        }

        fn is_new(&self) -> bool {
            true
        }

        fn inheritance() -> InheritanceInfo {
            InheritanceInfo {
                strategy: InheritanceStrategy::None,
                parent: Some("employees"),
                parent_fields_fn: None,
                discriminator_column: Some("type_"),
                discriminator_value: Some("manager"),
            }
        }
    }

    #[test]
    fn build_collects_params_across_joins_where_having() {
        let query = Select::<Hero>::new()
            .join(Join::inner(
                "teams",
                Expr::qualified("teams", "active").eq(true),
            ))
            .filter(Expr::col("age").gt(18))
            .group_by(&["team_id"])
            .having(Expr::col("count").gt(1));

        let (sql, params) = query.build();

        // With a JOIN the model's own columns are projected (`"heroes".*`), so a
        // same-named column on the joined table cannot shadow them.
        assert_eq!(
            sql,
            "SELECT \"heroes\".* FROM \"heroes\" INNER JOIN \"teams\" ON \"teams\".\"active\" = $1 WHERE \"age\" > $2 GROUP BY team_id HAVING \"count\" > $3"
        );
        assert_eq!(
            params,
            vec![Value::Bool(true), Value::Int(18), Value::Int(1)]
        );
    }

    #[test]
    fn test_select_all_columns() {
        let query = Select::<Hero>::new();
        let (sql, params) = query.build();

        assert_eq!(sql, "SELECT * FROM \"heroes\"");
        assert!(params.is_empty());
    }

    #[test]
    fn test_sti_child_select_adds_discriminator_filter() {
        let query = Select::<StiManager>::new();
        let (sql, params) = query.build();
        assert_eq!(
            sql,
            "SELECT * FROM \"employees\" WHERE \"employees\".\"type_\" = $1"
        );
        assert_eq!(params, vec![Value::Text("manager".to_string())]);
    }

    #[test]
    fn test_sti_child_select_ands_discriminator_with_user_filter() {
        let query = Select::<StiManager>::new().filter(Expr::col("active").eq(true));
        let (sql, params) = query.build();
        assert_eq!(
            sql,
            "SELECT * FROM \"employees\" WHERE \"active\" = $1 AND \"employees\".\"type_\" = $2"
        );
        assert_eq!(
            params,
            vec![Value::Bool(true), Value::Text("manager".to_string())]
        );
    }

    #[derive(Debug, Clone)]
    struct JoinedParent;

    impl Model for JoinedParent {
        const TABLE_NAME: &'static str = "persons";
        const PRIMARY_KEY: &'static [&'static str] = &["id"];

        fn fields() -> &'static [FieldInfo] {
            static FIELDS: &[FieldInfo] = &[
                FieldInfo::new("id", "id", sqlmodel_core::SqlType::BigInt).primary_key(true),
                FieldInfo::new("name", "name", sqlmodel_core::SqlType::Text),
            ];
            FIELDS
        }

        fn to_row(&self) -> Vec<(&'static str, Value)> {
            Vec::new()
        }

        fn from_row(_row: &Row) -> Result<Self> {
            Err(Error::Custom("not used in tests".to_string()))
        }

        fn primary_key_value(&self) -> Vec<Value> {
            Vec::new()
        }

        fn is_new(&self) -> bool {
            true
        }
    }

    #[derive(Debug, Clone)]
    struct JoinedChild;

    impl Model for JoinedChild {
        const TABLE_NAME: &'static str = "employees";
        const PRIMARY_KEY: &'static [&'static str] = &["id"];

        fn fields() -> &'static [FieldInfo] {
            static FIELDS: &[FieldInfo] = &[
                FieldInfo::new("id", "id", sqlmodel_core::SqlType::BigInt).primary_key(true),
                FieldInfo::new("dept", "department", sqlmodel_core::SqlType::Text),
            ];
            FIELDS
        }

        fn to_row(&self) -> Vec<(&'static str, Value)> {
            Vec::new()
        }

        fn from_row(_row: &Row) -> Result<Self> {
            Err(Error::Custom("not used in tests".to_string()))
        }

        fn primary_key_value(&self) -> Vec<Value> {
            Vec::new()
        }

        fn is_new(&self) -> bool {
            true
        }

        fn inheritance() -> InheritanceInfo {
            InheritanceInfo {
                strategy: InheritanceStrategy::Joined,
                parent: Some("persons"),
                parent_fields_fn: Some(<JoinedParent as Model>::fields),
                discriminator_column: None,
                discriminator_value: None,
            }
        }
    }

    #[test]
    fn test_joined_inheritance_child_select_projects_parent_and_joins() {
        let query = Select::<JoinedChild>::new();
        let (sql, params) = query.build();

        assert!(params.is_empty());
        assert!(sql.starts_with("SELECT "));
        assert!(sql.contains("\"employees\".\"id\" AS \"employees__id\""));
        assert!(sql.contains("\"employees\".\"department\" AS \"employees__department\""));
        assert!(sql.contains("\"persons\".\"id\" AS \"persons__id\""));
        assert!(sql.contains("\"persons\".\"name\" AS \"persons__name\""));
        assert!(sql.contains(
            "FROM \"employees\" INNER JOIN \"persons\" ON \"employees\".\"id\" = \"persons\".\"id\""
        ));
    }

    #[test]
    fn test_select_specific_columns() {
        let query = Select::<Hero>::new().columns(&["id", "name", "power"]);
        let (sql, params) = query.build();

        assert_eq!(sql, "SELECT id, name, power FROM \"heroes\"");
        assert!(params.is_empty());
    }

    #[test]
    fn test_select_distinct() {
        let query = Select::<Hero>::new().columns(&["team_id"]).distinct();
        let (sql, params) = query.build();

        assert_eq!(sql, "SELECT DISTINCT team_id FROM \"heroes\"");
        assert!(params.is_empty());
    }

    #[test]
    fn test_select_with_simple_filter() {
        let query = Select::<Hero>::new().filter(Expr::col("active").eq(true));
        let (sql, params) = query.build();

        assert_eq!(sql, "SELECT * FROM \"heroes\" WHERE \"active\" = $1");
        assert_eq!(params, vec![Value::Bool(true)]);
    }

    #[test]
    fn test_select_with_multiple_and_filters() {
        let query = Select::<Hero>::new()
            .filter(Expr::col("active").eq(true))
            .filter(Expr::col("age").gt(18));
        let (sql, params) = query.build();

        assert_eq!(
            sql,
            "SELECT * FROM \"heroes\" WHERE \"active\" = $1 AND \"age\" > $2"
        );
        assert_eq!(params, vec![Value::Bool(true), Value::Int(18)]);
    }

    #[test]
    fn test_select_with_or_filter() {
        let query = Select::<Hero>::new()
            .filter(Expr::col("role").eq("warrior"))
            .or_filter(Expr::col("role").eq("mage"));
        let (sql, params) = query.build();

        assert_eq!(
            sql,
            "SELECT * FROM \"heroes\" WHERE \"role\" = $1 OR \"role\" = $2"
        );
        assert_eq!(
            params,
            vec![
                Value::Text("warrior".to_string()),
                Value::Text("mage".to_string())
            ]
        );
    }

    #[test]
    fn test_select_with_order_by_asc() {
        let query = Select::<Hero>::new().order_by(OrderBy::asc(Expr::col("name")));
        let (sql, params) = query.build();

        assert_eq!(sql, "SELECT * FROM \"heroes\" ORDER BY \"name\" ASC");
        assert!(params.is_empty());
    }

    #[test]
    fn test_select_with_order_by_desc() {
        let query = Select::<Hero>::new().order_by(OrderBy::desc(Expr::col("created_at")));
        let (sql, params) = query.build();

        assert_eq!(sql, "SELECT * FROM \"heroes\" ORDER BY \"created_at\" DESC");
        assert!(params.is_empty());
    }

    #[test]
    fn test_select_with_multiple_order_by() {
        let query = Select::<Hero>::new()
            .order_by(OrderBy::asc(Expr::col("team_id")))
            .order_by(OrderBy::asc(Expr::col("name")));
        let (sql, params) = query.build();

        assert_eq!(
            sql,
            "SELECT * FROM \"heroes\" ORDER BY \"team_id\" ASC, \"name\" ASC"
        );
        assert!(params.is_empty());
    }

    #[test]
    fn test_select_with_limit() {
        let query = Select::<Hero>::new().limit(10);
        let (sql, params) = query.build();

        assert_eq!(sql, "SELECT * FROM \"heroes\" LIMIT 10");
        assert!(params.is_empty());
    }

    #[test]
    fn test_select_with_offset() {
        let query = Select::<Hero>::new().offset(20);
        let (sql, params) = query.build();

        assert_eq!(sql, "SELECT * FROM \"heroes\" OFFSET 20");
        assert!(params.is_empty());
    }

    #[test]
    fn test_select_with_limit_and_offset() {
        let query = Select::<Hero>::new().limit(10).offset(20);
        let (sql, params) = query.build();

        assert_eq!(sql, "SELECT * FROM \"heroes\" LIMIT 10 OFFSET 20");
        assert!(params.is_empty());
    }

    #[test]
    fn test_select_with_group_by() {
        let query = Select::<Hero>::new()
            .columns(&["team_id", "COUNT(*) as count"])
            .group_by(&["team_id"]);
        let (sql, params) = query.build();

        assert_eq!(
            sql,
            "SELECT team_id, COUNT(*) as count FROM \"heroes\" GROUP BY team_id"
        );
        assert!(params.is_empty());
    }

    #[test]
    fn test_select_with_multiple_group_by() {
        let query = Select::<Hero>::new()
            .columns(&["team_id", "role", "COUNT(*) as count"])
            .group_by(&["team_id", "role"]);
        let (sql, params) = query.build();

        assert_eq!(
            sql,
            "SELECT team_id, role, COUNT(*) as count FROM \"heroes\" GROUP BY team_id, role"
        );
        assert!(params.is_empty());
    }

    #[test]
    fn test_select_with_for_update() {
        let query = Select::<Hero>::new()
            .filter(Expr::col("id").eq(1))
            .for_update();
        let (sql, params) = query.build();

        assert_eq!(sql, "SELECT * FROM \"heroes\" WHERE \"id\" = $1 FOR UPDATE");
        assert_eq!(params, vec![Value::Int(1)]);
    }

    #[test]
    fn test_select_inner_join() {
        let query = Select::<Hero>::new().join(Join::inner(
            "teams",
            Expr::qualified("heroes", "team_id").eq(Expr::qualified("teams", "id")),
        ));
        let (sql, _) = query.build();

        assert!(sql.contains("INNER JOIN \"teams\" ON"));
    }

    #[test]
    fn test_select_left_join() {
        let query = Select::<Hero>::new().join(Join::left(
            "teams",
            Expr::qualified("heroes", "team_id").eq(Expr::qualified("teams", "id")),
        ));
        let (sql, _) = query.build();

        assert!(sql.contains("LEFT JOIN \"teams\" ON"));
    }

    #[test]
    fn test_select_right_join() {
        let query = Select::<Hero>::new().join(Join::right(
            "teams",
            Expr::qualified("heroes", "team_id").eq(Expr::qualified("teams", "id")),
        ));
        let (sql, _) = query.build();

        assert!(sql.contains("RIGHT JOIN \"teams\" ON"));
    }

    #[test]
    fn test_select_multiple_joins() {
        let query = Select::<Hero>::new()
            .join(Join::inner(
                "teams",
                Expr::qualified("heroes", "team_id").eq(Expr::qualified("teams", "id")),
            ))
            .join(Join::left(
                "powers",
                Expr::qualified("heroes", "id").eq(Expr::qualified("powers", "hero_id")),
            ));
        let (sql, _) = query.build();

        assert!(sql.contains("INNER JOIN \"teams\" ON"));
        assert!(sql.contains("LEFT JOIN \"powers\" ON"));
    }

    #[test]
    fn test_select_complex_query() {
        let query = Select::<Hero>::new()
            .columns(&["heroes.id", "heroes.name", "teams.name as team_name"])
            .distinct()
            .join(Join::inner(
                "teams",
                Expr::qualified("heroes", "team_id").eq(Expr::qualified("teams", "id")),
            ))
            .filter(Expr::col("active").eq(true))
            .filter(Expr::col("level").gt(10))
            .group_by(&["heroes.id", "heroes.name", "teams.name"])
            .having(Expr::col("score").gt(100))
            .order_by(OrderBy::desc(Expr::col("level")))
            .limit(50)
            .offset(0);
        let (sql, params) = query.build();

        assert!(sql.starts_with(
            "SELECT DISTINCT heroes.id, heroes.name, teams.name as team_name FROM \"heroes\""
        ));
        assert!(sql.contains("INNER JOIN \"teams\" ON"));
        assert!(sql.contains("WHERE"));
        assert!(sql.contains("GROUP BY"));
        assert!(sql.contains("HAVING"));
        assert!(sql.contains("ORDER BY"));
        assert!(sql.contains("LIMIT 50"));
        assert!(sql.contains("OFFSET 0"));

        // Params: true (active), 10 (level), 100 (score)
        // Note: join condition uses column comparison, not value param
        assert_eq!(params.len(), 3);
    }

    #[test]
    fn test_select_default() {
        let query = Select::<Hero>::default();
        let (sql, _) = query.build();
        assert_eq!(sql, "SELECT * FROM \"heroes\"");
    }

    #[test]
    fn test_select_clone() {
        let query = Select::<Hero>::new()
            .filter(Expr::col("id").eq(1))
            .limit(10);
        let cloned = query.clone();

        let (sql1, params1) = query.build();
        let (sql2, params2) = cloned.build();

        assert_eq!(sql1, sql2);
        assert_eq!(params1, params2);
    }

    // ========================================================================
    // Eager Loading Tests
    // ========================================================================

    use sqlmodel_core::RelationshipInfo;

    /// A test team model for eager loading column projection.
    #[derive(Debug, Clone)]
    struct EagerTeam;

    impl Model for EagerTeam {
        const TABLE_NAME: &'static str = "teams";
        const PRIMARY_KEY: &'static [&'static str] = &["id"];

        fn fields() -> &'static [FieldInfo] {
            static FIELDS: &[FieldInfo] = &[
                FieldInfo::new("id", "id", sqlmodel_core::SqlType::BigInt),
                FieldInfo::new("name", "name", sqlmodel_core::SqlType::Text),
            ];
            FIELDS
        }

        fn to_row(&self) -> Vec<(&'static str, Value)> {
            Vec::new()
        }

        fn from_row(_row: &Row) -> Result<Self> {
            Err(Error::Custom("not used in tests".to_string()))
        }

        fn primary_key_value(&self) -> Vec<Value> {
            Vec::new()
        }

        fn is_new(&self) -> bool {
            true
        }
    }

    /// A test hero model with relationships defined.
    #[derive(Debug, Clone)]
    struct EagerHero;

    impl Model for EagerHero {
        const TABLE_NAME: &'static str = "heroes";
        const PRIMARY_KEY: &'static [&'static str] = &["id"];
        const RELATIONSHIPS: &'static [RelationshipInfo] =
            &[
                RelationshipInfo::new("team", "teams", RelationshipKind::ManyToOne)
                    .related_fields(EagerTeam::fields)
                    .local_key("team_id"),
            ];

        fn fields() -> &'static [FieldInfo] {
            static FIELDS: &[FieldInfo] = &[
                FieldInfo::new("id", "id", sqlmodel_core::SqlType::BigInt),
                FieldInfo::new("name", "name", sqlmodel_core::SqlType::Text),
                FieldInfo::new("team_id", "team_id", sqlmodel_core::SqlType::BigInt),
            ];
            FIELDS
        }

        fn to_row(&self) -> Vec<(&'static str, Value)> {
            Vec::new()
        }

        fn from_row(_row: &Row) -> Result<Self> {
            Err(Error::Custom("not used in tests".to_string()))
        }

        fn primary_key_value(&self) -> Vec<Value> {
            Vec::new()
        }

        fn is_new(&self) -> bool {
            true
        }
    }

    #[test]
    fn test_select_with_eager_loader() {
        let loader = EagerLoader::<EagerHero>::new().include("team");
        let query = Select::<EagerHero>::new().eager(loader);

        // Verify eager_loader is set
        assert!(query.eager_loader.is_some());
        assert!(query.eager_loader.as_ref().unwrap().has_includes());
    }

    #[test]
    fn test_select_eager_generates_join() {
        let loader = EagerLoader::<EagerHero>::new().include("team");
        let query = Select::<EagerHero>::new().eager(loader);

        let (sql, params, join_info) = query.build_eager_with_dialect(Dialect::default());

        // Should have LEFT JOIN for team relationship
        assert!(sql.contains("LEFT JOIN \"teams\""));
        assert!(sql.contains("\"heroes\".\"team_id\" = \"teams\".\"id\""));

        // Should have aliased columns for parent table
        assert!(sql.contains("\"heroes\".\"id\" AS \"heroes__id\""));
        assert!(sql.contains("\"heroes\".\"name\" AS \"heroes__name\""));
        assert!(sql.contains("\"heroes\".\"team_id\" AS \"heroes__team_id\""));

        // Should have aliased columns for related table (so subset_by_prefix works)
        assert!(sql.contains("\"teams\".\"id\" AS \"teams__id\""));
        assert!(sql.contains("\"teams\".\"name\" AS \"teams__name\""));

        // Should have join info
        assert_eq!(join_info.len(), 1);
        assert!(params.is_empty());
    }

    #[test]
    fn test_select_eager_with_filter() {
        let loader = EagerLoader::<EagerHero>::new().include("team");
        let query = Select::<EagerHero>::new()
            .eager(loader)
            .filter(Expr::col("active").eq(true));

        let (sql, params, _) = query.build_eager_with_dialect(Dialect::default());

        assert!(sql.contains("LEFT JOIN \"teams\""));
        assert!(sql.contains("WHERE"));
        assert!(sql.contains("\"active\" = $1"));
        assert_eq!(params, vec![Value::Bool(true)]);
    }

    #[test]
    fn test_select_eager_with_order_and_limit() {
        let loader = EagerLoader::<EagerHero>::new().include("team");
        let query = Select::<EagerHero>::new()
            .eager(loader)
            .order_by(OrderBy::asc(Expr::col("name")))
            .limit(10)
            .offset(5);

        let (sql, _, _) = query.build_eager_with_dialect(Dialect::default());

        assert!(sql.contains("LEFT JOIN \"teams\""));
        assert!(sql.contains("ORDER BY"));
        assert!(sql.contains("LIMIT 10"));
        assert!(sql.contains("OFFSET 5"));
    }

    #[test]
    fn test_select_eager_no_includes_fallback() {
        // Eager loader with no includes
        let loader = EagerLoader::<EagerHero>::new();
        let query = Select::<EagerHero>::new().eager(loader);

        // all_eager should fall back to regular all() when no includes
        // We can't test async execution here, but we can verify the state
        assert!(query.eager_loader.is_some());
        assert!(!query.eager_loader.as_ref().unwrap().has_includes());
    }

    #[test]
    fn test_select_eager_distinct() {
        let loader = EagerLoader::<EagerHero>::new().include("team");
        let query = Select::<EagerHero>::new().eager(loader).distinct();

        let (sql, _, _) = query.build_eager_with_dialect(Dialect::default());

        assert!(sql.starts_with("SELECT DISTINCT"));
    }

    #[test]
    fn in_query_embeds_a_typed_subquery_with_renumbered_params() {
        let red_teams = Select::<EagerTeam>::new()
            .columns(&["id"])
            .filter(Expr::col("name").eq("red"))
            .into_query();
        let (sql, params) = Select::<EagerHero>::new()
            .filter(Expr::col("name").ne("ghost"))
            .filter(Expr::col("team_id").in_query(red_teams))
            .build_with_dialect(Dialect::Postgres);
        assert_eq!(
            sql,
            "SELECT * FROM \"heroes\" WHERE \"name\" <> $1 AND \"team_id\" IN \
             (SELECT id FROM \"teams\" WHERE \"name\" = $2)"
        );
        assert_eq!(params, vec![Value::from("ghost"), Value::from("red")]);

        let (mysql, _) = Select::<EagerHero>::new()
            .filter(
                Expr::col("team_id").not_in_query(
                    Select::<EagerTeam>::new()
                        .columns(&["id"])
                        .filter(Expr::col("name").eq("red"))
                        .into_query(),
                ),
            )
            .build_with_dialect(Dialect::Mysql);
        assert_eq!(
            mysql,
            "SELECT * FROM `heroes` WHERE `team_id` NOT IN (SELECT id FROM `teams` WHERE `name` = ?)"
        );
    }

    // ==================== EXISTS Tests ====================

    #[test]
    fn test_select_into_exists() {
        // Convert a SELECT query into an EXISTS expression
        let exists_expr = Select::<Hero>::new()
            .filter(Expr::raw("orders.customer_id = customers.id"))
            .into_exists();

        let mut params = Vec::new();
        let sql = exists_expr.build(&mut params, 0);

        // Should generate EXISTS (SELECT 1 FROM \"heroes\" WHERE ...)
        assert_eq!(
            sql,
            "EXISTS (SELECT 1 FROM \"heroes\" WHERE orders.customer_id = customers.id)"
        );
    }

    #[test]
    fn test_select_into_not_exists() {
        // Convert a SELECT query into a NOT EXISTS expression
        let not_exists_expr = Select::<Hero>::new()
            .filter(Expr::raw("orders.customer_id = customers.id"))
            .into_not_exists();

        let mut params = Vec::new();
        let sql = not_exists_expr.build(&mut params, 0);

        assert_eq!(
            sql,
            "NOT EXISTS (SELECT 1 FROM \"heroes\" WHERE orders.customer_id = customers.id)"
        );
    }

    #[test]
    fn test_select_into_exists_with_params() {
        // EXISTS subquery with bound parameters
        let exists_expr = Select::<Hero>::new()
            .filter(Expr::col("status").eq("active"))
            .into_exists();

        let mut params = Vec::new();
        let sql = exists_expr.build(&mut params, 0);

        assert_eq!(
            sql,
            "EXISTS (SELECT 1 FROM \"heroes\" WHERE \"status\" = $1)"
        );
        assert_eq!(params.len(), 1);
        assert_eq!(params[0], Value::Text("active".to_string()));
    }

    #[test]
    fn test_select_into_exists_propagates_dialect_mysql() {
        let exists_expr = Select::<Hero>::new()
            .filter(Expr::col("status").eq("active"))
            .into_exists();

        let mut params = Vec::new();
        let sql = exists_expr.build_with_dialect(Dialect::Mysql, &mut params, 0);

        assert_eq!(sql, "EXISTS (SELECT 1 FROM `heroes` WHERE `status` = ?)");
        assert_eq!(params, vec![Value::Text("active".to_string())]);
    }

    #[test]
    fn test_select_into_exists_with_join() {
        // EXISTS subquery with JOIN
        let exists_expr = Select::<Hero>::new()
            .join(Join::inner(
                "teams",
                Expr::qualified("heroes", "team_id").eq(Expr::qualified("teams", "id")),
            ))
            .filter(Expr::col("active").eq(true))
            .into_exists();

        let mut params = Vec::new();
        let sql = exists_expr.build(&mut params, 0);

        assert!(sql.starts_with("EXISTS (SELECT 1 FROM \"heroes\""));
        assert!(sql.contains("INNER JOIN \"teams\" ON"));
        assert!(sql.contains("WHERE"));
    }

    #[test]
    fn test_select_into_exists_omits_order_by_limit() {
        // ORDER BY, LIMIT, OFFSET should be omitted from EXISTS subquery
        // as they have no effect and add unnecessary overhead
        let exists_expr = Select::<Hero>::new()
            .filter(Expr::col("active").eq(true))
            .order_by(OrderBy::asc(Expr::col("name")))
            .limit(10)
            .offset(5)
            .into_exists();

        let mut params = Vec::new();
        let sql = exists_expr.build(&mut params, 0);

        // Should NOT contain ORDER BY, LIMIT, OFFSET
        assert!(!sql.contains("ORDER BY"));
        assert!(!sql.contains("LIMIT"));
        assert!(!sql.contains("OFFSET"));
        assert_eq!(
            sql,
            "EXISTS (SELECT 1 FROM \"heroes\" WHERE \"active\" = $1)"
        );
    }

    #[test]
    fn test_exists_in_outer_query() {
        // Use EXISTS expression in a WHERE clause of another query
        let has_heroes = Select::<Hero>::new()
            .filter(Expr::raw("heroes.team_id = teams.id"))
            .into_exists();

        let query = Select::<EagerTeam>::new().filter(Expr::col("active").eq(true).and(has_heroes));
        let (sql, params) = query.build_with_dialect(Dialect::default());

        assert_eq!(
            sql,
            "SELECT * FROM \"teams\" WHERE \"active\" = $1 AND EXISTS (SELECT 1 FROM \"heroes\" WHERE heroes.team_id = teams.id)"
        );
        assert_eq!(params, vec![Value::Bool(true)]);
    }

    #[test]
    fn test_lateral_join_propagates_dialect_sqlite() {
        let lateral = Select::<Hero>::new()
            .filter(Expr::col("status").eq("active"))
            .into_lateral_join("recent", JoinType::Left, Expr::raw("TRUE"));

        let query = Select::<Hero>::new()
            .filter(Expr::col("active").eq(true))
            .join(lateral);

        let (sql, params) = query.build_with_dialect(Dialect::Sqlite);

        assert!(sql.contains(
            "LEFT JOIN LATERAL (SELECT * FROM \"heroes\" WHERE \"status\" = ?1) AS recent ON TRUE"
        ));
        assert!(sql.contains("WHERE \"active\" = ?2"));
        assert_eq!(params.len(), 2);
        assert_eq!(params[0], Value::Text("active".to_string()));
        assert_eq!(params[1], Value::Bool(true));
    }

    // ---- Concrete-table inheritance polymorphic selects ----

    #[derive(Debug, Clone)]
    struct ConcreteBase;

    impl Model for ConcreteBase {
        const TABLE_NAME: &'static str = "persons";
        const PRIMARY_KEY: &'static [&'static str] = &["id"];

        fn fields() -> &'static [FieldInfo] {
            &[]
        }

        fn to_row(&self) -> Vec<(&'static str, Value)> {
            Vec::new()
        }

        fn from_row(_row: &Row) -> Result<Self> {
            Err(Error::Custom("not used in tests".to_string()))
        }

        fn primary_key_value(&self) -> Vec<Value> {
            Vec::new()
        }

        fn is_new(&self) -> bool {
            true
        }

        fn inheritance() -> InheritanceInfo {
            InheritanceInfo {
                strategy: InheritanceStrategy::Concrete,
                parent: None,
                parent_fields_fn: None,
                discriminator_column: None,
                discriminator_value: None,
            }
        }
    }

    /// Concrete child owning columns id/name/office in table `managers`.
    #[derive(Debug, Clone)]
    struct ConcreteManager;

    impl Model for ConcreteManager {
        const TABLE_NAME: &'static str = "managers";
        const PRIMARY_KEY: &'static [&'static str] = &["id"];

        fn fields() -> &'static [FieldInfo] {
            static FIELDS: &[FieldInfo] = &[
                FieldInfo::new("id", "id", sqlmodel_core::SqlType::BigInt).primary_key(true),
                FieldInfo::new("name", "name", sqlmodel_core::SqlType::Text),
                FieldInfo::new("office", "office", sqlmodel_core::SqlType::Text),
            ];
            FIELDS
        }

        fn to_row(&self) -> Vec<(&'static str, Value)> {
            Vec::new()
        }

        fn from_row(_row: &Row) -> Result<Self> {
            Err(Error::Custom("not used in tests".to_string()))
        }

        fn primary_key_value(&self) -> Vec<Value> {
            Vec::new()
        }

        fn is_new(&self) -> bool {
            true
        }

        fn inheritance() -> InheritanceInfo {
            InheritanceInfo {
                strategy: InheritanceStrategy::Concrete,
                parent: Some("persons"),
                parent_fields_fn: None,
                discriminator_column: None,
                discriminator_value: None,
            }
        }
    }

    /// Concrete child owning columns id/name/badge in table `engineers`; it
    /// has no `office` column, so its union branch must fill that column with
    /// a typed NULL.
    #[derive(Debug, Clone)]
    struct ConcreteEngineer;

    impl Model for ConcreteEngineer {
        const TABLE_NAME: &'static str = "engineers";
        const PRIMARY_KEY: &'static [&'static str] = &["id"];

        fn fields() -> &'static [FieldInfo] {
            static FIELDS: &[FieldInfo] = &[
                FieldInfo::new("id", "id", sqlmodel_core::SqlType::BigInt).primary_key(true),
                FieldInfo::new("name", "name", sqlmodel_core::SqlType::Text),
                FieldInfo::new("badge", "badge", sqlmodel_core::SqlType::BigInt),
            ];
            FIELDS
        }

        fn to_row(&self) -> Vec<(&'static str, Value)> {
            Vec::new()
        }

        fn from_row(_row: &Row) -> Result<Self> {
            Err(Error::Custom("not used in tests".to_string()))
        }

        fn primary_key_value(&self) -> Vec<Value> {
            Vec::new()
        }

        fn is_new(&self) -> bool {
            true
        }

        fn inheritance() -> InheritanceInfo {
            InheritanceInfo {
                strategy: InheritanceStrategy::Concrete,
                parent: Some("persons"),
                parent_fields_fn: None,
                discriminator_column: None,
                discriminator_value: None,
            }
        }
    }

    #[test]
    fn test_concrete_polymorphic2_union_fillers_and_dense_placeholders_sqlite() {
        let query = Select::<ConcreteBase>::new()
            .polymorphic_concrete2::<ConcreteManager, ConcreteEngineer>()
            .filter(Expr::col("name").eq("ada"))
            .order_by(OrderBy::asc(Expr::col("id")))
            .limit(10);
        let (sql, params) = query.build_with_dialect(Dialect::Sqlite);

        // Branch 1 owns `office`, branch 2 owns `badge`; the other branch of
        // each fills the column with a typed NULL. Placeholders stay dense
        // across branches ($-free dialect: ?1, ?2), and ORDER BY / LIMIT scope
        // the whole union.
        assert_eq!(
            sql,
            "SELECT \"managers\".\"id\" AS \"id\", \"managers\".\"name\" AS \"name\", \
             \"managers\".\"office\" AS \"office\", CAST(NULL AS BIGINT) AS \"badge\", \
             'managers' AS \"__type\" FROM \"managers\" WHERE \"name\" = ?1 \
             UNION ALL SELECT \"engineers\".\"id\" AS \"id\", \"engineers\".\"name\" AS \"name\", \
             CAST(NULL AS TEXT) AS \"office\", \"engineers\".\"badge\" AS \"badge\", \
             'engineers' AS \"__type\" FROM \"engineers\" WHERE \"name\" = ?2 \
             ORDER BY \"id\" ASC LIMIT 10"
        );
        assert_eq!(
            params,
            vec![
                Value::Text("ada".to_string()),
                Value::Text("ada".to_string())
            ]
        );
    }

    #[test]
    fn test_concrete_polymorphic2_postgres_placeholders_and_casts() {
        let query = Select::<ConcreteBase>::new()
            .polymorphic_concrete2::<ConcreteManager, ConcreteEngineer>()
            .filter(Expr::col("name").eq("grace"));
        let (sql, params) = query.build_with_dialect(Dialect::Postgres);

        assert!(sql.contains("CAST(NULL AS BIGINT) AS \"badge\""));
        assert!(sql.contains("CAST(NULL AS TEXT) AS \"office\""));
        assert!(sql.contains("WHERE \"name\" = $1"));
        assert!(sql.contains("WHERE \"name\" = $2"));
        assert_eq!(params.len(), 2);
    }

    #[test]
    fn test_concrete_polymorphic2_mysql_cast_targets() {
        let query = Select::<ConcreteBase>::new()
            .polymorphic_concrete2::<ConcreteManager, ConcreteEngineer>();
        let (sql, params) = query.build_with_dialect(Dialect::Mysql);

        // MySQL's CAST accepts neither BIGINT nor TEXT; integers fill with
        // SIGNED and text with CHAR.
        assert!(sql.contains("CAST(NULL AS SIGNED) AS `badge`"));
        assert!(sql.contains("CAST(NULL AS CHAR) AS `office`"));
        assert!(sql.contains("'managers' AS `__type`"));
        assert!(sql.contains("'engineers' AS `__type`"));
        assert!(params.is_empty());
    }

    #[test]
    fn test_concrete_polymorphic_single_child_builds_one_branch() {
        let query = Select::<ConcreteBase>::new().polymorphic_concrete::<ConcreteManager>();
        let (sql, params) = query.build_with_dialect(Dialect::Sqlite);

        assert!(!sql.contains("UNION ALL"));
        assert!(sql.contains("'managers' AS \"__type\""));
        assert_eq!(params.len(), 0);
    }

    #[test]
    fn test_concrete_polymorphic3_three_branches_with_renumbering() {
        let query = Select::<ConcreteBase>::new()
            .polymorphic_concrete3::<ConcreteManager, ConcreteEngineer, ConcreteManager>()
            .filter(Expr::col("name").eq("x"));
        let (sql, _) = query.build_with_dialect(Dialect::Sqlite);

        assert_eq!(sql.matches(" UNION ALL ").count(), 2);
        // Filters are applied per branch: three branches, three placeholders.
        assert_eq!(sql.matches('?').count(), 3);
    }

    #[test]
    fn test_joined_row_guard_reports_ambiguity_and_allows_base() {
        // All child prefixes NULL -> base row, guard passes.
        let base_row = Row::new(
            vec!["persons__id".to_string(), "students__id".to_string()],
            vec![Value::Int(1), Value::Null],
        );
        joined_row_guard(&base_row, &["students"], "polymorphic_joined").expect("base row passes");

        // Exactly one non-NULL prefix -> guard passes.
        let student_row = Row::new(
            vec!["persons__id".to_string(), "students__id".to_string()],
            vec![Value::Int(1), Value::Int(1)],
        );
        joined_row_guard(&student_row, &["students"], "polymorphic_joined")
            .expect("single match passes");

        // Two non-NULL prefixes -> the ambiguity error names both prefixes.
        let ambiguous = Row::new(
            vec!["students__id".to_string(), "teachers__id".to_string()],
            vec![Value::Int(1), Value::Int(1)],
        );
        let error = joined_row_guard(&ambiguous, &["students", "teachers"], "polymorphic_joined2")
            .expect_err("ambiguous row errors");
        let Error::Custom(message) = error else {
            panic!("expected Custom error, got {error:?}");
        };
        assert!(
            message.contains("multiple child prefixes are non-NULL: students, teachers"),
            "message names both prefixes: {message}"
        );
    }

    // Eight joined children, macro-generated so the N=8 entry point has real
    // models to join without eight hand-written copies.
    macro_rules! define_test_joined_child {
        ($name:ident, $table:literal) => {
            #[derive(Debug, Clone)]
            struct $name;

            impl Model for $name {
                const TABLE_NAME: &'static str = $table;
                const PRIMARY_KEY: &'static [&'static str] = &["id"];

                fn fields() -> &'static [FieldInfo] {
                    static FIELDS: &[FieldInfo] =
                        &[FieldInfo::new("id", "id", sqlmodel_core::SqlType::BigInt)
                            .primary_key(true)];
                    FIELDS
                }

                fn to_row(&self) -> Vec<(&'static str, Value)> {
                    Vec::new()
                }

                fn from_row(_row: &Row) -> Result<Self> {
                    Err(Error::Custom("not used in tests".to_string()))
                }

                fn primary_key_value(&self) -> Vec<Value> {
                    Vec::new()
                }

                fn is_new(&self) -> bool {
                    true
                }

                fn inheritance() -> InheritanceInfo {
                    InheritanceInfo {
                        strategy: InheritanceStrategy::Joined,
                        parent: Some(JoinedParent::TABLE_NAME),
                        parent_fields_fn: Some(<JoinedParent as Model>::fields),
                        discriminator_column: None,
                        discriminator_value: None,
                    }
                }
            }
        };
    }

    define_test_joined_child!(JoinedChildA, "jc_a");
    define_test_joined_child!(JoinedChildB, "jc_b");
    define_test_joined_child!(JoinedChildC, "jc_c");
    define_test_joined_child!(JoinedChildD, "jc_d");
    define_test_joined_child!(JoinedChildE, "jc_e");
    define_test_joined_child!(JoinedChildF, "jc_f");
    define_test_joined_child!(JoinedChildG, "jc_g");
    define_test_joined_child!(JoinedChildH, "jc_h");

    #[test]
    fn test_polymorphic_joined8_left_joins_every_child() {
        let query = Select::<JoinedParent>::new().polymorphic_joined8::<
            JoinedChildA,
            JoinedChildB,
            JoinedChildC,
            JoinedChildD,
            JoinedChildE,
            JoinedChildF,
            JoinedChildG,
            JoinedChildH,
        >();
        let (sql, params) = query.build_with_dialect(Dialect::Sqlite);

        for table in [
            "jc_a", "jc_b", "jc_c", "jc_d", "jc_e", "jc_f", "jc_g", "jc_h",
        ] {
            assert!(
                sql.contains(&format!("LEFT JOIN \"{table}\"")),
                "missing join for {table}: {sql}"
            );
            assert!(
                sql.contains(&format!("\"{table}\".\"id\" AS \"{table}__id\"")),
                "missing projection for {table}: {sql}"
            );
        }
        assert_eq!(sql.matches("LEFT JOIN").count(), 8);
        assert!(params.is_empty());
    }
}