qail-core 3.0.1

use super::ToSql;
use super::traits::{SqlGenerator, escape_sql_string_literal};
use crate::ast::*;
use crate::transpiler::identifier::{qualifier_for_column_path, qualifier_for_column_reference};

/// Context for parameterized query building.
#[derive(Debug, Default)]
pub struct ParamContext {
    /// Current parameter index (1-based).
    pub index: usize,
    /// Collected parameter values in order
    pub params: Vec<Value>,
    /// Names of named parameters in order (for :name → $n mapping)
    pub named_params: Vec<String>,
}

impl ParamContext {
    /// Create a new empty parameter context.
    pub fn new() -> Self {
        Self {
            index: 0,
            params: Vec::new(),
            named_params: Vec::new(),
        }
    }

    /// Add a value and return the placeholder for it.
    pub fn add_param(&mut self, value: Value, generator: &dyn SqlGenerator) -> String {
        self.index += 1;
        self.params.push(value);
        generator.placeholder(self.index)
    }

    /// Add a named parameter and return the placeholder for it.
    pub fn add_named_param(&mut self, name: String, generator: &dyn SqlGenerator) -> String {
        self.index += 1;
        self.named_params.push(name);
        generator.placeholder(self.index)
    }
}

/// Heuristic:
/// 1. Split by '.'
/// 2. If single part -> quote_identifier
/// 3. If multiple parts:
///    - If first part matches table name or any join alias -> Treat as "Table"."Col".
///    - Else -> Treat as "Col"->"Field" (JSON).
pub(crate) fn resolve_known_col_syntax(
    col: &str,
    cmd: &Qail,
    generator: &dyn SqlGenerator,
) -> Option<String> {
    let parts: Vec<&str> = col.split('.').collect();
    if parts.len() <= 1 {
        return None;
    }

    let first = parts[0];

    if let Some((sql_qualifier, consumed)) = qualifier_for_column_path(&cmd.table, &parts)
        && consumed < parts.len()
    {
        return Some(render_qualified_column(
            sql_qualifier,
            &parts[consumed..],
            generator,
        ));
    }

    let auxiliary_tables = cmd
        .joins
        .iter()
        .map(|join| join.table.as_str())
        .chain(cmd.from_tables.iter().map(String::as_str))
        .chain(cmd.using_tables.iter().map(String::as_str));

    for table_ref in auxiliary_tables {
        if let Some((sql_qualifier, consumed)) = qualifier_for_column_path(table_ref, &parts)
            && consumed < parts.len()
        {
            return Some(render_qualified_column(
                sql_qualifier,
                &parts[consumed..],
                generator,
            ));
        }

        if let Some(sql_qualifier) = qualifier_for_column_reference(table_ref, first) {
            return Some(render_qualified_column(
                sql_qualifier,
                &parts[1..],
                generator,
            ));
        }
    }

    None
}

fn resolve_col_syntax(col: &str, cmd: &Qail, generator: &dyn SqlGenerator) -> String {
    if col.starts_with('{') && col.ends_with('}') {
        return col[1..col.len() - 1].to_string();
    }

    let parts: Vec<&str> = col.split('.').collect();
    if parts.len() <= 1 {
        return generator.quote_identifier(col);
    }

    if let Some(sql) = resolve_known_col_syntax(col, cmd, generator) {
        return sql;
    }

    // Default: treated as JSON access on the first part
    let col_name = parts[0];
    let path = &parts[1..];
    generator.json_access(col_name, path)
}

fn render_qualified_column(
    qualifier: &str,
    column_parts: &[&str],
    generator: &dyn SqlGenerator,
) -> String {
    if column_parts == ["*"] {
        return format!("{}.*", generator.quote_identifier(qualifier));
    }
    format!(
        "{}.{}",
        generator.quote_identifier(qualifier),
        generator.quote_identifier(&column_parts.join("."))
    )
}

/// `base[inner]`, with the base parenthesized when PostgreSQL cannot subscript
/// it directly (function results, constructors, literals).
pub(crate) fn subscript_sql(base: &Expr, base_sql: String, inner: &str) -> String {
    if base.needs_parens_for_subscript() {
        format!("({base_sql})[{inner}]")
    } else {
        format!("{base_sql}[{inner}]")
    }
}

/// Slice bounds `lower:upper`; an omitted bound renders empty.
pub(crate) fn slice_bounds_sql(
    lower: &Option<Box<Expr>>,
    upper: &Option<Box<Expr>>,
    mut render: impl FnMut(&Expr) -> String,
) -> String {
    let lower = lower.as_deref().map(&mut render).unwrap_or_default();
    let upper = upper.as_deref().map(&mut render).unwrap_or_default();
    format!("{lower}:{upper}")
}

fn resolve_text_search_vector(
    expr: &Expr,
    generator: &dyn SqlGenerator,
    context: Option<&Qail>,
) -> Option<String> {
    let Expr::Named(columns) = expr else {
        return None;
    };

    let parts: Vec<String> = columns
        .split(',')
        .map(str::trim)
        .filter(|column| !column.is_empty())
        .map(|column| {
            let rendered = if let Some(cmd) = context {
                resolve_col_syntax(column, cmd, generator)
            } else {
                generator.quote_identifier(column)
            };
            format!("coalesce({}, '')", rendered)
        })
        .collect();

    if parts.is_empty() {
        None
    } else {
        Some(parts.join(" || ' ' || "))
    }
}

fn condition_left_sql(expr: &Expr, generator: &dyn SqlGenerator, context: Option<&Qail>) -> String {
    match expr {
        Expr::Named(name) | Expr::Aliased { name, .. } => {
            if name.starts_with('{') && name.ends_with('}') {
                name[1..name.len() - 1].to_string()
            } else if let Some(cmd) = context {
                resolve_col_syntax(name, cmd, generator)
            } else {
                generator.quote_identifier(name)
            }
        }
        Expr::JsonAccess {
            column,
            path_segments,
            ..
        } => render_json_access(column, path_segments, generator),
        Expr::Literal(value) => condition_value_sql_with_context(value, generator, context),
        Expr::Aggregate { .. } => crate::transpiler::aggregate::aggregate_call_sql(
            expr,
            &|col| match context {
                Some(cmd) => resolve_col_syntax(col, cmd, generator),
                None => generator.quote_identifier(col),
            },
            &|arg| condition_left_sql(arg, generator, context),
            &|conditions| {
                conditions
                    .iter()
                    .map(|condition| condition.to_sql(generator, context))
                    .collect::<Vec<_>>()
                    .join(" AND ")
            },
        ),
        Expr::Case {
            when_clauses,
            else_value,
            ..
        } => {
            let mut sql = String::from("CASE");
            for (condition, value) in when_clauses {
                sql.push_str(&format!(
                    " WHEN {} THEN {}",
                    condition.to_sql(generator, context),
                    condition_left_sql(value, generator, context)
                ));
            }
            if let Some(value) = else_value {
                sql.push_str(&format!(
                    " ELSE {}",
                    condition_left_sql(value, generator, context)
                ));
            }
            sql.push_str(" END");
            sql
        }
        Expr::Binary {
            left, op, right, ..
        } => {
            let left = condition_left_sql(left, generator, context);
            let right = condition_left_sql(right, generator, context);
            match op {
                op if op.is_postfix() => format!("({left} {op})"),
                _ => op.infix_sql(&left, &right),
            }
        }
        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, |arg| {
                condition_left_sql(arg, generator, context)
            }) {
                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 = condition_left_sql(expr, generator, context);
                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::Cast {
            expr, target_type, ..
        } => {
            let Some(target_type) = checked_sql_type_fragment(target_type) else {
                return "/* ERROR: Invalid cast target type */".to_string();
            };
            format!(
                "{}::{}",
                condition_left_sql(expr, generator, context),
                target_type
            )
        }
        Expr::Collate {
            expr, collation, ..
        } => format!(
            "{} COLLATE {}",
            condition_left_sql(expr, generator, context),
            render_qualified_identifier(collation, generator)
        ),
        Expr::FieldAccess { expr, field, .. } => format!(
            "({}).{}",
            condition_left_sql(expr, generator, context),
            render_qualified_identifier(field, generator)
        ),
        Expr::ArrayConstructor { elements, .. } => {
            let elements = elements
                .iter()
                .map(|element| condition_left_sql(element, generator, context))
                .collect::<Vec<_>>()
                .join(", ");
            format!("ARRAY[{elements}]")
        }
        Expr::RowConstructor { elements, .. } => {
            let elements = elements
                .iter()
                .map(|element| condition_left_sql(element, generator, context))
                .collect::<Vec<_>>()
                .join(", ");
            format!("ROW({elements})")
        }
        Expr::Subscript { expr, index, .. } => subscript_sql(
            expr,
            condition_left_sql(expr, generator, context),
            &condition_left_sql(index, generator, context),
        ),
        Expr::ArraySlice {
            expr, lower, upper, ..
        } => subscript_sql(
            expr,
            condition_left_sql(expr, generator, context),
            &slice_bounds_sql(lower, upper, |bound| {
                condition_left_sql(bound, generator, context)
            }),
        ),
        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))
            }
        }
        _ => "/* ERROR: Invalid condition expression */".to_string(),
    }
}

/// RETURNING for INSERT, UPDATE, DELETE and MERGE previews, matching the
/// native encoder: `None` and an empty list render nothing, `[Star]` is `*`,
/// output aliases are kept. `render` writes an expression without its alias.
pub(crate) fn returning_clause_sql(
    cmd: &Qail,
    generator: &dyn SqlGenerator,
    render: impl Fn(&Expr) -> String,
) -> String {
    let Some(exprs) = cmd.returning.as_ref().filter(|exprs| !exprs.is_empty()) else {
        return String::new();
    };
    let items = exprs
        .iter()
        .map(|expr| match expr {
            Expr::Star => "*".to_string(),
            Expr::Aliased { name, alias } => format!(
                "{} AS {}",
                render(&Expr::Named(name.clone())),
                generator.quote_identifier(alias)
            ),
            _ => match output_alias(expr) {
                Some(alias) => format!("{} AS {}", render(expr), generator.quote_identifier(alias)),
                None => render(expr),
            },
        })
        .collect::<Vec<_>>()
        .join(", ");
    // PG 18 `RETURNING WITH (OLD AS o, NEW AS n)` precedes the items.
    let aliases = crate::transpiler::dml::returning_aliases_sql(cmd, generator);
    format!(" RETURNING {aliases}{items}")
}

/// Render an output expression without its alias (RETURNING items).
pub(crate) fn output_expr_sql(expr: &Expr, generator: &dyn SqlGenerator) -> String {
    match expr {
        Expr::Named(name) if name == "*" => "*".to_string(),
        Expr::Named(name) => generator.quote_identifier(name),
        _ => condition_left_sql(expr, generator, None),
    }
}

fn output_alias(expr: &Expr) -> Option<&str> {
    match expr {
        Expr::Aliased { alias, .. } => Some(alias),
        Expr::Window { name, .. } if !name.is_empty() => Some(name),
        Expr::Aggregate { alias, .. }
        | Expr::Cast { alias, .. }
        | Expr::Case { alias, .. }
        | Expr::JsonAccess { alias, .. }
        | Expr::FunctionCall { alias, .. }
        | Expr::SpecialFunction { alias, .. }
        | Expr::Binary { alias, .. }
        | Expr::ArrayConstructor { alias, .. }
        | Expr::RowConstructor { alias, .. }
        | Expr::Subscript { alias, .. }
        | Expr::ArraySlice { alias, .. }
        | Expr::Collate { alias, .. }
        | Expr::FieldAccess { alias, .. }
        | Expr::Subquery { alias, .. }
        | Expr::Exists { alias, .. } => alias.as_deref(),
        Expr::Star
        | Expr::Named(_)
        | Expr::Window { .. }
        | Expr::Def { .. }
        | Expr::Mod { .. }
        | Expr::Literal(_)
        | Expr::Default
        | Expr::FunctionArg { .. } => None,
    }
}

pub(crate) fn read_only_subquery_sql(query: &Qail) -> String {
    if let Some(error) = validate_read_only_subquery(query) {
        format!("/* ERROR: {error} */")
    } else {
        query.to_sql()
    }
}

pub(crate) fn validate_read_only_subquery(query: &Qail) -> Option<String> {
    if !matches!(query.action, Action::Get | Action::Cnt | Action::With) {
        return Some(format!(
            "subquery must be read-only SELECT, got {}",
            query.action
        ));
    }

    for column in &query.columns {
        if let Some(error) = validate_read_only_expr(column) {
            return Some(error);
        }
    }
    for expr in &query.distinct_on {
        if let Some(error) = validate_read_only_expr(expr) {
            return Some(error);
        }
    }
    for cage in &query.cages {
        for condition in &cage.conditions {
            if let Some(error) = validate_read_only_condition(condition) {
                return Some(error);
            }
        }
    }
    for condition in &query.having {
        if let Some(error) = validate_read_only_condition(condition) {
            return Some(error);
        }
    }
    for join in &query.joins {
        if let Some(conditions) = &join.on {
            for condition in conditions {
                if let Some(error) = validate_read_only_condition(condition) {
                    return Some(error);
                }
            }
        }
    }
    for cte in &query.ctes {
        if let Some(error) = validate_read_only_subquery(&cte.base_query) {
            return Some(error);
        }
        if let Some(recursive_query) = &cte.recursive_query
            && let Some(error) = validate_read_only_subquery(recursive_query)
        {
            return Some(error);
        }
    }
    for (_, set_query) in &query.set_ops {
        if let Some(error) = validate_read_only_subquery(set_query) {
            return Some(error);
        }
    }
    if let Some(source_query) = &query.source_query
        && let Some(error) = validate_read_only_subquery(source_query)
    {
        return Some(error);
    }
    if let Some(returning) = &query.returning {
        for expr in returning {
            if let Some(error) = validate_read_only_expr(expr) {
                return Some(error);
            }
        }
    }

    None
}

fn validate_read_only_condition(condition: &Condition) -> Option<String> {
    validate_read_only_expr(&condition.left).or_else(|| validate_read_only_value(&condition.value))
}

fn validate_read_only_value(value: &Value) -> Option<String> {
    match value {
        Value::Subquery(query) => validate_read_only_subquery(query),
        Value::Expr(expr) => validate_read_only_expr(expr),
        Value::Array(values) => values.iter().find_map(validate_read_only_value),
        _ => None,
    }
}

fn validate_read_only_expr(expr: &Expr) -> Option<String> {
    match expr {
        Expr::Case {
            when_clauses,
            else_value,
            ..
        } => {
            for (condition, value) in when_clauses {
                if let Some(error) = validate_read_only_condition(condition)
                    .or_else(|| validate_read_only_expr(value))
                {
                    return Some(error);
                }
            }
            else_value
                .as_ref()
                .and_then(|expr| validate_read_only_expr(expr))
        }
        Expr::Binary { left, right, .. } => {
            validate_read_only_expr(left).or_else(|| validate_read_only_expr(right))
        }
        Expr::FunctionCall { args, .. } => args.iter().find_map(validate_read_only_expr),
        Expr::SpecialFunction { args, .. } => args
            .iter()
            .find_map(|(_, expr)| validate_read_only_expr(expr)),
        Expr::Cast { expr, .. }
        | Expr::FieldAccess { expr, .. }
        | Expr::Collate { expr, .. }
        | Expr::FunctionArg { value: expr, .. } => validate_read_only_expr(expr),
        Expr::ArrayConstructor { elements, .. } | Expr::RowConstructor { elements, .. } => {
            elements.iter().find_map(validate_read_only_expr)
        }
        Expr::Subscript { expr, index, .. } => {
            validate_read_only_expr(expr).or_else(|| validate_read_only_expr(index))
        }
        Expr::ArraySlice {
            expr, lower, upper, ..
        } => validate_read_only_expr(expr).or_else(|| {
            [lower, upper]
                .into_iter()
                .flatten()
                .find_map(|bound| validate_read_only_expr(bound))
        }),
        Expr::Subquery { query, .. } | Expr::Exists { query, .. } => {
            validate_read_only_subquery(query)
        }
        _ => 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 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
}

fn fuzzy_pattern_sql(value: &Value, generator: &dyn SqlGenerator) -> String {
    match value {
        Value::String(s) => format!("'%{}%'", escape_sql_string_literal(s)),
        Value::Param(n) => {
            let p = generator.placeholder(*n);
            generator.string_concat(&["'%'", &p, "'%'"])
        }
        Value::NamedParam(name) => {
            let p = render_named_param(name);
            generator.string_concat(&["'%'", &p, "'%'"])
        }
        // Inside quotes the error marker would become pattern text.
        v @ Value::Float(n) if !n.is_finite() => v.to_string(),
        v => format!("'%{}%'", escape_sql_string_literal(&v.to_string())),
    }
}

fn json_path_arg(condition: &Condition, generator: &dyn SqlGenerator) -> String {
    match &condition.value {
        Value::String(path) => path.clone(),
        Value::Param(n) => generator.placeholder(*n),
        Value::NamedParam(name) => render_named_param(name),
        _ => condition.to_value_sql(generator),
    }
}

fn condition_value_sql(value: &Value, generator: &dyn SqlGenerator) -> String {
    condition_value_sql_with_context(value, generator, None)
}

fn condition_value_sql_with_context(
    value: &Value,
    generator: &dyn SqlGenerator,
    context: Option<&Qail>,
) -> String {
    match value {
        Value::Param(n) => generator.placeholder(*n),
        Value::String(s) => format!("'{}'", escape_sql_string_literal(s)),
        Value::Bool(b) => generator.bool_literal(*b),
        Value::Subquery(cmd) => format!("({})", read_only_subquery_sql(cmd)),
        Value::Column(col) => {
            if let Some(cmd) = context {
                resolve_col_syntax(col, cmd, generator)
            } else {
                generator.quote_identifier(col)
            }
        }
        Value::Array(values) => {
            let values = values
                .iter()
                .map(|value| condition_value_sql_with_context(value, generator, context))
                .collect::<Vec<_>>()
                .join(", ");
            format!("({values})")
        }
        Value::Expr(expr) => condition_left_sql(expr, generator, context),
        Value::NamedParam(name) => render_named_param(name),
        Value::Function(function) => render_raw_function_value(function),
        v => v.to_string(),
    }
}

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 in_condition_sql(
    col: &str,
    op: Operator,
    value: &Value,
    generator: &dyn SqlGenerator,
    context: Option<&Qail>,
) -> String {
    match value {
        Value::Array(values) if !values.is_empty() => {
            let values = values
                .iter()
                .map(|value| condition_value_sql_with_context(value, generator, context))
                .collect::<Vec<_>>()
                .join(", ");
            format!("{col} {} ({values})", op.sql_symbol())
        }
        Value::Subquery(_) => {
            format!(
                "{col} {} {}",
                op.sql_symbol(),
                condition_value_sql_with_context(value, generator, context)
            )
        }
        Value::Param(_) | Value::NamedParam(_) if op == Operator::In => {
            generator.in_array(col, &condition_value_sql(value, generator))
        }
        Value::Param(_) | Value::NamedParam(_) => {
            generator.not_in_array(col, &condition_value_sql(value, generator))
        }
        _ => invalid_in_condition_sql(),
    }
}

fn invalid_exists_condition_sql() -> String {
    "FALSE /* ERROR: EXISTS condition requires subquery value */".to_string()
}

fn invalid_in_condition_sql() -> String {
    "FALSE /* ERROR: IN condition requires a non-empty array, subquery, or array parameter */"
        .to_string()
}

fn invalid_between_condition_sql() -> String {
    "FALSE /* ERROR: BETWEEN condition requires exactly two array values */".to_string()
}

fn invalid_array_membership_sql(op: Operator) -> String {
    if op == Operator::ArrayElemContainedInText {
        "FALSE /* ERROR: ArrayElemContainedInText requires is_array_unnest */".to_string()
    } else {
        format!(
            "FALSE /* ERROR: is_array_unnest supports comparisons, Fuzzy and \
             ArrayElemContainedInText, got {op:?} */"
        )
    }
}

/// Trait for converting AST conditions to SQL strings.
pub trait ConditionToSql {
    /// Render this condition as a SQL string.
    fn to_sql(&self, generator: &dyn SqlGenerator, context: Option<&Qail>) -> String;
    /// Render the right-hand value of this condition as a SQL string.
    fn to_value_sql(&self, generator: &dyn SqlGenerator) -> String;

    /// Convert condition to SQL with parameterized values.
    fn to_sql_parameterized(
        &self,
        generator: &dyn SqlGenerator,
        context: Option<&Qail>,
        params: &mut ParamContext,
    ) -> String;
}

impl ConditionToSql for Condition {
    /// Convert condition to SQL string.
    fn to_sql(&self, generator: &dyn SqlGenerator, context: Option<&Qail>) -> String {
        let col = condition_left_sql(&self.left, generator, context);
        let value_sql = || condition_value_sql_with_context(&self.value, generator, context);

        if self.is_array_unnest {
            let inner_condition = match self.op {
                Operator::Eq => format!("_el = {}", value_sql()),
                Operator::Ne => format!("_el != {}", value_sql()),
                Operator::Gt => format!("_el > {}", value_sql()),
                Operator::Gte => format!("_el >= {}", value_sql()),
                Operator::Lt => format!("_el < {}", value_sql()),
                Operator::Lte => format!("_el <= {}", value_sql()),
                Operator::Fuzzy => {
                    let val = fuzzy_pattern_sql(&self.value, generator);
                    format!("_el {} {}", generator.fuzzy_operator(), val)
                }
                Operator::ArrayElemContainedInText => {
                    format!("LOWER({}) LIKE '%' || LOWER(_el) || '%'", value_sql())
                }
                _ => return invalid_array_membership_sql(self.op),
            };
            return format!(
                "EXISTS (SELECT 1 FROM unnest({}) _el WHERE {})",
                col, inner_condition
            );
        }

        // Normal conditions
        // Simple binary operators use sql_symbol() for unified handling
        if self.op.is_simple_binary() {
            return format!("{} {} {}", col, self.op.sql_symbol(), value_sql());
        }

        // Special operators that need custom handling
        match self.op {
            Operator::Fuzzy => {
                let val = fuzzy_pattern_sql(&self.value, generator);
                format!("{} {} {}", col, generator.fuzzy_operator(), val)
            }
            Operator::TextSearch => {
                let vector = resolve_text_search_vector(&self.left, generator, context)
                    .unwrap_or_else(|| col.clone());
                format!(
                    "to_tsvector('english', {}) @@ websearch_to_tsquery('english', {})",
                    vector,
                    value_sql()
                )
            }
            Operator::In | Operator::NotIn => {
                in_condition_sql(&col, self.op, &self.value, generator, context)
            }
            op if op.is_postfix() => format!("{} {}", col, op.sql_symbol()),
            Operator::ArrayElemContainedInText => invalid_array_membership_sql(self.op),
            Operator::Contains => generator.json_contains(&col, &value_sql()),
            Operator::KeyExists => generator.json_key_exists(&col, &value_sql()),
            // Postgres 17+ SQL/JSON standard functions
            Operator::JsonExists => {
                let path = json_path_arg(self, generator);
                generator.json_exists(&col, &path)
            }
            Operator::JsonQuery => {
                let path = json_path_arg(self, generator);
                format!("{} IS NOT NULL", generator.json_query(&col, &path))
            }
            Operator::JsonValue => {
                let path = json_path_arg(self, generator);
                format!("{} IS NOT NULL", generator.json_value(&col, &path))
            }
            op if op.is_range() => {
                // Value is Array with 2 elements [min, max]
                if let Value::Array(vals) = &self.value
                    && vals.len() == 2
                {
                    return format!(
                        "{} {} {} AND {}",
                        col,
                        op.sql_symbol(),
                        condition_value_sql_with_context(&vals[0], generator, context),
                        condition_value_sql_with_context(&vals[1], generator, context)
                    );
                }
                invalid_between_condition_sql()
            }
            Operator::Exists => {
                // EXISTS takes subquery, col is ignored
                if let Value::Subquery(cmd) = &self.value {
                    let subquery_sql = read_only_subquery_sql(cmd);
                    format!("EXISTS ({})", subquery_sql)
                } else {
                    invalid_exists_condition_sql()
                }
            }
            Operator::NotExists => {
                if let Value::Subquery(cmd) = &self.value {
                    let subquery_sql = read_only_subquery_sql(cmd);
                    format!("NOT EXISTS ({})", subquery_sql)
                } else {
                    invalid_exists_condition_sql()
                }
            }
            // Simple binary operators are handled above by is_simple_binary()
            _ => self.op.infix_sql(&col, &value_sql()),
        }
    }

    fn to_value_sql(&self, generator: &dyn SqlGenerator) -> String {
        condition_value_sql(&self.value, generator)
    }

    fn to_sql_parameterized(
        &self,
        generator: &dyn SqlGenerator,
        context: Option<&Qail>,
        params: &mut ParamContext,
    ) -> String {
        let col = condition_left_sql(&self.left, generator, context);

        // Helper to convert value to placeholder
        let value_placeholder = |v: &Value, p: &mut ParamContext| -> String {
            match v {
                Value::Param(n) => generator.placeholder(*n), // Already a placeholder
                Value::NamedParam(name) => p.add_named_param(name.clone(), generator),
                Value::Column(_) => condition_value_sql_with_context(v, generator, context),
                Value::Expr(expr) => condition_left_sql(expr, generator, context),
                Value::Subquery(cmd) => format!("({})", read_only_subquery_sql(cmd)),
                Value::Null => "NULL".to_string(),
                other => p.add_param(other.clone(), generator),
            }
        };

        if self.is_array_unnest {
            let inner_condition = match self.op {
                Operator::Eq => format!("_el = {}", value_placeholder(&self.value, params)),
                Operator::Ne => format!("_el != {}", value_placeholder(&self.value, params)),
                Operator::Gt => format!("_el > {}", value_placeholder(&self.value, params)),
                Operator::Gte => format!("_el >= {}", value_placeholder(&self.value, params)),
                Operator::Lt => format!("_el < {}", value_placeholder(&self.value, params)),
                Operator::Lte => format!("_el <= {}", value_placeholder(&self.value, params)),
                Operator::Fuzzy => {
                    let val = generator.string_concat(&[
                        "'%'",
                        &value_placeholder(&self.value, params),
                        "'%'",
                    ]);
                    format!("_el {} {}", generator.fuzzy_operator(), val)
                }
                Operator::ArrayElemContainedInText => format!(
                    "LOWER({}) LIKE '%' || LOWER(_el) || '%'",
                    value_placeholder(&self.value, params)
                ),
                _ => return invalid_array_membership_sql(self.op),
            };

            return format!(
                "EXISTS (SELECT 1 FROM unnest({}) _el WHERE {})",
                col, inner_condition
            );
        }

        match self.op {
            Operator::Eq => {
                // Raw conditions ({...}, op=Eq, value=Null) are now handled at col resolution
                if matches!(self.value, Value::Null)
                    && let Expr::Named(name) = &self.left
                    && name.starts_with('{')
                    && name.ends_with('}')
                {
                    return col; // col already contains raw SQL content
                }
                format!("{} = {}", col, value_placeholder(&self.value, params))
            }
            Operator::Fuzzy => {
                // For LIKE, we need to wrap in wildcards
                let placeholder = value_placeholder(&self.value, params);
                let pattern = generator.string_concat(&["'%'", &placeholder, "'%'"]);
                format!("{} {} {}", col, generator.fuzzy_operator(), pattern)
            }
            Operator::TextSearch => {
                let vector = resolve_text_search_vector(&self.left, generator, context)
                    .unwrap_or_else(|| col.clone());
                format!(
                    "to_tsvector('english', {}) @@ websearch_to_tsquery('english', {})",
                    vector,
                    value_placeholder(&self.value, params)
                )
            }
            op if op.is_postfix() => format!("{} {}", col, op.sql_symbol()),
            Operator::ArrayElemContainedInText => invalid_array_membership_sql(self.op),
            Operator::In | Operator::NotIn => match &self.value {
                Value::Array(values) if !values.is_empty() => {
                    let value = value_placeholder(&self.value, params);
                    if self.op == Operator::In {
                        generator.in_array(&col, &value)
                    } else {
                        generator.not_in_array(&col, &value)
                    }
                }
                Value::Subquery(cmd) => format!(
                    "{} {} ({})",
                    col,
                    self.op.sql_symbol(),
                    read_only_subquery_sql(cmd)
                ),
                Value::Param(_) | Value::NamedParam(_) => {
                    let value = value_placeholder(&self.value, params);
                    if self.op == Operator::In {
                        generator.in_array(&col, &value)
                    } else {
                        generator.not_in_array(&col, &value)
                    }
                }
                _ => invalid_in_condition_sql(),
            },
            Operator::Contains => {
                generator.json_contains(&col, &value_placeholder(&self.value, params))
            }
            Operator::KeyExists => {
                generator.json_key_exists(&col, &value_placeholder(&self.value, params))
            }
            Operator::JsonExists => {
                let path = value_placeholder(&self.value, params);
                generator.json_exists(&col, &path)
            }
            Operator::JsonQuery => {
                let path = value_placeholder(&self.value, params);
                format!("{} IS NOT NULL", generator.json_query(&col, &path))
            }
            Operator::JsonValue => {
                let path = value_placeholder(&self.value, params);
                format!("{} IS NOT NULL", generator.json_value(&col, &path))
            }
            op if op.is_range() => {
                if let Value::Array(vals) = &self.value
                    && vals.len() == 2
                {
                    let low = value_placeholder(&vals[0], params);
                    let high = value_placeholder(&vals[1], params);
                    return format!("{} {} {} AND {}", col, op.sql_symbol(), low, high);
                }
                invalid_between_condition_sql()
            }
            Operator::Exists => {
                if let Value::Subquery(cmd) = &self.value {
                    let subquery_sql = read_only_subquery_sql(cmd);
                    format!("EXISTS ({})", subquery_sql)
                } else {
                    invalid_exists_condition_sql()
                }
            }
            Operator::NotExists => {
                if let Value::Subquery(cmd) = &self.value {
                    let subquery_sql = read_only_subquery_sql(cmd);
                    format!("NOT EXISTS ({})", subquery_sql)
                } else {
                    invalid_exists_condition_sql()
                }
            }
            // Simple operators (Ne, Gt, Gte, Lt, Lte, Like, NotLike, ILike, NotILike) use sql_symbol()
            _ => self
                .op
                .infix_sql(&col, &value_placeholder(&self.value, params)),
        }
    }
}