qail-core 3.0.0

//! SELECT SQL generation.

use crate::ast::*;
use crate::transpiler::conditions::{
    ConditionToSql, read_only_subquery_sql, resolve_known_col_syntax, slice_bounds_sql,
    subscript_sql,
};
use crate::transpiler::dialect::Dialect;
use crate::transpiler::identifier::{
    render_table_reference, table_reference_base, table_reference_sql_qualifier,
};
use crate::transpiler::traits::SqlGenerator;

/// Generate SELECT SQL from a QAIL command, including CTEs, joins, filtering, grouping, and ordering.
pub fn build_select(cmd: &Qail, dialect: Dialect) -> String {
    build_select_with_columns(cmd, dialect, &cmd.columns)
}

pub(crate) fn build_select_with_columns(cmd: &Qail, dialect: Dialect, columns: &[Expr]) -> String {
    build_select_inner(cmd, dialect, columns, true)
}

pub(super) fn build_select_without_cte_prefix(cmd: &Qail, dialect: Dialect) -> String {
    build_select_inner(cmd, dialect, &cmd.columns, false)
}

fn build_select_inner(
    cmd: &Qail,
    dialect: Dialect,
    columns: &[Expr],
    include_ctes: bool,
) -> String {
    if !cmd.conflict_update_scope.is_empty() {
        return crate::transpiler::INVALID_INSERT_SCOPE_SQL.to_string();
    }
    let generator = dialect.generator();

    // CTE prefix: WITH cte1 AS (...), cte2 AS (...)
    let cte_prefix = if include_ctes && !cmd.ctes.is_empty() {
        let has_recursive = cmd.ctes.iter().any(|c| c.recursive);
        let cte_parts: Vec<String> = cmd
            .ctes
            .iter()
            .map(|cte| super::cte::build_single_cte(cte, dialect))
            .collect();
        if has_recursive {
            format!("WITH RECURSIVE {} ", cte_parts.join(", "))
        } else {
            format!("WITH {} ", cte_parts.join(", "))
        }
    } else {
        String::new()
    };

    let mut sql = if !cmd.distinct_on.is_empty() {
        let exprs: Vec<String> = cmd
            .distinct_on
            .iter()
            .map(|e| render_expr_for_orderby(e, generator.as_ref(), cmd))
            .collect();
        format!("{}SELECT DISTINCT ON ({}) ", cte_prefix, exprs.join(", "))
    } else if cmd.distinct {
        format!("{}SELECT DISTINCT ", cte_prefix)
    } else {
        format!("{}SELECT ", cte_prefix)
    };

    if columns.is_empty() {
        sql.push('*');
    } else {
        let cols: Vec<String> = columns
            .iter()
            .map(|c| {
                match c {
                    Expr::Star => "*".to_string(),
                    Expr::Named(name) => render_named_reference(name, generator.as_ref(), cmd),
                    Expr::Aliased { name, alias } => format!(
                        "{} AS {}",
                        render_named_reference(name, generator.as_ref(), cmd),
                        generator.quote_identifier(alias)
                    ),
                    Expr::Case {
                        when_clauses,
                        else_value,
                        alias,
                    } => {
                        let mut case_sql = String::from("CASE");
                        for (cond, val) in when_clauses {
                            case_sql.push_str(&format!(
                                " WHEN {} THEN {}",
                                cond.to_sql(generator.as_ref(), Some(cmd)),
                                render_expr_for_orderby(val, generator.as_ref(), cmd)
                            ));
                        }
                        if let Some(e) = else_value {
                            case_sql.push_str(&format!(
                                " ELSE {}",
                                render_expr_for_orderby(e, generator.as_ref(), cmd)
                            ));
                        }
                        case_sql.push_str(" END");
                        if let Some(a) = alias {
                            format!("{} AS {}", case_sql, generator.quote_identifier(a))
                        } else {
                            case_sql
                        }
                    }
                    Expr::JsonAccess {
                        column,
                        path_segments,
                        alias,
                    } => {
                        let expr = render_json_access(column, path_segments, generator.as_ref());
                        if let Some(a) = alias {
                            format!("{} AS {}", expr, generator.quote_identifier(a))
                        } else {
                            expr
                        }
                    }
                    Expr::FunctionCall { name, args, alias } => {
                        if name.eq_ignore_ascii_case("case") {
                            // case(when_cond, then_val, else_val) -> CASE WHEN ... THEN ... ELSE ... END
                            if args.len() >= 3 {
                                let cond_str = args[0].to_string();
                                let then_str = args[1].to_string();
                                let else_str = args[2].to_string();

                                // Arg 0: WHEN condition (Raw preferred)
                                let cond_sql =
                                    if cond_str.starts_with('{') && cond_str.ends_with('}') {
                                        cond_str[1..cond_str.len() - 1].to_string()
                                    } else {
                                        cond_str
                                    };

                                // Arg 1: THEN value (Quoted unless raw)
                                let then_sql =
                                    if then_str.starts_with('{') && then_str.ends_with('}') {
                                        then_str[1..then_str.len() - 1].to_string()
                                    } else {
                                        generator.quote_identifier(&then_str)
                                    };

                                // Arg 2: ELSE value (Quoted unless raw)
                                let else_sql =
                                    if else_str.starts_with('{') && else_str.ends_with('}') {
                                        else_str[1..else_str.len() - 1].to_string()
                                    } else {
                                        generator.quote_identifier(&else_str)
                                    };

                                let expr = format!(
                                    "CASE WHEN {} THEN {} ELSE {} END",
                                    cond_sql, then_sql, else_sql
                                );
                                if let Some(a) = alias {
                                    format!("{} AS {}", expr, generator.quote_identifier(a))
                                } else {
                                    expr
                                }
                            } else {
                                // Invalid case call, fallback to standard function
                                let args_sql: Vec<String> =
                                    args.iter().map(|a| a.to_string()).collect();
                                let expr = format!("CASE({})", args_sql.join(", "));
                                if let Some(a) = alias {
                                    format!("{} AS {}", expr, generator.quote_identifier(a))
                                } else {
                                    expr
                                }
                            }
                        } else {
                            // Standard Function - transpile each arg expression
                            let Some(function) = render_function_name(name) else {
                                return "/* ERROR: Invalid function name */".to_string();
                            };
                            let args_sql = crate::transpiler::render_function_args(
                                args,
                                generator.as_ref(),
                                |a| {
                                    let arg_str = a.to_string();
                                    if arg_str.starts_with('{') && arg_str.ends_with('}') {
                                        // Raw SQL block: {content} -> content
                                        arg_str[1..arg_str.len() - 1].to_string()
                                    } else {
                                        // For expressions (especially binary), don't quote
                                        match a {
                                            Expr::Named(n) => {
                                                // Don't quote if already quoted, is a param, or is numeric
                                                render_named_expression(n, generator.as_ref(), cmd)
                                            }
                                            Expr::Star => "*".to_string(),
                                            _ => {
                                                render_expr_for_orderby(a, generator.as_ref(), cmd)
                                            }
                                        }
                                    }
                                },
                            );
                            let args_sql = match args_sql {
                                Ok(args_sql) => args_sql,
                                Err(error) => return error,
                            };
                            let expr = format!("{}({})", function, args_sql);
                            if let Some(a) = alias {
                                format!("{} AS {}", expr, generator.quote_identifier(a))
                            } else {
                                expr
                            }
                        }
                    }
                    Expr::Aggregate { alias, .. } => {
                        // Render aggregate function: COUNT(*), COUNT(DISTINCT col), SUM(col), etc.
                        let expr = render_aggregate(c, generator.as_ref(), cmd);

                        if let Some(a) = alias {
                            format!("{} AS {}", expr, generator.quote_identifier(a))
                        } else {
                            expr
                        }
                    }
                    Expr::Cast {
                        expr,
                        target_type,
                        alias,
                    } => {
                        let Some(target_type) = checked_sql_type_fragment(target_type) else {
                            return "/* ERROR: Invalid cast target type */".to_string();
                        };
                        let cast_expr = format!(
                            "{}::{}",
                            render_expr_for_orderby(expr, generator.as_ref(), cmd),
                            target_type
                        );
                        if let Some(a) = alias {
                            format!("{} AS {}", cast_expr, generator.quote_identifier(a))
                        } else {
                            cast_expr
                        }
                    }
                    Expr::Collate {
                        expr,
                        collation,
                        alias,
                    } => {
                        let expr = render_expr_for_orderby(expr, generator.as_ref(), cmd);
                        let collation = render_qualified_identifier(collation, generator.as_ref());
                        let collate_expr = format!("{expr} COLLATE {collation}");
                        if let Some(a) = alias {
                            format!("{} AS {}", collate_expr, generator.quote_identifier(a))
                        } else {
                            collate_expr
                        }
                    }
                    Expr::FieldAccess { expr, field, alias } => {
                        let field_expr = format!(
                            "({}).{}",
                            render_expr_for_orderby(expr, generator.as_ref(), cmd),
                            render_qualified_identifier(field, generator.as_ref())
                        );
                        if let Some(a) = alias {
                            format!("{} AS {}", field_expr, generator.quote_identifier(a))
                        } else {
                            field_expr
                        }
                    }
                    Expr::Binary { alias, .. }
                    | Expr::SpecialFunction { alias, .. }
                    | Expr::ArrayConstructor { alias, .. }
                    | Expr::RowConstructor { alias, .. }
                    | Expr::Subscript { alias, .. }
                    | Expr::ArraySlice { alias, .. } => append_alias(
                        render_expr_for_orderby(c, generator.as_ref(), cmd),
                        alias,
                        generator.as_ref(),
                    ),
                    Expr::Literal(value) => {
                        render_value_for_expression(value, generator.as_ref(), cmd)
                    }
                    Expr::FunctionArg { .. } => {
                        crate::transpiler::MISPLACED_FUNCTION_ARG_SQL.to_string()
                    }
                    Expr::Subquery { query, alias } => append_alias(
                        format!("({})", read_only_subquery_sql(query)),
                        alias,
                        generator.as_ref(),
                    ),
                    Expr::Exists {
                        query,
                        negated,
                        alias,
                    } => {
                        let exists_sql = if *negated {
                            format!("NOT EXISTS ({})", read_only_subquery_sql(query))
                        } else {
                            format!("EXISTS ({})", read_only_subquery_sql(query))
                        };
                        append_alias(exists_sql, alias, generator.as_ref())
                    }
                    Expr::Window {
                        name,
                        func,
                        params,
                        filter,
                        partition,
                        order,
                        frame,
                    } => {
                        // Window function: FUNC(args) OVER (PARTITION BY x ORDER BY y) AS alias
                        let params_str = if params.is_empty() {
                            String::new()
                        } else {
                            params
                                .iter()
                                .map(|v| render_expr_for_orderby(v, generator.as_ref(), cmd))
                                .collect::<Vec<_>>()
                                .join(", ")
                        };
                        let Some(function) = render_function_name(func) else {
                            return "/* ERROR: Invalid window function name */".to_string();
                        };

                        let mut over_clause = String::from("OVER (");
                        if !partition.is_empty() {
                            over_clause.push_str("PARTITION BY ");
                            let quoted_partition: Vec<String> = partition
                                .iter()
                                .map(|p| render_named_reference(p, generator.as_ref(), cmd))
                                .collect();
                            over_clause.push_str(&quoted_partition.join(", "));
                            if !order.is_empty() {
                                over_clause.push(' ');
                            }
                        }
                        if !order.is_empty() {
                            over_clause.push_str("ORDER BY ");
                            let order_parts: Vec<String> = order
                                .iter()
                                .map(|cage| {
                                    let col_str = if let Some(cond) = cage.conditions.first() {
                                        match &cond.left {
                                            Expr::Named(n) => {
                                                render_named_reference(n, generator.as_ref(), cmd)
                                            }
                                            expr => render_expr_for_orderby(
                                                expr,
                                                generator.as_ref(),
                                                cmd,
                                            ),
                                        }
                                    } else {
                                        return String::new();
                                    };
                                    match &cage.kind {
                                        CageKind::Sort(SortOrder::Asc) => {
                                            format!("{} ASC", col_str)
                                        }
                                        CageKind::Sort(SortOrder::Desc) => {
                                            format!("{} DESC", col_str)
                                        }
                                        CageKind::Sort(SortOrder::AscNullsFirst) => {
                                            format!("{} ASC NULLS FIRST", col_str)
                                        }
                                        CageKind::Sort(SortOrder::AscNullsLast) => {
                                            format!("{} ASC NULLS LAST", col_str)
                                        }
                                        CageKind::Sort(SortOrder::DescNullsFirst) => {
                                            format!("{} DESC NULLS FIRST", col_str)
                                        }
                                        CageKind::Sort(SortOrder::DescNullsLast) => {
                                            format!("{} DESC NULLS LAST", col_str)
                                        }
                                        _ => String::new(),
                                    }
                                })
                                .filter(|s| !s.is_empty())
                                .collect();
                            over_clause.push_str(&order_parts.join(", "));
                        }

                        if let Some(fr) = frame {
                            over_clause.push(' ');
                            match fr.to_sql() {
                                Ok(frame_sql) => over_clause.push_str(&frame_sql),
                                Err(reason) => return format!("/* ERROR: {reason} */"),
                            }
                        }

                        over_clause.push(')');
                        format!(
                            "{}({}){} {} AS {}",
                            function,
                            params_str,
                            aggregate_filter_sql(filter, generator.as_ref(), cmd),
                            over_clause,
                            generator.quote_identifier(name)
                        )
                    }
                    Expr::Def { .. } | Expr::Mod { .. } | Expr::Default => {
                        "/* ERROR: Invalid select expression */".to_string()
                    }
                }
            })
            .collect();
        sql.push_str(&cols.join(", "));
    }

    // FROM (with optional ONLY for inheritance control)
    if let Some(source) = &cmd.from_source {
        sql.push_str(" FROM ");
        sql.push_str(&render_from_source(cmd, source, generator.as_ref()));
    } else if cmd.only_table {
        sql.push_str(" FROM ONLY ");
        sql.push_str(&render_table_reference(&cmd.table, generator.as_ref()));
    } else {
        sql.push_str(" FROM ");
        sql.push_str(&render_table_reference(&cmd.table, generator.as_ref()));
    }

    // TABLESAMPLE
    let sample = cmd.sample.or_else(|| {
        cmd.cages.iter().find_map(|cage| match &cage.kind {
            CageKind::Sample(percent) => Some((SampleMethod::Bernoulli, *percent as f64, None)),
            _ => None,
        })
    });

    if let Some((method, percent, seed)) = sample {
        let method_str = match method {
            SampleMethod::Bernoulli => "BERNOULLI",
            SampleMethod::System => "SYSTEM",
        };
        sql.push_str(&format!(" TABLESAMPLE {}({})", method_str, percent));
        if let Some(s) = seed {
            sql.push_str(&format!(" REPEATABLE({})", s));
        }
    }

    // JOINS
    for join in &cmd.joins {
        let (kind, needs_on) = match join.kind {
            JoinKind::Inner => ("INNER", true),
            JoinKind::Left => ("LEFT", true),
            JoinKind::Right => ("RIGHT", true),
            JoinKind::Lateral => ("LATERAL", true),
            JoinKind::Full => ("FULL OUTER", true),
            JoinKind::Cross => ("CROSS", false),
        };
        // Join: target.source_singular_id = source.id
        let source_base = table_reference_base(&cmd.table);
        let source_singular = source_base
            .rsplit('.')
            .next()
            .unwrap_or(source_base)
            .trim_end_matches('s');

        let target_table = render_table_reference(&join.table, generator.as_ref());
        let target_qualifier = table_reference_sql_qualifier(&join.table)
            .map(|qualifier| generator.quote_identifier(qualifier))
            .unwrap_or_else(|| generator.quote_identifier(&join.table));
        let source_fk = format!("{}_id", source_singular);
        let source_table = table_reference_sql_qualifier(&cmd.table)
            .map(|qualifier| generator.quote_identifier(qualifier))
            .unwrap_or_else(|| generator.quote_identifier(&cmd.table));

        if let Some(on_conds) = &join.on {
            let on_sql: Vec<String> = on_conds
                .iter()
                .map(|c| c.to_sql(generator.as_ref(), Some(cmd)))
                .collect();
            sql.push_str(&format!(
                " {} JOIN {} ON {}",
                kind,
                target_table,
                on_sql.join(" AND ")
            ));
        } else if join.on_true {
            // Explicit ON TRUE (unconditional join, used for CTE joins)
            sql.push_str(&format!(" {} JOIN {} ON TRUE", kind, target_table));
        } else if needs_on {
            sql.push_str(&format!(
                " {} JOIN {} ON {}.{} = {}.id",
                kind,
                target_table,
                target_qualifier,
                generator.quote_identifier(&source_fk),
                source_table
            ));
        } else {
            sql.push_str(&format!(" {} JOIN {}", kind, target_table));
        }
    }

    // Process cages
    let mut where_groups: Vec<String> = Vec::new();
    let mut order_by_clauses: Vec<String> = Vec::new();
    let mut limit: Option<usize> = None;
    let mut offset: Option<usize> = None;

    for cage in &cmd.cages {
        match &cage.kind {
            CageKind::Filter => {
                if !cage.conditions.is_empty() {
                    let joiner = match cage.logical_op {
                        LogicalOp::And => " AND ",
                        LogicalOp::Or => " OR ",
                    };
                    let conditions: Vec<String> = cage
                        .conditions
                        .iter()
                        .map(|c| c.to_sql(generator.as_ref(), Some(cmd)))
                        .collect();
                    let group = conditions.join(joiner);
                    // Wrap OR groups in parentheses for correct precedence
                    if cage.logical_op == LogicalOp::Or && cage.conditions.len() > 1 {
                        where_groups.push(format!("({})", group));
                    } else {
                        where_groups.push(group);
                    }
                }
            }
            CageKind::Sort(order) => {
                if let Some(cond) = cage.conditions.first() {
                    let dir = match order {
                        SortOrder::Asc => "ASC",
                        SortOrder::Desc => "DESC",
                        SortOrder::AscNullsFirst => "ASC NULLS FIRST",
                        SortOrder::AscNullsLast => "ASC NULLS LAST",
                        SortOrder::DescNullsFirst => "DESC NULLS FIRST",
                        SortOrder::DescNullsLast => "DESC NULLS LAST",
                    };
                    let col_sql = render_expr_for_orderby(&cond.left, generator.as_ref(), cmd);
                    order_by_clauses.push(format!("{} {}", col_sql, dir));
                }
            }
            CageKind::Limit(n) => {
                limit = Some(*n);
            }
            CageKind::Offset(n) => {
                offset = Some(*n);
            }
            CageKind::Payload => {
                // Not used in SELECT
            }
            CageKind::Sample(_) => {
                // Handled separately after FROM clause
            }
            CageKind::Qualify => {
                // Will be processed separately after ORDER BY for QUALIFY clause
            }
            CageKind::Partition => {
                // GROUP BY keys, rendered by build_group_by_clause
            }
        }
    }

    // WHERE - each cage group is joined with AND
    if !where_groups.is_empty() {
        sql.push_str(" WHERE ");
        sql.push_str(&where_groups.join(" AND "));
    }

    sql.push_str(&build_group_by_clause(cmd, columns, generator.as_ref()));

    // HAVING (filter on aggregates)
    if !cmd.having.is_empty() {
        let having_conds: Vec<String> = cmd
            .having
            .iter()
            .map(|c| c.to_sql(generator.as_ref(), Some(cmd)))
            .collect();
        sql.push_str(" HAVING ");
        sql.push_str(&having_conds.join(" AND "));
    }

    if !order_by_clauses.is_empty() {
        sql.push_str(" ORDER BY ");
        sql.push_str(&order_by_clauses.join(", "));
    }

    // QUALIFY (Snowflake, BigQuery, Databricks) - filter on window function results
    // Appears after ORDER BY, before LIMIT
    for cage in &cmd.cages {
        if let CageKind::Qualify = cage.kind
            && !cage.conditions.is_empty()
        {
            let qualify_conds: Vec<String> = cage
                .conditions
                .iter()
                .map(|c| c.to_sql(generator.as_ref(), Some(cmd)))
                .collect();
            sql.push_str(" QUALIFY ");
            sql.push_str(&qualify_conds.join(" AND "));
        }
    }

    sql.push_str(&generator.limit_offset(limit, offset));
    append_fetch_clause(&mut sql, cmd.fetch);

    if !cmd.set_ops.is_empty() && set_operand_has_branch_clauses(cmd) {
        sql = wrap_set_operand_sql(sql, dialect);
    }

    // SET OPERATIONS (UNION, INTERSECT, EXCEPT)
    for (set_op, other_cmd) in &cmd.set_ops {
        sql.push_str(&format!(
            " {} {}",
            set_op.sql_keyword(),
            build_set_operand(other_cmd, dialect)
        ));
    }

    // FOR UPDATE/SHARE (row locking)
    sql.push_str(&lock_clause_sql(cmd, generator.as_ref()));

    sql
}

fn lock_clause_sql(cmd: &Qail, generator: &dyn SqlGenerator) -> String {
    let Some(lock) = &cmd.lock_mode else {
        return if cmd.skip_locked || cmd.lock_nowait || !cmd.lock_of.is_empty() {
            " /* ERROR: NOWAIT, SKIP LOCKED and OF require a row lock mode */".to_string()
        } else {
            String::new()
        };
    };
    if cmd.skip_locked && cmd.lock_nowait {
        return " /* ERROR: NOWAIT and SKIP LOCKED are mutually exclusive */".to_string();
    }
    let mut sql = match lock {
        LockMode::Update => " FOR UPDATE",
        LockMode::NoKeyUpdate => " FOR NO KEY UPDATE",
        LockMode::Share => " FOR SHARE",
        LockMode::KeyShare => " FOR KEY SHARE",
    }
    .to_string();
    if !cmd.lock_of.is_empty() {
        // PostgreSQL rejects qualified names here; OF takes FROM names or aliases.
        if cmd
            .lock_of
            .iter()
            .any(|name| name.is_empty() || name.contains('.') || name.contains('\0'))
        {
            return " /* ERROR: Invalid row lock OF name */".to_string();
        }
        let names: Vec<String> = cmd
            .lock_of
            .iter()
            .map(|name| generator.quote_identifier(name))
            .collect();
        sql.push_str(" OF ");
        sql.push_str(&names.join(", "));
    }
    if cmd.lock_nowait {
        sql.push_str(" NOWAIT");
    } else if cmd.skip_locked {
        sql.push_str(" SKIP LOCKED");
    }
    sql
}

pub(super) fn build_set_operand(cmd: &Qail, dialect: Dialect) -> String {
    let sql = build_select(cmd, dialect);
    if set_operand_needs_wrapper(cmd) {
        wrap_set_operand_sql(sql, dialect)
    } else {
        sql
    }
}

fn set_operand_needs_wrapper(cmd: &Qail) -> bool {
    !cmd.set_ops.is_empty() || set_operand_has_branch_clauses(cmd)
}

fn set_operand_has_branch_clauses(cmd: &Qail) -> bool {
    cmd.fetch.is_some()
        || cmd.cages.iter().any(|cage| {
            matches!(
                cage.kind,
                CageKind::Sort(_) | CageKind::Limit(_) | CageKind::Offset(_)
            )
        })
}

fn wrap_set_operand_sql(sql: String, dialect: Dialect) -> String {
    match dialect {
        Dialect::Postgres => format!("({sql})"),
    }
}

/// ` GROUP BY ...` from [`Qail::group_by_clause`], or an empty string.
fn build_group_by_clause(
    cmd: &Qail,
    columns: &[Expr],
    generator: &dyn crate::transpiler::SqlGenerator,
) -> String {
    let render_keys = |keys: &[std::borrow::Cow<'_, Expr>]| {
        keys.iter()
            .map(|key| match key.as_ref() {
                Expr::Named(name) => render_named_reference(name, generator, cmd),
                other => render_expr_for_orderby(other, generator, cmd),
            })
            .collect::<Vec<_>>()
            .join(", ")
    };

    match cmd.group_by_clause(columns) {
        Ok(None) => String::new(),
        Ok(Some(GroupByClause::Keys(keys))) => format!(" GROUP BY {}", render_keys(&keys)),
        Ok(Some(GroupByClause::Rollup(keys))) => {
            format!(" GROUP BY ROLLUP({})", render_keys(&keys))
        }
        Ok(Some(GroupByClause::Cube(keys))) => format!(" GROUP BY CUBE({})", render_keys(&keys)),
        Ok(Some(GroupByClause::GroupingSets(sets))) => {
            let sets: Vec<String> = sets.iter().map(|s| format!("({})", s.join(", "))).collect();
            format!(" GROUP BY GROUPING SETS ({})", sets.join(", "))
        }
        Err(message) => format!(" GROUP BY /* ERROR: {message} */"),
    }
}

fn append_fetch_clause(sql: &mut String, fetch: Option<(u64, bool)>) {
    if let Some((count, with_ties)) = fetch {
        if with_ties {
            sql.push_str(&format!(" FETCH FIRST {} ROWS WITH TIES", count));
        } else {
            sql.push_str(&format!(" FETCH FIRST {} ROWS ONLY", count));
        }
    }
}

/// Render an expression for ORDER BY (and potentially other contexts).
/// Handles CASE, Binary, FunctionCall, SpecialFunction, and Named expressions.
/// An `Expr::Aggregate` call without its alias.
fn render_aggregate(
    expr: &Expr,
    generator: &dyn crate::transpiler::SqlGenerator,
    cmd: &Qail,
) -> String {
    crate::transpiler::aggregate::aggregate_call_sql(
        expr,
        &|col| render_named_reference(col, generator, cmd),
        &|arg| render_expr_for_orderby(arg, generator, cmd),
        &|conditions| {
            conditions
                .iter()
                .map(|condition| condition.to_sql(generator, Some(cmd)))
                .collect::<Vec<_>>()
                .join(" AND ")
        },
    )
}

fn render_expr_for_orderby(
    expr: &Expr,
    generator: &dyn crate::transpiler::SqlGenerator,
    cmd: &Qail,
) -> String {
    match expr {
        Expr::Star => "*".to_string(),
        Expr::Named(name) => {
            // Don't quote if already quoted, is a param, or is numeric
            if name.starts_with('\'')
                || name.starts_with('"')
                || name.starts_with(':')
                || name.starts_with('$')
                || name.parse::<f64>().is_ok()
                || name.eq_ignore_ascii_case("NULL")
                || name.eq_ignore_ascii_case("TRUE")
                || name.eq_ignore_ascii_case("FALSE")
            {
                name.clone()
            } else {
                render_named_reference(name, generator, cmd)
            }
        }
        Expr::Aliased { name, .. } => render_named_reference(name, generator, cmd),
        Expr::Literal(value) => render_value_for_expression(value, generator, cmd),
        Expr::Aggregate { .. } => render_aggregate(expr, generator, cmd),
        Expr::Case {
            when_clauses,
            else_value,
            ..
        } => {
            let mut case_sql = String::from("CASE");
            for (cond, val) in when_clauses {
                case_sql.push_str(&format!(
                    " WHEN {} THEN {}",
                    cond.to_sql(generator, Some(cmd)),
                    render_expr_for_orderby(val, generator, cmd)
                ));
            }
            if let Some(e) = else_value {
                case_sql.push_str(&format!(
                    " ELSE {}",
                    render_expr_for_orderby(e, generator, cmd)
                ));
            }
            case_sql.push_str(" END");
            case_sql
        }
        Expr::Binary {
            left, op, right, ..
        } => {
            let left_sql = render_expr_for_orderby(left, generator, cmd);
            let right_sql = render_expr_for_orderby(right, generator, cmd);
            match op {
                op if op.is_postfix() => format!("({} {})", left_sql, op),
                _ => op.infix_sql(&left_sql, &right_sql),
            }
        }
        Expr::FunctionCall { name, args, .. } => {
            let Some(function) = render_function_name(name) else {
                return "/* ERROR: Invalid function name */".to_string();
            };
            match crate::transpiler::render_function_args(args, generator, |a| {
                render_expr_for_orderby(a, generator, cmd)
            }) {
                Ok(args) => format!("{function}({args})"),
                Err(error) => error,
            }
        }
        Expr::FunctionArg { .. } => crate::transpiler::MISPLACED_FUNCTION_ARG_SQL.to_string(),
        Expr::SpecialFunction { name, args, .. } => {
            let Some(function) = render_function_name(name) else {
                return "/* ERROR: Invalid function name */".to_string();
            };
            let mut parts = Vec::new();
            for (keyword, expr) in args {
                let expr = render_expr_for_orderby(expr, generator, cmd);
                if let Some(keyword) = keyword {
                    let Some(keyword) = render_sql_keyword(keyword) else {
                        return "/* ERROR: Invalid function keyword */".to_string();
                    };
                    parts.push(format!("{keyword} {expr}"));
                } else {
                    parts.push(expr);
                }
            }
            format!("{function}({})", parts.join(" "))
        }
        Expr::JsonAccess {
            column,
            path_segments,
            ..
        } => render_json_access(column, path_segments, generator),
        Expr::Cast {
            expr, target_type, ..
        } => {
            let Some(target_type) = checked_sql_type_fragment(target_type) else {
                return "/* ERROR: Invalid cast target type */".to_string();
            };
            format!(
                "{}::{}",
                render_expr_for_orderby(expr, generator, cmd),
                target_type
            )
        }
        Expr::Collate {
            expr, collation, ..
        } => format!(
            "{} COLLATE {}",
            render_expr_for_orderby(expr, generator, cmd),
            render_qualified_identifier(collation, generator)
        ),
        Expr::FieldAccess { expr, field, .. } => format!(
            "({}).{}",
            render_expr_for_orderby(expr, generator, cmd),
            render_qualified_identifier(field, generator)
        ),
        Expr::ArrayConstructor { elements, .. } => {
            let elements = elements
                .iter()
                .map(|element| render_expr_for_orderby(element, generator, cmd))
                .collect::<Vec<_>>()
                .join(", ");
            format!("ARRAY[{elements}]")
        }
        Expr::RowConstructor { elements, .. } => {
            let elements = elements
                .iter()
                .map(|element| render_expr_for_orderby(element, generator, cmd))
                .collect::<Vec<_>>()
                .join(", ");
            format!("ROW({elements})")
        }
        Expr::Subscript { expr, index, .. } => subscript_sql(
            expr,
            render_expr_for_orderby(expr, generator, cmd),
            &render_expr_for_orderby(index, generator, cmd),
        ),
        Expr::ArraySlice {
            expr, lower, upper, ..
        } => subscript_sql(
            expr,
            render_expr_for_orderby(expr, generator, cmd),
            &slice_bounds_sql(lower, upper, |bound| {
                render_expr_for_orderby(bound, generator, cmd)
            }),
        ),
        Expr::Subquery { query, .. } => format!("({})", read_only_subquery_sql(query)),
        Expr::Exists { query, negated, .. } => {
            if *negated {
                format!("NOT EXISTS ({})", read_only_subquery_sql(query))
            } else {
                format!("EXISTS ({})", read_only_subquery_sql(query))
            }
        }
        Expr::Def { .. } | Expr::Mod { .. } | Expr::Window { .. } | Expr::Default => {
            "/* ERROR: Invalid select expression */".to_string()
        }
    }
}

fn append_alias(
    expr_sql: String,
    alias: &Option<String>,
    generator: &dyn crate::transpiler::SqlGenerator,
) -> String {
    if let Some(alias) = alias {
        format!("{} AS {}", expr_sql, generator.quote_identifier(alias))
    } else {
        expr_sql
    }
}

fn render_named_expression(
    name: &str,
    generator: &dyn crate::transpiler::SqlGenerator,
    cmd: &Qail,
) -> String {
    if name.starts_with('\'')
        || name.starts_with('"')
        || name.starts_with(':')
        || name.starts_with('$')
        || name.parse::<f64>().is_ok()
        || name.eq_ignore_ascii_case("NULL")
        || name.eq_ignore_ascii_case("TRUE")
        || name.eq_ignore_ascii_case("FALSE")
    {
        name.to_string()
    } else {
        render_named_reference(name, generator, cmd)
    }
}

fn render_named_reference(
    name: &str,
    generator: &dyn crate::transpiler::SqlGenerator,
    cmd: &Qail,
) -> String {
    resolve_known_col_syntax(name, cmd, generator)
        .unwrap_or_else(|| generator.quote_identifier(name))
}

fn render_value_for_expression(
    value: &Value,
    generator: &dyn crate::transpiler::SqlGenerator,
    cmd: &Qail,
) -> String {
    match value {
        Value::Column(column) => render_named_reference(column, generator, cmd),
        Value::Expr(expr) => render_expr_for_orderby(expr, generator, cmd),
        Value::Subquery(query) => format!("({})", read_only_subquery_sql(query)),
        Value::Function(function) => render_raw_function_value(function),
        Value::NamedParam(name) => render_named_param(name),
        Value::Array(values) => {
            let values = values
                .iter()
                .map(|value| render_value_for_expression(value, generator, cmd))
                .collect::<Vec<_>>()
                .join(", ");
            format!("({values})")
        }
        _ => value.to_string(),
    }
}

/// `(subquery) AS alias (cols)` or `func(args) [WITH ORDINALITY] AS alias (cols)`.
fn render_from_source(cmd: &Qail, source: &FromSource, generator: &dyn SqlGenerator) -> String {
    if let Some(error) = from_source_error(cmd, source) {
        return format!("/* ERROR: {error} */");
    }
    let mut sql = match source {
        FromSource::Subquery { query, .. } => format!("({})", read_only_subquery_sql(query)),
        FromSource::Function {
            name,
            args,
            with_ordinality,
            ..
        } => {
            let Some(function) = render_function_name(name) else {
                return "/* ERROR: Invalid FROM function name */".to_string();
            };
            let args = args
                .iter()
                .map(|arg| render_expr_for_orderby(arg, generator, cmd))
                .collect::<Vec<_>>()
                .join(", ");
            let ordinality = if *with_ordinality {
                " WITH ORDINALITY"
            } else {
                ""
            };
            format!("{function}({args}){ordinality}")
        }
    };
    sql.push_str(" AS ");
    sql.push_str(&generator.quote_identifier(source.alias()));
    let columns = source.column_alias_list();
    if !columns.is_empty() {
        let columns = columns
            .iter()
            .map(|column| generator.quote_identifier(column))
            .collect::<Vec<_>>()
            .join(", ");
        sql.push_str(&format!(" ({columns})"));
    }
    sql
}

/// Shape errors shared by preview and native output for a typed FROM item.
pub fn from_source_error(cmd: &Qail, source: &FromSource) -> Option<String> {
    if !matches!(
        cmd.action,
        Action::Get
            | Action::With
            | Action::Cnt
            | Action::Export
            | Action::Explain
            | Action::ExplainAnalyze
    ) {
        return Some(format!(
            "typed FROM source is supported only for SELECT, got {}",
            cmd.action
        ));
    }
    if cmd.table != source.alias() {
        return Some(format!(
            "typed FROM source alias `{}` must equal the command table `{}`",
            source.alias(),
            cmd.table
        ));
    }
    if cmd.only_table {
        return Some("ONLY applies to tables, not a typed FROM source".to_string());
    }
    let sampled = cmd.sample.is_some()
        || cmd
            .cages
            .iter()
            .any(|cage| matches!(cage.kind, CageKind::Sample(_)));
    if sampled {
        return Some("TABLESAMPLE applies to tables, not a typed FROM source".to_string());
    }
    None
}

fn render_function_name(name: &str) -> Option<String> {
    if name.is_empty()
        || name.contains('\0')
        || name.split('.').any(str::is_empty)
        || !name
            .bytes()
            .all(|b| b.is_ascii_alphanumeric() || b == b'_' || b == b'.')
    {
        None
    } else {
        Some(name.to_uppercase())
    }
}

fn render_sql_keyword(keyword: &str) -> Option<String> {
    if keyword.is_empty()
        || keyword.contains('\0')
        || !keyword
            .bytes()
            .all(|b| b.is_ascii_alphabetic() || b == b'_')
    {
        None
    } else {
        Some(keyword.to_uppercase())
    }
}

fn render_named_param(name: &str) -> String {
    let mut chars = name.chars();
    let Some(first) = chars.next() else {
        return "/* ERROR: Invalid parameter name */".to_string();
    };
    if !(first.is_ascii_alphabetic() || first == '_')
        || !chars.all(|ch| ch.is_ascii_alphanumeric() || ch == '_')
    {
        return "/* ERROR: Invalid parameter name */".to_string();
    }
    format!(":{}", name)
}

fn render_raw_function_value(value: &str) -> String {
    if value.len() > 1024
        || value.contains('\0')
        || value.contains(';')
        || value.contains("--")
        || value.contains("/*")
        || value.contains("*/")
    {
        "/* ERROR: Invalid function expression */".to_string()
    } else {
        value.to_string()
    }
}

fn checked_sql_type_fragment(fragment: &str) -> Option<String> {
    let fragment = fragment.trim();
    if fragment.is_empty()
        || fragment.contains('\0')
        || fragment.contains(';')
        || fragment.contains('\'')
        || fragment.contains('"')
        || fragment.contains("--")
        || fragment.contains("/*")
        || fragment.contains("*/")
        || !fragment.bytes().all(|b| {
            b.is_ascii_alphanumeric()
                || matches!(
                    b,
                    b'_' | b'.' | b' ' | b'(' | b')' | b',' | b'[' | b']' | b'%' | b'+' | b'-'
                )
        })
    {
        None
    } else {
        Some(fragment.to_string())
    }
}

fn render_qualified_identifier(value: &str, generator: &dyn SqlGenerator) -> String {
    if value.is_empty() || value.as_bytes().contains(&0) || value.split('.').any(str::is_empty) {
        "/* ERROR: Invalid identifier */".to_string()
    } else {
        generator.quote_identifier(value)
    }
}

fn render_json_access(
    column: &str,
    path_segments: &[(JsonPathSegment, bool)],
    generator: &dyn SqlGenerator,
) -> String {
    let mut result = generator.quote_identifier(column);
    for (segment, as_text) in path_segments {
        let op = if *as_text { "->>" } else { "->" };
        result.push_str(&format!("{}{}", op, segment));
    }
    result
}

/// ` FILTER (WHERE ...)` for an aggregate call, or "" when there is none.
pub(super) fn aggregate_filter_sql(
    filter: &Option<Vec<Condition>>,
    generator: &dyn SqlGenerator,
    cmd: &Qail,
) -> String {
    match filter {
        Some(conditions) if !conditions.is_empty() => {
            let filter_parts: Vec<String> = conditions
                .iter()
                .map(|c| c.to_sql(generator, Some(cmd)))
                .collect();
            format!(" FILTER (WHERE {})", filter_parts.join(" AND "))
        }
        _ => String::new(),
    }
}