icydb-core 0.212.3

//! Module: db::sql::parser::tests
//! Covers SQL parser behavior, error classification, and syntax edge cases.
//! Does not own: cross-module orchestration outside this module.
//! Boundary: exposes this module API while keeping implementation details internal.

use super::{
    SqlAggregateCall, SqlAggregateKind, SqlAlterColumnAction, SqlAlterTableAddColumnStatement,
    SqlAlterTableAlterColumnStatement, SqlAlterTableDropColumnStatement,
    SqlAlterTableRenameColumnStatement, SqlAssignment, SqlCaseArm,
    SqlCreateIndexExpressionFunction, SqlCreateIndexExpressionKey, SqlCreateIndexKeyItem,
    SqlCreateIndexStatement, SqlCreateIndexUniqueness, SqlDdlSchemaVersionContract,
    SqlDdlStatement, SqlDeleteStatement, SqlDescribeStatement, SqlDropIndexStatement, SqlExpr,
    SqlExprBinaryOp, SqlInsertSource, SqlInsertStatement, SqlIntegrityStatement, SqlOrderDirection,
    SqlOrderTerm, SqlParseError, SqlProjection, SqlReturningProjection, SqlScalarFunction,
    SqlSelectItem, SqlSelectStatement, SqlShowColumnsStatement, SqlShowConstraintsStatement,
    SqlShowEntitiesStatement, SqlShowIndexesStatement, SqlShowMemoryStatement,
    SqlShowStoresStatement, SqlStatement, SqlUpdateStatement, SqlWriteValue, parse_integrity_sql,
    parse_sql,
};
#[cfg(feature = "sql-explain")]
use super::{SqlExplainMode, SqlExplainStatement, SqlExplainTarget};
use crate::{
    db::predicate::{CoercionId, CompareFieldsPredicate, CompareOp, ComparePredicate, Predicate},
    db::sql_shared::{
        MAX_SQL_EXPR_DEPTH, MAX_SQL_INPUT_BYTES, MAX_SQL_TOKENS, SqlClauseOrderRule,
        SqlSyntaxErrorKind,
    },
    value::Value,
};
use icydb_diagnostic_code::SqlFeatureCode;

fn sql_write_literal(value: Value) -> SqlWriteValue {
    SqlWriteValue::Literal(value)
}

fn ddl_field_paths(paths: &[&str]) -> Vec<SqlCreateIndexKeyItem> {
    paths
        .iter()
        .map(|path| SqlCreateIndexKeyItem::FieldPath((*path).to_string()))
        .collect()
}

fn ddl_expression_key(
    function: SqlCreateIndexExpressionFunction,
    field_path: &str,
) -> SqlCreateIndexKeyItem {
    SqlCreateIndexKeyItem::Expression(SqlCreateIndexExpressionKey {
        function,
        field_path: field_path.to_string(),
    })
}

macro_rules! option_sql_pred {
    ($predicate:expr) => {
        Some(sql_expr_from_runtime_predicate($predicate))
    };
}

fn sql_order_expr(term: &str) -> SqlExpr {
    let sql = format!("SELECT id FROM ParserOrderEntity ORDER BY {term}");
    let SqlStatement::Select(statement) =
        parse_sql(&sql).expect("ORDER BY term helper SQL should parse")
    else {
        unreachable!("ORDER BY term helper should always produce one SELECT");
    };

    statement
        .order_by
        .into_iter()
        .next()
        .expect("ORDER BY term helper SQL should carry one ORDER BY term")
        .field
}

fn sql_scalar_function_expr(function: SqlScalarFunction, args: Vec<SqlExpr>) -> SqlExpr {
    SqlExpr::FunctionCall { function, args }
}

fn sql_scalar_function_field_expr(function: SqlScalarFunction, field: &str) -> SqlExpr {
    sql_scalar_function_expr(function, vec![SqlExpr::Field(field.to_string())])
}

fn sql_scalar_function_field_item(function: SqlScalarFunction, field: &str) -> SqlSelectItem {
    SqlSelectItem::Expr(sql_scalar_function_field_expr(function, field))
}

fn sql_binary_expr(left: SqlExpr, op: SqlExprBinaryOp, right: SqlExpr) -> SqlExpr {
    SqlExpr::Binary {
        op,
        left: Box::new(left),
        right: Box::new(right),
    }
}

fn sql_like_expr(expr: SqlExpr, pattern: &str, negated: bool, casefold: bool) -> SqlExpr {
    SqlExpr::Like {
        expr: Box::new(expr),
        pattern: pattern.to_string(),
        negated,
        casefold,
    }
}

fn sql_round_item(input: SqlExpr, scale: Value) -> SqlSelectItem {
    SqlSelectItem::Expr(SqlExpr::FunctionCall {
        function: SqlScalarFunction::Round,
        args: vec![input, SqlExpr::Literal(scale)],
    })
}

fn sql_expr_from_runtime_predicate(predicate: Predicate) -> SqlExpr {
    match predicate {
        Predicate::True => SqlExpr::Literal(Value::Bool(true)),
        Predicate::False => SqlExpr::Literal(Value::Bool(false)),
        Predicate::And(children) => fold_predicate_children(children, SqlExprBinaryOp::And),
        Predicate::Or(children) => fold_predicate_children(children, SqlExprBinaryOp::Or),
        Predicate::Not(inner) => SqlExpr::Unary {
            op: super::SqlExprUnaryOp::Not,
            expr: Box::new(sql_expr_from_runtime_predicate(*inner)),
        },
        Predicate::Compare(compare) => sql_expr_from_compare(compare),
        Predicate::CompareFields(compare) => SqlExpr::Binary {
            op: sql_binary_from_compare(compare.op()),
            left: Box::new(SqlExpr::Field(compare.left_field().to_string())),
            right: Box::new(SqlExpr::Field(compare.right_field().to_string())),
        },
        Predicate::IsNull { field } => SqlExpr::NullTest {
            expr: Box::new(SqlExpr::Field(field)),
            negated: false,
        },
        Predicate::IsNotNull { field } => SqlExpr::NullTest {
            expr: Box::new(SqlExpr::Field(field)),
            negated: true,
        },
        Predicate::IsMissing { field } => SqlExpr::FunctionCall {
            function: SqlScalarFunction::Contains,
            args: vec![SqlExpr::Field(field), SqlExpr::Literal(Value::Null)],
        },
        Predicate::IsEmpty { field } => SqlExpr::FunctionCall {
            function: SqlScalarFunction::Length,
            args: vec![SqlExpr::Field(field)],
        },
        Predicate::IsNotEmpty { field } => SqlExpr::Unary {
            op: super::SqlExprUnaryOp::Not,
            expr: Box::new(SqlExpr::FunctionCall {
                function: SqlScalarFunction::Length,
                args: vec![SqlExpr::Field(field)],
            }),
        },
        Predicate::TextContains { field, value } => SqlExpr::FunctionCall {
            function: SqlScalarFunction::Contains,
            args: vec![SqlExpr::Field(field), SqlExpr::Literal(value)],
        },
        Predicate::TextContainsCi { field, value } => SqlExpr::FunctionCall {
            function: SqlScalarFunction::Contains,
            args: vec![
                sql_scalar_function_field_expr(SqlScalarFunction::Lower, &field),
                SqlExpr::Literal(value),
            ],
        },
    }
}

#[test]
fn parse_sql_preserves_placeholder_slot_order_across_where_and_having() {
    let SqlStatement::Select(statement) = parse_sql(
        "SELECT name, COUNT(*) \
         FROM ParserEntity \
         WHERE age > ? AND name = ? \
         GROUP BY name \
         HAVING COUNT(*) > ?",
    )
    .expect("placeholder SQL should parse") else {
        panic!("placeholder SQL should produce one SELECT statement");
    };

    let Some(SqlExpr::Binary { left, right, .. }) = statement.predicate.as_ref() else {
        panic!("placeholder SQL should preserve one compound WHERE predicate");
    };
    let SqlExpr::Binary {
        right: where_first_param,
        ..
    } = left.as_ref()
    else {
        panic!("left WHERE predicate child should stay one compare expression");
    };
    let SqlExpr::Binary {
        right: where_second_param,
        ..
    } = right.as_ref()
    else {
        panic!("right WHERE predicate child should stay one compare expression");
    };
    let [
        SqlExpr::Binary {
            right: having_param,
            ..
        },
    ] = statement.having.as_slice()
    else {
        panic!("HAVING placeholder should stay one compare expression");
    };

    std::assert_matches!(where_first_param.as_ref(), SqlExpr::Param { index: 0 });
    std::assert_matches!(where_second_param.as_ref(), SqlExpr::Param { index: 1 });
    std::assert_matches!(having_param.as_ref(), SqlExpr::Param { index: 2 });
}

fn sql_expr_from_compare(compare: ComparePredicate) -> SqlExpr {
    match compare.op() {
        CompareOp::In | CompareOp::NotIn => {
            let Value::List(values) = compare.value().clone() else {
                panic!("IN/NOT IN compare expects list literal in parser tests");
            };

            SqlExpr::Membership {
                expr: Box::new(SqlExpr::Field(compare.field().to_string())),
                values,
                negated: compare.op() == CompareOp::NotIn,
            }
        }
        CompareOp::StartsWith | CompareOp::EndsWith | CompareOp::Contains => {
            SqlExpr::FunctionCall {
                function: match compare.op() {
                    CompareOp::StartsWith => SqlScalarFunction::StartsWith,
                    CompareOp::EndsWith => SqlScalarFunction::EndsWith,
                    CompareOp::Contains => SqlScalarFunction::Contains,
                    _ => unreachable!(),
                },
                args: vec![
                    if compare.coercion().id() == CoercionId::TextCasefold {
                        sql_scalar_function_field_expr(SqlScalarFunction::Lower, compare.field())
                    } else {
                        SqlExpr::Field(compare.field().to_string())
                    },
                    SqlExpr::Literal(compare.value().clone()),
                ],
            }
        }
        op => SqlExpr::Binary {
            op: sql_binary_from_compare(op),
            left: Box::new(match compare.coercion().id() {
                CoercionId::TextCasefold => {
                    sql_scalar_function_field_expr(SqlScalarFunction::Lower, compare.field())
                }
                _ => SqlExpr::Field(compare.field().to_string()),
            }),
            right: Box::new(SqlExpr::Literal(compare.value().clone())),
        },
    }
}

fn fold_predicate_children(children: Vec<Predicate>, op: SqlExprBinaryOp) -> SqlExpr {
    fold_exprs(
        children
            .into_iter()
            .map(sql_expr_from_runtime_predicate)
            .collect(),
        op,
    )
}

fn fold_exprs(mut exprs: Vec<SqlExpr>, op: SqlExprBinaryOp) -> SqlExpr {
    let first = exprs.remove(0);
    exprs
        .into_iter()
        .fold(first, |left, right| SqlExpr::Binary {
            op,
            left: Box::new(left),
            right: Box::new(right),
        })
}

const fn sql_binary_from_compare(op: CompareOp) -> SqlExprBinaryOp {
    match op {
        CompareOp::Eq
        | CompareOp::In
        | CompareOp::NotIn
        | CompareOp::Contains
        | CompareOp::StartsWith
        | CompareOp::EndsWith => SqlExprBinaryOp::Eq,
        CompareOp::Ne => SqlExprBinaryOp::Ne,
        CompareOp::Lt => SqlExprBinaryOp::Lt,
        CompareOp::Lte => SqlExprBinaryOp::Lte,
        CompareOp::Gt => SqlExprBinaryOp::Gt,
        CompareOp::Gte => SqlExprBinaryOp::Gte,
    }
}

#[test]
fn parse_select_statement_with_predicate_order_and_window() {
    let sql = "  SeLeCt DISTINCT name, COUNT(*) FROM users \
               WHERE age >= 21 AND active = TRUE \
               ORDER BY age DESC, name ASC LIMIT 10 OFFSET 5;  ";
    let statement = parse_sql(sql).expect("select statement should parse");

    assert_eq!(
        statement,
        SqlStatement::Select(SqlSelectStatement {
            entity: "users".to_string(),
            table_alias: None,
            projection: SqlProjection::Items(vec![
                SqlSelectItem::Field("name".to_string()),
                SqlSelectItem::Aggregate(SqlAggregateCall {
                    kind: SqlAggregateKind::Count,
                    input: None,
                    filter_expr: None,
                    distinct: false,
                }),
            ]),
            projection_aliases: vec![None, None],
            predicate: option_sql_pred!(Predicate::And(vec![
                Predicate::Compare(ComparePredicate::with_coercion(
                    "age",
                    CompareOp::Gte,
                    Value::Int64(21),
                    CoercionId::NumericWiden,
                )),
                Predicate::Compare(ComparePredicate::with_coercion(
                    "active",
                    CompareOp::Eq,
                    Value::Bool(true),
                    CoercionId::Strict,
                )),
            ])),
            distinct: true,
            group_by: vec![],
            having: vec![],
            order_by: vec![
                SqlOrderTerm {
                    field: sql_order_expr("age"),
                    direction: SqlOrderDirection::Desc,
                },
                SqlOrderTerm {
                    field: sql_order_expr("name"),
                    direction: SqlOrderDirection::Asc,
                },
            ],
            limit: Some(10),
            offset: Some(5),
        }),
    );
}

#[test]
fn parse_select_statement_with_trim_ltrim_rtrim_lower_upper_and_length_projection_items() {
    let statement = parse_sql(
        "SELECT TRIM(name), LTRIM(name), RTRIM(name), LOWER(name), UPPER(name), LENGTH(name), age FROM users",
    )
    .expect("scalar-function projection select statement should parse");

    assert_eq!(
        statement,
        SqlStatement::Select(SqlSelectStatement {
            entity: "users".to_string(),
            table_alias: None,
            projection: SqlProjection::Items(vec![
                sql_scalar_function_field_item(SqlScalarFunction::Trim, "name"),
                sql_scalar_function_field_item(SqlScalarFunction::Ltrim, "name"),
                sql_scalar_function_field_item(SqlScalarFunction::Rtrim, "name"),
                sql_scalar_function_field_item(SqlScalarFunction::Lower, "name"),
                sql_scalar_function_field_item(SqlScalarFunction::Upper, "name"),
                sql_scalar_function_field_item(SqlScalarFunction::Length, "name"),
                SqlSelectItem::Field("age".to_string()),
            ]),
            projection_aliases: vec![None, None, None, None, None, None, None],
            predicate: None,
            distinct: false,
            group_by: vec![],
            having: vec![],
            order_by: vec![],
            limit: None,
            offset: None,
        }),
    );
}

#[test]
fn parse_select_statement_with_abs_ceil_ceiling_and_floor_projection_items() {
    let statement =
        parse_sql("SELECT ABS(age), CEIL(age), CEILING(age), FLOOR(age), name FROM users")
            .expect("numeric scalar-function projection select statement should parse");

    assert_eq!(
        statement,
        SqlStatement::Select(SqlSelectStatement {
            entity: "users".to_string(),
            table_alias: None,
            projection: SqlProjection::Items(vec![
                sql_scalar_function_field_item(SqlScalarFunction::Abs, "age"),
                sql_scalar_function_field_item(SqlScalarFunction::Ceiling, "age"),
                sql_scalar_function_field_item(SqlScalarFunction::Ceiling, "age"),
                sql_scalar_function_field_item(SqlScalarFunction::Floor, "age"),
                SqlSelectItem::Field("name".to_string()),
            ]),
            projection_aliases: vec![None, None, None, None, None],
            predicate: None,
            distinct: false,
            group_by: vec![],
            having: vec![],
            order_by: vec![],
            limit: None,
            offset: None,
        }),
    );
}

#[test]
fn parse_select_statement_with_unary_numeric_expression_projection_items() {
    let statement = parse_sql(
        "SELECT ABS(age - 30), CEIL(age / 10), CEILING(age / 10), FLOOR(age / 10) FROM users",
    )
    .expect("numeric scalar-function expression projection select statement should parse");

    assert_eq!(
        statement,
        SqlStatement::Select(SqlSelectStatement {
            entity: "users".to_string(),
            table_alias: None,
            projection: SqlProjection::Items(vec![
                SqlSelectItem::Expr(sql_scalar_function_expr(
                    SqlScalarFunction::Abs,
                    vec![sql_binary_expr(
                        SqlExpr::Field("age".to_string()),
                        SqlExprBinaryOp::Sub,
                        SqlExpr::Literal(Value::Int64(30)),
                    )],
                )),
                SqlSelectItem::Expr(sql_scalar_function_expr(
                    SqlScalarFunction::Ceiling,
                    vec![sql_binary_expr(
                        SqlExpr::Field("age".to_string()),
                        SqlExprBinaryOp::Div,
                        SqlExpr::Literal(Value::Int64(10)),
                    )],
                )),
                SqlSelectItem::Expr(sql_scalar_function_expr(
                    SqlScalarFunction::Ceiling,
                    vec![sql_binary_expr(
                        SqlExpr::Field("age".to_string()),
                        SqlExprBinaryOp::Div,
                        SqlExpr::Literal(Value::Int64(10)),
                    )],
                )),
                SqlSelectItem::Expr(sql_scalar_function_expr(
                    SqlScalarFunction::Floor,
                    vec![sql_binary_expr(
                        SqlExpr::Field("age".to_string()),
                        SqlExprBinaryOp::Div,
                        SqlExpr::Literal(Value::Int64(10)),
                    )],
                )),
            ]),
            projection_aliases: vec![None, None, None, None],
            predicate: None,
            distinct: false,
            group_by: vec![],
            having: vec![],
            order_by: vec![],
            limit: None,
            offset: None,
        }),
    );
}

#[test]
fn parse_select_statement_with_unary_text_expression_projection_items() {
    let statement = parse_sql(
        "SELECT LOWER(COALESCE(name, 'fallback')), \
                UPPER(NULLIF(name, 'guest')), \
                TRIM(COALESCE(name, 'fallback')), \
                LENGTH(TRIM(name)) \
         FROM users",
    )
    .expect("text scalar-function expression projection select statement should parse");

    assert_eq!(
        statement,
        SqlStatement::Select(SqlSelectStatement {
            entity: "users".to_string(),
            table_alias: None,
            projection: SqlProjection::Items(vec![
                SqlSelectItem::Expr(sql_scalar_function_expr(
                    SqlScalarFunction::Lower,
                    vec![sql_scalar_function_expr(
                        SqlScalarFunction::Coalesce,
                        vec![
                            SqlExpr::Field("name".to_string()),
                            SqlExpr::Literal(Value::Text("fallback".to_string())),
                        ],
                    )],
                )),
                SqlSelectItem::Expr(sql_scalar_function_expr(
                    SqlScalarFunction::Upper,
                    vec![sql_scalar_function_expr(
                        SqlScalarFunction::NullIf,
                        vec![
                            SqlExpr::Field("name".to_string()),
                            SqlExpr::Literal(Value::Text("guest".to_string())),
                        ],
                    )],
                )),
                SqlSelectItem::Expr(sql_scalar_function_expr(
                    SqlScalarFunction::Trim,
                    vec![sql_scalar_function_expr(
                        SqlScalarFunction::Coalesce,
                        vec![
                            SqlExpr::Field("name".to_string()),
                            SqlExpr::Literal(Value::Text("fallback".to_string())),
                        ],
                    )],
                )),
                SqlSelectItem::Expr(sql_scalar_function_expr(
                    SqlScalarFunction::Length,
                    vec![sql_scalar_function_expr(
                        SqlScalarFunction::Trim,
                        vec![SqlExpr::Field("name".to_string())],
                    )],
                )),
            ]),
            projection_aliases: vec![None, None, None, None],
            predicate: None,
            distinct: false,
            group_by: vec![],
            having: vec![],
            order_by: vec![],
            limit: None,
            offset: None,
        }),
    );
}

#[test]
fn parse_select_statement_with_coalesce_and_nullif_projection_items() {
    let statement = parse_sql("SELECT COALESCE(NULL, name), NULLIF(age, 20), name FROM users")
        .expect("coalesce/nullif projection select statement should parse");

    assert_eq!(
        statement,
        SqlStatement::Select(SqlSelectStatement {
            entity: "users".to_string(),
            table_alias: None,
            projection: SqlProjection::Items(vec![
                SqlSelectItem::Expr(sql_scalar_function_expr(
                    SqlScalarFunction::Coalesce,
                    vec![
                        SqlExpr::Literal(Value::Null),
                        SqlExpr::Field("name".to_string()),
                    ],
                )),
                SqlSelectItem::Expr(sql_scalar_function_expr(
                    SqlScalarFunction::NullIf,
                    vec![
                        SqlExpr::Field("age".to_string()),
                        SqlExpr::Literal(Value::Int64(20)),
                    ],
                )),
                SqlSelectItem::Field("name".to_string()),
            ]),
            projection_aliases: vec![None, None, None],
            predicate: None,
            distinct: false,
            group_by: vec![],
            having: vec![],
            order_by: vec![],
            limit: None,
            offset: None,
        }),
    );
}

#[test]
fn parse_select_statement_with_scalar_add_projection_item() {
    let statement = parse_sql("SELECT age + 1 FROM users")
        .expect("scalar arithmetic projection select statement should parse");

    assert_eq!(
        statement,
        SqlStatement::Select(SqlSelectStatement {
            entity: "users".to_string(),
            table_alias: None,
            projection: SqlProjection::Items(vec![SqlSelectItem::Expr(sql_binary_expr(
                SqlExpr::Field("age".to_string()),
                SqlExprBinaryOp::Add,
                SqlExpr::Literal(Value::Int64(1)),
            ))]),
            projection_aliases: vec![None],
            predicate: None,
            distinct: false,
            group_by: vec![],
            having: vec![],
            order_by: vec![],
            limit: None,
            offset: None,
        }),
    );
}

#[test]
fn parse_select_statement_with_field_path_projection_expr() {
    let statement = parse_sql("SELECT profile.rank + 1 FROM users")
        .expect("field-path projection expression should parse");

    assert_eq!(
        statement,
        SqlStatement::Select(SqlSelectStatement {
            entity: "users".to_string(),
            table_alias: None,
            projection: SqlProjection::Items(vec![SqlSelectItem::Expr(sql_binary_expr(
                SqlExpr::FieldPath {
                    root: "profile".to_string(),
                    segments: vec!["rank".to_string()],
                },
                SqlExprBinaryOp::Add,
                SqlExpr::Literal(Value::Int64(1)),
            ))]),
            projection_aliases: vec![None],
            predicate: None,
            distinct: false,
            group_by: vec![],
            having: vec![],
            order_by: vec![],
            limit: None,
            offset: None,
        })
    );
}

#[test]
fn parse_select_statement_with_scalar_sub_mul_div_projection_items() {
    for (sql, op, literal, context) in [
        (
            "SELECT age - 1 FROM users",
            SqlExprBinaryOp::Sub,
            Value::Int64(1),
            "subtraction projection",
        ),
        (
            "SELECT age * 2 FROM users",
            SqlExprBinaryOp::Mul,
            Value::Int64(2),
            "multiplication projection",
        ),
        (
            "SELECT age / 2 FROM users",
            SqlExprBinaryOp::Div,
            Value::Int64(2),
            "division projection",
        ),
    ] {
        let statement =
            parse_sql(sql).unwrap_or_else(|err| panic!("{context} should parse: {err:?}"));

        assert_eq!(
            statement,
            SqlStatement::Select(SqlSelectStatement {
                entity: "users".to_string(),
                table_alias: None,
                projection: SqlProjection::Items(vec![SqlSelectItem::Expr(sql_binary_expr(
                    SqlExpr::Field("age".to_string()),
                    op,
                    SqlExpr::Literal(literal),
                ))]),
                projection_aliases: vec![None],
                predicate: None,
                distinct: false,
                group_by: vec![],
                having: vec![],
                order_by: vec![],
                limit: None,
                offset: None,
            }),
            "{context} should lower to one bounded arithmetic projection item",
        );
    }
}

#[test]
fn parse_select_statement_with_scalar_field_to_field_projection_item() {
    let statement = parse_sql("SELECT dexterity + charisma AS total FROM users")
        .expect("field-to-field arithmetic projection select statement should parse");

    assert_eq!(
        statement,
        SqlStatement::Select(SqlSelectStatement {
            entity: "users".to_string(),
            table_alias: None,
            projection: SqlProjection::Items(vec![SqlSelectItem::Expr(sql_binary_expr(
                SqlExpr::Field("dexterity".to_string()),
                SqlExprBinaryOp::Add,
                SqlExpr::Field("charisma".to_string()),
            ))]),
            projection_aliases: vec![Some("total".to_string())],
            predicate: None,
            distinct: false,
            group_by: vec![],
            having: vec![],
            order_by: vec![],
            limit: None,
            offset: None,
        }),
    );
}

#[test]
fn parse_select_statement_with_chained_scalar_projection_item_preserves_precedence() {
    let statement = parse_sql("SELECT age + 1 * 2 FROM users")
        .expect("chained scalar arithmetic projection select statement should parse");

    assert_eq!(
        statement,
        SqlStatement::Select(SqlSelectStatement {
            entity: "users".to_string(),
            table_alias: None,
            projection: SqlProjection::Items(vec![SqlSelectItem::Expr(sql_binary_expr(
                SqlExpr::Field("age".to_string()),
                SqlExprBinaryOp::Add,
                sql_binary_expr(
                    SqlExpr::Literal(Value::Int64(1)),
                    SqlExprBinaryOp::Mul,
                    SqlExpr::Literal(Value::Int64(2)),
                ),
            ))]),
            projection_aliases: vec![None],
            predicate: None,
            distinct: false,
            group_by: vec![],
            having: vec![],
            order_by: vec![],
            limit: None,
            offset: None,
        }),
        "chained scalar arithmetic projection should preserve operator precedence in the parser model",
    );
}

#[test]
fn parse_select_statement_with_parenthesized_round_projection_item() {
    let statement = parse_sql("SELECT ROUND((age + salary) / 2, 2) FROM users")
        .expect("parenthesized ROUND projection select statement should parse");

    assert_eq!(
        statement,
        SqlStatement::Select(SqlSelectStatement {
            entity: "users".to_string(),
            table_alias: None,
            projection: SqlProjection::Items(vec![sql_round_item(
                sql_binary_expr(
                    sql_binary_expr(
                        SqlExpr::Field("age".to_string()),
                        SqlExprBinaryOp::Add,
                        SqlExpr::Field("salary".to_string()),
                    ),
                    SqlExprBinaryOp::Div,
                    SqlExpr::Literal(Value::Int64(2)),
                ),
                Value::Int64(2),
            )]),
            projection_aliases: vec![None],
            predicate: None,
            distinct: false,
            group_by: vec![],
            having: vec![],
            order_by: vec![],
            limit: None,
            offset: None,
        }),
        "parenthesized ROUND projection should preserve nested arithmetic structure in the parser model",
    );
}

#[test]
fn parse_select_statement_with_searched_case_projection_item() {
    let statement =
        parse_sql("SELECT CASE WHEN age >= 21 THEN 'adult' ELSE 'minor' END AS cohort FROM users")
            .expect("searched CASE projection select statement should parse");

    assert_eq!(
        statement,
        SqlStatement::Select(SqlSelectStatement {
            entity: "users".to_string(),
            table_alias: None,
            projection: SqlProjection::Items(vec![SqlSelectItem::Expr(SqlExpr::Case {
                arms: vec![SqlCaseArm {
                    condition: SqlExpr::Binary {
                        op: SqlExprBinaryOp::Gte,
                        left: Box::new(SqlExpr::Field("age".to_string())),
                        right: Box::new(SqlExpr::Literal(Value::Int64(21))),
                    },
                    result: SqlExpr::Literal(Value::Text("adult".to_string())),
                }],
                else_expr: Some(Box::new(SqlExpr::Literal(
                    Value::Text("minor".to_string(),)
                ))),
            })]),
            projection_aliases: vec![Some("cohort".to_string())],
            predicate: None,
            distinct: false,
            group_by: vec![],
            having: vec![],
            order_by: vec![],
            limit: None,
            offset: None,
        }),
        "searched CASE projection should stay on the shared SQL-expression boundary",
    );
}

#[test]
fn parse_select_statement_with_searched_case_is_null_condition_projection_item() {
    let statement = parse_sql(
        "SELECT CASE WHEN guild_rank IS NULL THEN 'unguilded' ELSE guild_rank END AS guild_label FROM users",
    )
    .expect("searched CASE projection with IS NULL condition should parse");

    assert_eq!(
        statement,
        SqlStatement::Select(SqlSelectStatement {
            entity: "users".to_string(),
            table_alias: None,
            projection: SqlProjection::Items(vec![SqlSelectItem::Expr(SqlExpr::Case {
                arms: vec![SqlCaseArm {
                    condition: SqlExpr::NullTest {
                        expr: Box::new(SqlExpr::Field("guild_rank".to_string())),
                        negated: false,
                    },
                    result: SqlExpr::Literal(Value::Text("unguilded".to_string())),
                }],
                else_expr: Some(Box::new(SqlExpr::Field("guild_rank".to_string()))),
            })]),
            projection_aliases: vec![Some("guild_label".to_string())],
            predicate: None,
            distinct: false,
            group_by: vec![],
            having: vec![],
            order_by: vec![],
            limit: None,
            offset: None,
        }),
        "searched CASE projection should keep IS NULL conditions on the shared SQL-expression boundary",
    );
}

#[test]
fn parse_select_statement_with_searched_case_aggregate_input_expression() {
    let statement = parse_sql("SELECT SUM(CASE WHEN age >= 21 THEN 1 ELSE 0 END) FROM users")
        .expect("searched CASE aggregate-input select statement should parse");

    assert_eq!(
        statement,
        SqlStatement::Select(SqlSelectStatement {
            entity: "users".to_string(),
            table_alias: None,
            projection: SqlProjection::Items(vec![SqlSelectItem::Aggregate(SqlAggregateCall {
                kind: SqlAggregateKind::Sum,
                input: Some(Box::new(SqlExpr::Case {
                    arms: vec![SqlCaseArm {
                        condition: SqlExpr::Binary {
                            op: SqlExprBinaryOp::Gte,
                            left: Box::new(SqlExpr::Field("age".to_string())),
                            right: Box::new(SqlExpr::Literal(Value::Int64(21))),
                        },
                        result: SqlExpr::Literal(Value::Int64(1)),
                    }],
                    else_expr: Some(Box::new(SqlExpr::Literal(Value::Int64(0)))),
                })),
                filter_expr: None,
                distinct: false,
            })]),
            projection_aliases: vec![None],
            predicate: None,
            distinct: false,
            group_by: vec![],
            having: vec![],
            order_by: vec![],
            limit: None,
            offset: None,
        }),
        "searched CASE aggregate inputs should stay on the shared SQL-expression boundary",
    );
}

#[test]
fn parse_select_statement_with_searched_case_where_expression() {
    let statement = parse_sql(
        "SELECT name FROM users \
         WHERE CASE WHEN age >= 30 THEN TRUE ELSE age = 20 END \
         ORDER BY age ASC",
    )
    .expect("searched CASE WHERE select statement should parse");

    assert_eq!(
        statement,
        SqlStatement::Select(SqlSelectStatement {
            entity: "users".to_string(),
            table_alias: None,
            projection: SqlProjection::Items(vec![SqlSelectItem::Field("name".to_string())]),
            projection_aliases: vec![None],
            predicate: Some(SqlExpr::Case {
                arms: vec![SqlCaseArm {
                    condition: SqlExpr::Binary {
                        op: SqlExprBinaryOp::Gte,
                        left: Box::new(SqlExpr::Field("age".to_string())),
                        right: Box::new(SqlExpr::Literal(Value::Int64(30))),
                    },
                    result: SqlExpr::Literal(Value::Bool(true)),
                }],
                else_expr: Some(Box::new(SqlExpr::Binary {
                    op: SqlExprBinaryOp::Eq,
                    left: Box::new(SqlExpr::Field("age".to_string())),
                    right: Box::new(SqlExpr::Literal(Value::Int64(20))),
                })),
            }),
            distinct: false,
            group_by: vec![],
            having: vec![],
            order_by: vec![SqlOrderTerm {
                field: sql_order_expr("age"),
                direction: SqlOrderDirection::Asc,
            }],
            limit: None,
            offset: None,
        }),
        "searched CASE WHERE should stay on the shared pre-aggregate SQL-expression boundary",
    );
}

#[test]
fn parse_select_statement_distinguishes_is_null_from_eq_null() {
    let is_null = parse_sql("SELECT * FROM users WHERE age IS NULL ORDER BY age ASC")
        .expect("IS NULL select statement should parse");
    let eq_null = parse_sql("SELECT * FROM users WHERE age = NULL ORDER BY age ASC")
        .expect("= NULL select statement should parse");

    let SqlStatement::Select(is_null) = is_null else {
        panic!("expected parsed IS NULL select statement");
    };
    let SqlStatement::Select(eq_null) = eq_null else {
        panic!("expected parsed = NULL select statement");
    };

    assert_eq!(
        is_null.predicate,
        Some(SqlExpr::NullTest {
            expr: Box::new(SqlExpr::Field("age".to_string())),
            negated: false,
        }),
        "IS NULL should preserve one dedicated null-test SQL expression node",
    );
    assert_eq!(
        eq_null.predicate,
        Some(SqlExpr::Binary {
            op: SqlExprBinaryOp::Eq,
            left: Box::new(SqlExpr::Field("age".to_string())),
            right: Box::new(SqlExpr::Literal(Value::Null)),
        }),
        "= NULL should stay one ordinary equality expression instead of collapsing into the IS NULL node",
    );
}

#[test]
fn parse_select_statement_with_field_to_field_predicate() {
    let statement =
        parse_sql("SELECT * FROM users WHERE age > rank AND name = label ORDER BY age ASC")
            .expect("field-to-field predicate select statement should parse");

    assert_eq!(
        statement,
        SqlStatement::Select(SqlSelectStatement {
            entity: "users".to_string(),
            table_alias: None,
            projection: SqlProjection::All,
            projection_aliases: vec![],
            predicate: option_sql_pred!(Predicate::And(vec![
                Predicate::CompareFields(CompareFieldsPredicate::with_coercion(
                    "age",
                    CompareOp::Gt,
                    "rank",
                    CoercionId::NumericWiden,
                )),
                Predicate::eq_fields("name".to_string(), "label".to_string()),
            ])),
            distinct: false,
            group_by: vec![],
            having: vec![],
            order_by: vec![SqlOrderTerm {
                field: sql_order_expr("age"),
                direction: SqlOrderDirection::Asc,
            }],
            limit: None,
            offset: None,
        }),
    );
}

#[test]
fn parse_select_statement_with_field_path_predicate() {
    let statement = parse_sql("SELECT * FROM users WHERE profile.rank = 5")
        .expect("field-path predicate select statement should parse");

    assert_eq!(
        statement,
        SqlStatement::Select(SqlSelectStatement {
            entity: "users".to_string(),
            table_alias: None,
            projection: SqlProjection::All,
            projection_aliases: vec![],
            predicate: Some(sql_binary_expr(
                SqlExpr::Field("profile.rank".to_string()),
                SqlExprBinaryOp::Eq,
                SqlExpr::Literal(Value::Int64(5)),
            )),
            distinct: false,
            group_by: vec![],
            having: vec![],
            order_by: vec![],
            limit: None,
            offset: None,
        }),
    );
}

#[test]
fn parse_select_statement_with_symmetric_predicate_forms() {
    let statement =
        parse_sql("SELECT * FROM users WHERE 5 < age AND dexterity = strength ORDER BY age ASC")
            .expect("symmetric predicate forms should parse");

    assert_eq!(
        statement,
        SqlStatement::Select(SqlSelectStatement {
            entity: "users".to_string(),
            table_alias: None,
            projection: SqlProjection::All,
            projection_aliases: vec![],
            predicate: Some(sql_binary_expr(
                sql_binary_expr(
                    SqlExpr::Literal(Value::Int64(5)),
                    SqlExprBinaryOp::Lt,
                    SqlExpr::Field("age".to_string()),
                ),
                SqlExprBinaryOp::And,
                sql_binary_expr(
                    SqlExpr::Field("dexterity".to_string()),
                    SqlExprBinaryOp::Eq,
                    SqlExpr::Field("strength".to_string()),
                ),
            )),
            distinct: false,
            group_by: vec![],
            having: vec![],
            order_by: vec![SqlOrderTerm {
                field: sql_order_expr("age"),
                direction: SqlOrderDirection::Asc,
            }],
            limit: None,
            offset: None,
        }),
    );
}

#[test]
fn parse_select_statement_with_round_projection_items() {
    for (sql, expected_item, context) in [
        (
            "SELECT ROUND(age, 2) FROM users",
            sql_round_item(SqlExpr::Field("age".to_string()), Value::Int64(2)),
            "round over plain field",
        ),
        (
            "SELECT ROUND(age / 3, 2) FROM users",
            sql_round_item(
                sql_binary_expr(
                    SqlExpr::Field("age".to_string()),
                    SqlExprBinaryOp::Div,
                    SqlExpr::Literal(Value::Int64(3)),
                ),
                Value::Int64(2),
            ),
            "round over bounded arithmetic expression",
        ),
        (
            "SELECT ROUND(age + salary, 2) FROM users",
            sql_round_item(
                sql_binary_expr(
                    SqlExpr::Field("age".to_string()),
                    SqlExprBinaryOp::Add,
                    SqlExpr::Field("salary".to_string()),
                ),
                Value::Int64(2),
            ),
            "round over bounded field-to-field arithmetic expression",
        ),
    ] {
        let statement =
            parse_sql(sql).unwrap_or_else(|err| panic!("{context} should parse: {err:?}"));

        assert_eq!(
            statement,
            SqlStatement::Select(SqlSelectStatement {
                entity: "users".to_string(),
                table_alias: None,
                projection: SqlProjection::Items(vec![expected_item]),
                projection_aliases: vec![None],
                predicate: None,
                distinct: false,
                group_by: vec![],
                having: vec![],
                order_by: vec![],
                limit: None,
                offset: None,
            }),
            "{context} should lower to one bounded ROUND projection item",
        );
    }
}

#[test]
fn parse_select_statement_with_scalar_field_plus_field_projection_item() {
    let statement = parse_sql("SELECT age + salary FROM users")
        .expect("field-plus-field projection should parse in the bounded projection slice");

    assert_eq!(
        statement,
        SqlStatement::Select(SqlSelectStatement {
            entity: "users".to_string(),
            table_alias: None,
            projection: SqlProjection::Items(vec![SqlSelectItem::Expr(sql_binary_expr(
                SqlExpr::Field("age".to_string()),
                SqlExprBinaryOp::Add,
                SqlExpr::Field("salary".to_string()),
            ))]),
            projection_aliases: vec![None],
            predicate: None,
            distinct: false,
            group_by: vec![],
            having: vec![],
            order_by: vec![],
            limit: None,
            offset: None,
        }),
    );
}

#[test]
fn parse_select_statement_rejects_round_without_integer_scale() {
    let err = parse_sql("SELECT ROUND(age, name) FROM users")
        .expect_err("ROUND scale should remain literal-only in the bounded slice");

    std::assert_matches!(err, SqlParseError::InvalidSyntax { .. });
}

#[test]
fn parse_select_statement_accepts_arithmetic_predicates_for_shared_where_expr_lowering() {
    let statement = parse_sql("SELECT * FROM users WHERE age + 1 > 10")
        .expect("arithmetic WHERE predicates should parse through the shared SqlExpr seam");

    let SqlStatement::Select(statement) = statement else {
        panic!("expected SELECT statement");
    };

    assert!(
        matches!(
            statement.predicate,
            Some(SqlExpr::Binary {
                op: SqlExprBinaryOp::Gt,
                ..
            })
        ),
        "arithmetic WHERE predicate should stay parser-owned syntax and leave admission to lowering",
    );
}

#[test]
fn parse_select_statement_accepts_expression_predicate_near_misses_for_lowering_validation() {
    for sql in [
        "SELECT * FROM users WHERE strength = dexterity + 1",
        "SELECT * FROM users WHERE strength + dexterity = 10",
        "SELECT * FROM users WHERE ROUND(strength, 1) = dexterity",
    ] {
        let statement =
            parse_sql(sql).expect("WHERE near-miss expressions should parse and fail later");

        assert!(
            matches!(
                statement,
                SqlStatement::Select(SqlSelectStatement {
                    predicate: Some(_),
                    ..
                })
            ),
            "expression predicate near-miss should remain parser-owned syntax for lowering validation: {sql}",
        );
    }
}

#[test]
fn parse_select_statement_with_expression_order_terms() {
    let statement =
        parse_sql("SELECT * FROM users ORDER BY LOWER(name) DESC, UPPER(email) ASC LIMIT 2")
            .expect("expression order select statement should parse");

    assert_eq!(
        statement,
        SqlStatement::Select(SqlSelectStatement {
            entity: "users".to_string(),
            table_alias: None,
            projection: SqlProjection::All,
            projection_aliases: Vec::default(),
            predicate: None,
            distinct: false,
            group_by: vec![],
            having: vec![],
            order_by: vec![
                SqlOrderTerm {
                    field: sql_order_expr("LOWER(name)"),
                    direction: SqlOrderDirection::Desc,
                },
                SqlOrderTerm {
                    field: sql_order_expr("UPPER(email)"),
                    direction: SqlOrderDirection::Asc,
                },
            ],
            limit: Some(2),
            offset: None,
        }),
    );
}

#[test]
fn parse_select_statement_with_direct_bounded_computed_order_terms() {
    let statement = parse_sql(
        "SELECT * FROM users ORDER BY age + 1 ASC, age + salary DESC, ROUND(age / 3, 2) ASC LIMIT 2",
    )
    .expect("direct bounded computed ORDER BY terms should parse");

    assert_eq!(
        statement,
        SqlStatement::Select(SqlSelectStatement {
            entity: "users".to_string(),
            table_alias: None,
            projection: SqlProjection::All,
            projection_aliases: Vec::default(),
            predicate: None,
            distinct: false,
            group_by: vec![],
            having: vec![],
            order_by: vec![
                SqlOrderTerm {
                    field: sql_order_expr("age + 1"),
                    direction: SqlOrderDirection::Asc,
                },
                SqlOrderTerm {
                    field: sql_order_expr("age + salary"),
                    direction: SqlOrderDirection::Desc,
                },
                SqlOrderTerm {
                    field: sql_order_expr("ROUND(age / 3, 2)"),
                    direction: SqlOrderDirection::Asc,
                },
            ],
            limit: Some(2),
            offset: None,
        }),
    );
}

#[test]
fn parse_select_statement_with_direct_scalar_function_expression_order_terms() {
    let statement = parse_sql(
        "SELECT * FROM users \
         ORDER BY ABS(age - 30) ASC, COALESCE(NULLIF(age, 20), 99) DESC LIMIT 2",
    )
    .expect("direct scalar-function expression ORDER BY terms should parse");

    assert_eq!(
        statement,
        SqlStatement::Select(SqlSelectStatement {
            entity: "users".to_string(),
            table_alias: None,
            projection: SqlProjection::All,
            projection_aliases: Vec::default(),
            predicate: None,
            distinct: false,
            group_by: vec![],
            having: vec![],
            order_by: vec![
                SqlOrderTerm {
                    field: sql_order_expr("ABS(age - 30)"),
                    direction: SqlOrderDirection::Asc,
                },
                SqlOrderTerm {
                    field: sql_order_expr("COALESCE(NULLIF(age, 20), 99)"),
                    direction: SqlOrderDirection::Desc,
                },
            ],
            limit: Some(2),
            offset: None,
        }),
    );
}

#[test]
fn parse_select_statement_with_direct_searched_case_order_terms() {
    let statement = parse_sql(
        "SELECT * FROM users \
         ORDER BY CASE WHEN age >= 21 THEN rank ELSE age END DESC LIMIT 2",
    )
    .expect("direct searched CASE ORDER BY terms should parse");

    assert_eq!(
        statement,
        SqlStatement::Select(SqlSelectStatement {
            entity: "users".to_string(),
            table_alias: None,
            projection: SqlProjection::All,
            projection_aliases: Vec::default(),
            predicate: None,
            distinct: false,
            group_by: vec![],
            having: vec![],
            order_by: vec![SqlOrderTerm {
                field: sql_order_expr("CASE WHEN age >= 21 THEN rank ELSE age END"),
                direction: SqlOrderDirection::Desc,
            }],
            limit: Some(2),
            offset: None,
        }),
    );
}

#[test]
fn parse_select_statement_with_direct_unary_text_function_expression_order_terms() {
    let statement = parse_sql(
        "SELECT * FROM users \
         ORDER BY LOWER(COALESCE(name, 'fallback')) ASC, LENGTH(TRIM(name)) DESC LIMIT 2",
    )
    .expect("direct unary text-function expression ORDER BY terms should parse");

    assert_eq!(
        statement,
        SqlStatement::Select(SqlSelectStatement {
            entity: "users".to_string(),
            table_alias: None,
            projection: SqlProjection::All,
            projection_aliases: Vec::default(),
            predicate: None,
            distinct: false,
            group_by: vec![],
            having: vec![],
            order_by: vec![
                SqlOrderTerm {
                    field: sql_order_expr("LOWER(COALESCE(name, 'fallback'))"),
                    direction: SqlOrderDirection::Asc,
                },
                SqlOrderTerm {
                    field: sql_order_expr("LENGTH(TRIM(name))"),
                    direction: SqlOrderDirection::Desc,
                },
            ],
            limit: Some(2),
            offset: None,
        }),
    );
}

#[test]
fn parse_select_statement_with_supported_scalar_text_order_terms() {
    let statement = parse_sql(
        "SELECT * FROM users ORDER BY TRIM(name), LTRIM(name), RTRIM(name), LENGTH(name) DESC, LEFT(name, 2), POSITION('a', name) DESC",
    )
    .expect("supported scalar-function ORDER BY terms should parse");

    assert_eq!(
        statement,
        SqlStatement::Select(SqlSelectStatement {
            entity: "users".to_string(),
            table_alias: None,
            projection: SqlProjection::All,
            projection_aliases: vec![],
            distinct: false,
            predicate: None,
            group_by: vec![],
            having: vec![],
            order_by: vec![
                SqlOrderTerm {
                    field: sql_order_expr("TRIM(name)"),
                    direction: SqlOrderDirection::Asc,
                },
                SqlOrderTerm {
                    field: sql_order_expr("LTRIM(name)"),
                    direction: SqlOrderDirection::Asc,
                },
                SqlOrderTerm {
                    field: sql_order_expr("RTRIM(name)"),
                    direction: SqlOrderDirection::Asc,
                },
                SqlOrderTerm {
                    field: sql_order_expr("LENGTH(name)"),
                    direction: SqlOrderDirection::Desc,
                },
                SqlOrderTerm {
                    field: sql_order_expr("LEFT(name, 2)"),
                    direction: SqlOrderDirection::Asc,
                },
                SqlOrderTerm {
                    field: sql_order_expr("POSITION('a', name)"),
                    direction: SqlOrderDirection::Desc,
                },
            ],
            limit: None,
            offset: None,
        }),
    );
}

#[test]
fn parse_select_statement_with_left_and_right_projection_items() {
    let statement = parse_sql("SELECT LEFT(name, 2), RIGHT(name, 3) FROM users")
        .expect("left/right projection select statement should parse");

    assert_eq!(
        statement,
        SqlStatement::Select(SqlSelectStatement {
            entity: "users".to_string(),
            table_alias: None,
            projection: SqlProjection::Items(vec![
                SqlSelectItem::Expr(sql_scalar_function_expr(
                    SqlScalarFunction::Left,
                    vec![
                        SqlExpr::Field("name".to_string()),
                        SqlExpr::Literal(Value::Int64(2)),
                    ],
                )),
                SqlSelectItem::Expr(sql_scalar_function_expr(
                    SqlScalarFunction::Right,
                    vec![
                        SqlExpr::Field("name".to_string()),
                        SqlExpr::Literal(Value::Int64(3)),
                    ],
                )),
            ]),
            projection_aliases: vec![None, None],
            predicate: None,
            distinct: false,
            group_by: vec![],
            having: vec![],
            order_by: vec![],
            limit: None,
            offset: None,
        }),
    );
}

#[test]
fn parse_select_statement_with_starts_ends_and_position_projection_items() {
    let statement = parse_sql(
        "SELECT STARTS_WITH(name, 'A'), ENDS_WITH(name, 'z'), CONTAINS(name, 'd'), POSITION('da', name) FROM users",
    )
    .expect("text predicate projection select statement should parse");

    assert_eq!(
        statement,
        SqlStatement::Select(SqlSelectStatement {
            entity: "users".to_string(),
            table_alias: None,
            projection: SqlProjection::Items(vec![
                SqlSelectItem::Expr(sql_scalar_function_expr(
                    SqlScalarFunction::StartsWith,
                    vec![
                        SqlExpr::Field("name".to_string()),
                        SqlExpr::Literal(Value::Text("A".to_string())),
                    ],
                )),
                SqlSelectItem::Expr(sql_scalar_function_expr(
                    SqlScalarFunction::EndsWith,
                    vec![
                        SqlExpr::Field("name".to_string()),
                        SqlExpr::Literal(Value::Text("z".to_string())),
                    ],
                )),
                SqlSelectItem::Expr(sql_scalar_function_expr(
                    SqlScalarFunction::Contains,
                    vec![
                        SqlExpr::Field("name".to_string()),
                        SqlExpr::Literal(Value::Text("d".to_string())),
                    ],
                )),
                SqlSelectItem::Expr(sql_scalar_function_expr(
                    SqlScalarFunction::Position,
                    vec![
                        SqlExpr::Literal(Value::Text("da".to_string())),
                        SqlExpr::Field("name".to_string()),
                    ],
                )),
            ]),
            projection_aliases: vec![None, None, None, None],
            predicate: None,
            distinct: false,
            group_by: vec![],
            having: vec![],
            order_by: vec![],
            limit: None,
            offset: None,
        }),
    );
}

#[test]
fn parse_select_statement_with_replace_projection_item() {
    let statement = parse_sql("SELECT REPLACE(name, 'A', 'E') FROM users")
        .expect("replace projection select statement should parse");

    assert_eq!(
        statement,
        SqlStatement::Select(SqlSelectStatement {
            entity: "users".to_string(),
            table_alias: None,
            projection: SqlProjection::Items(vec![SqlSelectItem::Expr(sql_scalar_function_expr(
                SqlScalarFunction::Replace,
                vec![
                    SqlExpr::Field("name".to_string()),
                    SqlExpr::Literal(Value::Text("A".to_string())),
                    SqlExpr::Literal(Value::Text("E".to_string())),
                ],
            ))]),
            projection_aliases: vec![None],
            predicate: None,
            distinct: false,
            group_by: vec![],
            having: vec![],
            order_by: vec![],
            limit: None,
            offset: None,
        }),
    );
}

#[test]
fn parse_select_statement_with_substring_projection_item() {
    let statement = parse_sql("SELECT SUBSTRING(name, 2, 3), SUBSTRING(name, 2) FROM users")
        .expect("substring projection select statement should parse");

    assert_eq!(
        statement,
        SqlStatement::Select(SqlSelectStatement {
            entity: "users".to_string(),
            table_alias: None,
            projection: SqlProjection::Items(vec![
                SqlSelectItem::Expr(sql_scalar_function_expr(
                    SqlScalarFunction::Substring,
                    vec![
                        SqlExpr::Field("name".to_string()),
                        SqlExpr::Literal(Value::Int64(2)),
                        SqlExpr::Literal(Value::Int64(3)),
                    ],
                )),
                SqlSelectItem::Expr(sql_scalar_function_expr(
                    SqlScalarFunction::Substring,
                    vec![
                        SqlExpr::Field("name".to_string()),
                        SqlExpr::Literal(Value::Int64(2)),
                    ],
                )),
            ]),
            projection_aliases: vec![None, None],
            predicate: None,
            distinct: false,
            group_by: vec![],
            having: vec![],
            order_by: vec![],
            limit: None,
            offset: None,
        }),
    );
}

#[test]
fn parse_delete_statement_with_limit() {
    let statement = parse_sql("DELETE FROM users WHERE age < 18 ORDER BY age LIMIT 3")
        .expect("delete statement should parse");

    assert_eq!(
        statement,
        SqlStatement::Delete(SqlDeleteStatement {
            entity: "users".to_string(),
            table_alias: None,
            predicate: option_sql_pred!(Predicate::Compare(ComparePredicate::with_coercion(
                "age",
                CompareOp::Lt,
                Value::Int64(18),
                CoercionId::NumericWiden,
            ))),
            order_by: vec![SqlOrderTerm {
                field: sql_order_expr("age"),
                direction: SqlOrderDirection::Asc,
            }],
            limit: Some(3),
            offset: None,
            returning: None,
        }),
    );
}

#[test]
fn parse_delete_statement_with_limit_and_offset() {
    let statement = parse_sql("DELETE FROM users WHERE age < 18 ORDER BY age LIMIT 3 OFFSET 1")
        .expect("delete statement with offset should parse");

    assert_eq!(
        statement,
        SqlStatement::Delete(SqlDeleteStatement {
            entity: "users".to_string(),
            table_alias: None,
            predicate: option_sql_pred!(Predicate::Compare(ComparePredicate::with_coercion(
                "age",
                CompareOp::Lt,
                Value::Int64(18),
                CoercionId::NumericWiden,
            ))),
            order_by: vec![SqlOrderTerm {
                field: sql_order_expr("age"),
                direction: SqlOrderDirection::Asc,
            }],
            limit: Some(3),
            offset: Some(1),
            returning: None,
        }),
    );
}

#[test]
fn parse_delete_statement_accepts_single_table_alias() {
    let statement = parse_sql(
        "DELETE FROM users u WHERE u.age < 18 ORDER BY LOWER(u.name) ASC LIMIT 3 OFFSET 1",
    )
    .expect("delete statement with one table alias should parse");

    assert_eq!(
        statement,
        SqlStatement::Delete(SqlDeleteStatement {
            entity: "users".to_string(),
            table_alias: Some("u".to_string()),
            predicate: option_sql_pred!(Predicate::Compare(ComparePredicate::with_coercion(
                "u.age",
                CompareOp::Lt,
                Value::Int64(18),
                CoercionId::NumericWiden,
            ))),
            order_by: vec![SqlOrderTerm {
                field: sql_order_expr("LOWER(u.name)"),
                direction: SqlOrderDirection::Asc,
            }],
            limit: Some(3),
            offset: Some(1),
            returning: None,
        }),
    );
}

#[test]
fn parse_delete_statement_with_direct_starts_with_family() {
    let cases = [
        (
            "DELETE FROM users WHERE STARTS_WITH(name, 'Al') ORDER BY id ASC LIMIT 1",
            sql_scalar_function_expr(
                SqlScalarFunction::StartsWith,
                vec![
                    SqlExpr::Field("name".to_string()),
                    SqlExpr::Literal(Value::Text("Al".to_string())),
                ],
            ),
        ),
        (
            "DELETE FROM users WHERE STARTS_WITH(LOWER(name), 'Al') ORDER BY id ASC LIMIT 1",
            sql_scalar_function_expr(
                SqlScalarFunction::StartsWith,
                vec![
                    sql_scalar_function_field_expr(SqlScalarFunction::Lower, "name"),
                    SqlExpr::Literal(Value::Text("Al".to_string())),
                ],
            ),
        ),
        (
            "DELETE FROM users WHERE STARTS_WITH(UPPER(name), 'AL') ORDER BY id ASC LIMIT 1",
            sql_scalar_function_expr(
                SqlScalarFunction::StartsWith,
                vec![
                    sql_scalar_function_field_expr(SqlScalarFunction::Upper, "name"),
                    SqlExpr::Literal(Value::Text("AL".to_string())),
                ],
            ),
        ),
    ];

    for (sql, expected_predicate) in cases {
        let statement = parse_sql(sql).expect("direct STARTS_WITH delete statement should parse");

        assert_eq!(
            statement,
            SqlStatement::Delete(SqlDeleteStatement {
                entity: "users".to_string(),
                table_alias: None,
                predicate: Some(expected_predicate),
                order_by: vec![SqlOrderTerm {
                    field: sql_order_expr("id"),
                    direction: SqlOrderDirection::Asc,
                }],
                limit: Some(1),
                offset: None,
                returning: None,
            }),
        );
    }
}

#[test]
#[cfg(feature = "sql-explain")]
fn parse_explain_json_wrapped_select() {
    let statement = parse_sql("EXPLAIN JSON SELECT * FROM users LIMIT 1")
        .expect("explain statement should parse");

    assert_eq!(
        statement,
        SqlStatement::Explain(SqlExplainStatement {
            mode: SqlExplainMode::Json,
            verbose: false,
            statement: SqlExplainTarget::Select(SqlSelectStatement {
                entity: "users".to_string(),
                table_alias: None,
                projection: SqlProjection::All,
                projection_aliases: Vec::default(),
                predicate: None,
                distinct: false,
                group_by: vec![],
                having: vec![],
                order_by: vec![],
                limit: Some(1),
                offset: None,
            }),
        }),
    );
}

#[test]
#[cfg(feature = "sql-explain")]
fn parse_explain_execution_json_wrapped_select() {
    let statement = parse_sql("EXPLAIN EXECUTION JSON SELECT * FROM users LIMIT 1")
        .expect("execution-json explain statement should parse");

    assert_eq!(
        statement,
        SqlStatement::Explain(SqlExplainStatement {
            mode: SqlExplainMode::ExecutionJson,
            verbose: false,
            statement: SqlExplainTarget::Select(SqlSelectStatement {
                entity: "users".to_string(),
                table_alias: None,
                projection: SqlProjection::All,
                projection_aliases: Vec::default(),
                predicate: None,
                distinct: false,
                group_by: vec![],
                having: vec![],
                order_by: vec![],
                limit: Some(1),
                offset: None,
            }),
        }),
    );
}

#[test]
#[cfg(not(feature = "sql-explain"))]
fn parse_explain_requires_sql_explain_feature() {
    let err = parse_sql("EXPLAIN SELECT * FROM users").expect_err("EXPLAIN should be gated");

    assert_eq!(
        err,
        SqlParseError::UnsupportedFeature {
            feature: SqlFeatureCode::Other,
        },
    );
}

#[test]
#[cfg(feature = "sql-explain")]
fn parse_explain_json_wrapped_delete_with_direct_starts_with_family() {
    let cases = [
        (
            "EXPLAIN JSON DELETE FROM users WHERE STARTS_WITH(name, 'Al') ORDER BY id ASC LIMIT 1",
            sql_scalar_function_expr(
                SqlScalarFunction::StartsWith,
                vec![
                    SqlExpr::Field("name".to_string()),
                    SqlExpr::Literal(Value::Text("Al".to_string())),
                ],
            ),
        ),
        (
            "EXPLAIN JSON DELETE FROM users WHERE STARTS_WITH(LOWER(name), 'Al') ORDER BY id ASC LIMIT 1",
            sql_scalar_function_expr(
                SqlScalarFunction::StartsWith,
                vec![
                    sql_scalar_function_field_expr(SqlScalarFunction::Lower, "name"),
                    SqlExpr::Literal(Value::Text("Al".to_string())),
                ],
            ),
        ),
        (
            "EXPLAIN JSON DELETE FROM users WHERE STARTS_WITH(UPPER(name), 'AL') ORDER BY id ASC LIMIT 1",
            sql_scalar_function_expr(
                SqlScalarFunction::StartsWith,
                vec![
                    sql_scalar_function_field_expr(SqlScalarFunction::Upper, "name"),
                    SqlExpr::Literal(Value::Text("AL".to_string())),
                ],
            ),
        ),
    ];

    for (sql, expected_predicate) in cases {
        let statement =
            parse_sql(sql).expect("EXPLAIN JSON direct STARTS_WITH delete should parse");

        assert_eq!(
            statement,
            SqlStatement::Explain(SqlExplainStatement {
                mode: SqlExplainMode::Json,
                verbose: false,
                statement: SqlExplainTarget::Delete(SqlDeleteStatement {
                    entity: "users".to_string(),
                    table_alias: None,
                    predicate: Some(expected_predicate),
                    order_by: vec![SqlOrderTerm {
                        field: sql_order_expr("id"),
                        direction: SqlOrderDirection::Asc,
                    }],
                    limit: Some(1),
                    offset: None,
                    returning: None,
                }),
            }),
        );
    }
}

#[test]
fn parse_describe_statement_with_schema_qualified_entity() {
    let statement = parse_sql("DESCRIBE public.users").expect("describe statement should parse");

    assert_eq!(
        statement,
        SqlStatement::Describe(SqlDescribeStatement {
            entity: "public.users".to_string(),
        }),
    );
}

#[test]
fn parse_show_indexes_statement_with_schema_qualified_entity() {
    let statement =
        parse_sql("SHOW INDEXES FROM public.users").expect("show indexes statement should parse");

    assert_eq!(
        statement,
        SqlStatement::ShowIndexes(SqlShowIndexesStatement {
            entity: "public.users".to_string(),
        }),
    );
}

#[test]
fn parse_show_constraints_statement_with_schema_qualified_entity() {
    let statement = parse_sql("SHOW CONSTRAINTS FROM public.users")
        .expect("show constraints statement should parse");

    assert_eq!(
        statement,
        SqlStatement::ShowConstraints(SqlShowConstraintsStatement {
            entity: "public.users".to_string(),
        }),
    );

    assert_eq!(
        parse_sql("SHOW CONSTRAINTS IN public.users")
            .expect("show constraints IN statement should parse"),
        statement,
    );
}

#[test]
fn parse_show_constraints_rejects_non_sql_entity_shortcut() {
    let err = parse_sql("SHOW CONSTRAINTS public.users")
        .expect_err("SHOW CONSTRAINTS requires FROM or IN");

    assert!(
        matches!(err, SqlParseError::InvalidSyntax { .. }),
        "SHOW CONSTRAINTS without FROM/IN should fail as invalid SQL syntax",
    );
}

#[test]
fn parse_show_indexes_statement_accepts_in_synonym() {
    let statement =
        parse_sql("SHOW INDEXES IN public.users").expect("show indexes IN statement should parse");

    assert_eq!(
        statement,
        SqlStatement::ShowIndexes(SqlShowIndexesStatement {
            entity: "public.users".to_string(),
        }),
    );
}

#[test]
fn parse_show_indexes_rejects_non_sql_entity_shortcut() {
    let err = parse_sql("SHOW INDEXES public.users").expect_err("SHOW INDEXES requires FROM or IN");

    assert!(
        matches!(err, SqlParseError::InvalidSyntax { .. }),
        "SHOW INDEXES without FROM/IN should fail as invalid SQL syntax",
    );
}

#[test]
fn parse_show_columns_statement_with_schema_qualified_entity() {
    let statement =
        parse_sql("SHOW COLUMNS public.users").expect("show columns statement should parse");

    assert_eq!(
        statement,
        SqlStatement::ShowColumns(SqlShowColumnsStatement {
            entity: "public.users".to_string(),
        }),
    );
}

#[test]
fn parse_show_entities_statement() {
    let statement = parse_sql("SHOW ENTITIES").expect("show entities statement should parse");

    assert_eq!(
        statement,
        SqlStatement::ShowEntities(SqlShowEntitiesStatement {
            entity: None,
            verbose: false,
        })
    );
}

#[test]
fn parse_show_entities_verbose_statement() {
    let statement =
        parse_sql("SHOW ENTITIES VERBOSE").expect("show entities verbose statement should parse");

    assert_eq!(
        statement,
        SqlStatement::ShowEntities(SqlShowEntitiesStatement {
            entity: None,
            verbose: true,
        })
    );
}

#[test]
fn parse_show_entity_statement() {
    let statement = parse_sql("SHOW ENTITY Character").expect("show entity statement should parse");

    assert_eq!(
        statement,
        SqlStatement::ShowEntities(SqlShowEntitiesStatement {
            entity: Some("Character".to_string()),
            verbose: false,
        })
    );
}

#[test]
fn parse_show_entity_verbose_statement() {
    let statement = parse_sql("SHOW ENTITY Character VERBOSE")
        .expect("show entity verbose statement should parse");

    assert_eq!(
        statement,
        SqlStatement::ShowEntities(SqlShowEntitiesStatement {
            entity: Some("Character".to_string()),
            verbose: true,
        })
    );
}

#[test]
fn parse_show_stores_statement() {
    let statement = parse_sql("SHOW STORES").expect("show stores statement should parse");

    assert_eq!(
        statement,
        SqlStatement::ShowStores(SqlShowStoresStatement { verbose: false })
    );
}

#[test]
fn parse_show_stores_verbose_statement() {
    let statement =
        parse_sql("SHOW STORES VERBOSE").expect("show stores verbose statement should parse");

    assert_eq!(
        statement,
        SqlStatement::ShowStores(SqlShowStoresStatement { verbose: true })
    );
}

#[test]
fn parse_show_memory_statement() {
    let statement = parse_sql("SHOW MEMORY").expect("show memory statement should parse");

    assert_eq!(statement, SqlStatement::ShowMemory(SqlShowMemoryStatement));
}

#[test]
fn parse_create_index_statement_keeps_ddl_intent_unresolved() {
    let statement = parse_sql("CREATE INDEX user_age_idx ON public.users (profile.age)")
        .expect("CREATE INDEX statement should parse");

    assert_eq!(
        statement,
        SqlStatement::Ddl(SqlDdlStatement::CreateIndex(SqlCreateIndexStatement {
            name: "user_age_idx".to_string(),
            entity: "public.users".to_string(),
            key_items: ddl_field_paths(&["profile.age"]),
            predicate_sql: None,
            uniqueness: SqlCreateIndexUniqueness::NonUnique,
            if_not_exists: false,
            schema_version_contract: SqlDdlSchemaVersionContract::default(),
        })),
    );
}

#[test]
fn parse_create_multi_field_index_statement_keeps_ddl_intent_unresolved() {
    let statement = parse_sql("CREATE INDEX user_age_name_idx ON public.users (age, name)")
        .expect("CREATE INDEX with multiple field paths should parse");

    assert_eq!(
        statement,
        SqlStatement::Ddl(SqlDdlStatement::CreateIndex(SqlCreateIndexStatement {
            name: "user_age_name_idx".to_string(),
            entity: "public.users".to_string(),
            key_items: ddl_field_paths(&["age", "name"]),
            predicate_sql: None,
            uniqueness: SqlCreateIndexUniqueness::NonUnique,
            if_not_exists: false,
            schema_version_contract: SqlDdlSchemaVersionContract::default(),
        })),
    );
}

#[test]
fn parse_create_index_treats_asc_as_default_order() {
    let statement = parse_sql("CREATE INDEX user_age_name_idx ON public.users (age ASC, name ASC)")
        .expect("CREATE INDEX with explicit ASC field paths should parse");

    assert_eq!(
        statement,
        SqlStatement::Ddl(SqlDdlStatement::CreateIndex(SqlCreateIndexStatement {
            name: "user_age_name_idx".to_string(),
            entity: "public.users".to_string(),
            key_items: ddl_field_paths(&["age", "name"]),
            predicate_sql: None,
            uniqueness: SqlCreateIndexUniqueness::NonUnique,
            if_not_exists: false,
            schema_version_contract: SqlDdlSchemaVersionContract::default(),
        })),
    );
}

#[test]
fn parse_alter_table_add_column_statement_keeps_ddl_intent_unresolved() {
    let statement = parse_sql("ALTER TABLE public.users ADD COLUMN nickname text NULL")
        .expect("ALTER TABLE ADD COLUMN should parse as DDL intent");

    assert_eq!(
        statement,
        SqlStatement::Ddl(SqlDdlStatement::AlterTableAddColumn(
            SqlAlterTableAddColumnStatement {
                entity: "public.users".to_string(),
                column_name: "nickname".to_string(),
                column_type: "text".to_string(),
                nullable: true,
                default: None,
                schema_version_contract: SqlDdlSchemaVersionContract::default(),
            },
        )),
    );
}

#[test]
fn parse_alter_table_add_column_statement_keeps_default_and_nullability_intent() {
    let statement = parse_sql("ALTER TABLE users ADD COLUMN score nat64 DEFAULT 7 NOT NULL")
        .expect("ALTER TABLE ADD COLUMN with DEFAULT and NOT NULL should parse");

    assert_eq!(
        statement,
        SqlStatement::Ddl(SqlDdlStatement::AlterTableAddColumn(
            SqlAlterTableAddColumnStatement {
                entity: "users".to_string(),
                column_name: "score".to_string(),
                column_type: "nat64".to_string(),
                nullable: false,
                default: Some(Value::Int64(7)),
                schema_version_contract: SqlDdlSchemaVersionContract::default(),
            },
        )),
    );
}

#[test]
fn parse_alter_table_add_column_statement_accepts_not_null_before_default() {
    let statement = parse_sql("ALTER TABLE users ADD COLUMN score nat64 NOT NULL DEFAULT 0")
        .expect("ALTER TABLE ADD COLUMN with NOT NULL before DEFAULT should parse");

    assert_eq!(
        statement,
        SqlStatement::Ddl(SqlDdlStatement::AlterTableAddColumn(
            SqlAlterTableAddColumnStatement {
                entity: "users".to_string(),
                column_name: "score".to_string(),
                column_type: "nat64".to_string(),
                nullable: false,
                default: Some(Value::Int64(0)),
                schema_version_contract: SqlDdlSchemaVersionContract::default(),
            },
        )),
    );
}

#[test]
fn parse_alter_table_add_column_statement_keeps_nat_big_max_bytes_type_modifier() {
    let statement =
        parse_sql("ALTER TABLE users ADD COLUMN score nat_big(max_bytes = 512) DEFAULT 0")
            .expect("ALTER TABLE ADD COLUMN with nat_big max_bytes modifier should parse");

    assert_eq!(
        statement,
        SqlStatement::Ddl(SqlDdlStatement::AlterTableAddColumn(
            SqlAlterTableAddColumnStatement {
                entity: "users".to_string(),
                column_name: "score".to_string(),
                column_type: "nat_big(max_bytes=512)".to_string(),
                nullable: true,
                default: Some(Value::Int64(0)),
                schema_version_contract: SqlDdlSchemaVersionContract::default(),
            },
        )),
    );
}

#[test]
fn parse_alter_table_add_column_statement_rejects_malformed_nat_big_max_bytes_modifier() {
    for sql in [
        "ALTER TABLE users ADD COLUMN score nat_big(max_len = 512)",
        "ALTER TABLE users ADD COLUMN score nat_big(max_bytes = -1)",
        "ALTER TABLE users ADD COLUMN score nat_big(max_bytes = 4294967296)",
        "ALTER TABLE users ADD COLUMN score nat_big(max_bytes = 1, max_bytes = 2)",
    ] {
        parse_sql(sql).expect_err("malformed nat_big max_bytes modifier should fail parsing");
    }
}

#[test]
fn parse_alter_table_alter_column_statement_keeps_default_intent_unresolved() {
    let statement = parse_sql("ALTER TABLE users ALTER COLUMN score SET DEFAULT 7")
        .expect("ALTER TABLE ALTER COLUMN SET DEFAULT should parse as DDL intent");

    assert_eq!(
        statement,
        SqlStatement::Ddl(SqlDdlStatement::AlterTableAlterColumn(
            SqlAlterTableAlterColumnStatement {
                entity: "users".to_string(),
                column_name: "score".to_string(),
                action: SqlAlterColumnAction::SetDefault(Value::Int64(7)),
                schema_version_contract: SqlDdlSchemaVersionContract::default(),
            },
        )),
    );
}

#[test]
fn parse_alter_table_alter_column_statement_keeps_nullability_intent_unresolved() {
    let set_statement = parse_sql("ALTER TABLE users ALTER COLUMN score SET NOT NULL")
        .expect("ALTER TABLE ALTER COLUMN SET NOT NULL should parse as DDL intent");
    let drop_statement = parse_sql("ALTER TABLE users ALTER COLUMN score DROP NOT NULL")
        .expect("ALTER TABLE ALTER COLUMN DROP NOT NULL should parse as DDL intent");
    let drop_default_statement = parse_sql("ALTER TABLE users ALTER COLUMN score DROP DEFAULT")
        .expect("ALTER TABLE ALTER COLUMN DROP DEFAULT should parse as DDL intent");

    assert_eq!(
        set_statement,
        SqlStatement::Ddl(SqlDdlStatement::AlterTableAlterColumn(
            SqlAlterTableAlterColumnStatement {
                entity: "users".to_string(),
                column_name: "score".to_string(),
                action: SqlAlterColumnAction::SetNotNull,
                schema_version_contract: SqlDdlSchemaVersionContract::default(),
            },
        )),
    );
    assert_eq!(
        drop_statement,
        SqlStatement::Ddl(SqlDdlStatement::AlterTableAlterColumn(
            SqlAlterTableAlterColumnStatement {
                entity: "users".to_string(),
                column_name: "score".to_string(),
                action: SqlAlterColumnAction::DropNotNull,
                schema_version_contract: SqlDdlSchemaVersionContract::default(),
            },
        )),
    );
    assert_eq!(
        drop_default_statement,
        SqlStatement::Ddl(SqlDdlStatement::AlterTableAlterColumn(
            SqlAlterTableAlterColumnStatement {
                entity: "users".to_string(),
                column_name: "score".to_string(),
                action: SqlAlterColumnAction::DropDefault,
                schema_version_contract: SqlDdlSchemaVersionContract::default(),
            },
        )),
    );
}

#[test]
fn parse_alter_table_check_constraint_lifecycle_intent() {
    let add = parse_sql(
        "ALTER TABLE users ADD CONSTRAINT adult CHECK (age >= 18) NOT VALID \
         EXPECT SCHEMA VERSION 2 SET SCHEMA VERSION 3",
    )
    .expect("ADD CONSTRAINT CHECK NOT VALID should parse");
    let drop = parse_sql(
        "ALTER TABLE users DROP CONSTRAINT IF EXISTS adult \
         EXPECT SCHEMA VERSION 3 SET SCHEMA VERSION 4",
    )
    .expect("DROP CONSTRAINT IF EXISTS should parse");
    let validate = parse_sql("ALTER TABLE users VALIDATE CONSTRAINT adult AFTER 7")
        .expect("VALIDATE CONSTRAINT AFTER should parse");

    let SqlStatement::Ddl(SqlDdlStatement::AlterTableAddCheckConstraint(add)) = add else {
        panic!("ADD CONSTRAINT should keep typed DDL intent");
    };
    assert_eq!(add.entity, "users");
    assert_eq!(add.constraint_name, "adult");
    assert!(add.not_valid);
    assert_eq!(add.schema_version_contract.expected_schema_version, Some(2));
    assert_eq!(add.schema_version_contract.next_schema_version, Some(3));

    let SqlStatement::Ddl(SqlDdlStatement::AlterTableDropConstraint(drop)) = drop else {
        panic!("DROP CONSTRAINT should keep typed DDL intent");
    };
    assert!(drop.if_exists);
    assert_eq!(drop.constraint_name, "adult");

    let SqlStatement::Ddl(SqlDdlStatement::AlterTableValidateConstraint(validate)) = validate
    else {
        panic!("VALIDATE CONSTRAINT should keep typed progress intent");
    };
    assert_eq!(validate.constraint_name, "adult");
    assert_eq!(validate.after_page_sequence, Some(7));

    let SqlStatement::Ddl(SqlDdlStatement::AlterTableValidateConstraint(validate)) =
        parse_sql("ALTER TABLE Character VALIDATE CONSTRAINT adult_age AFTER 18446744073709551615")
            .expect("validation acknowledgements should admit the complete u64 identity domain")
    else {
        panic!("expected VALIDATE CONSTRAINT statement");
    };
    assert_eq!(validate.after_page_sequence, Some(u64::MAX));
    assert!(
        parse_sql(
            "ALTER TABLE Character VALIDATE CONSTRAINT adult_age AFTER 18446744073709551616",
        )
        .is_err(),
        "validation acknowledgement identities above u64 must fail closed",
    );
}

#[test]
fn parse_alter_table_drop_column_statement_keeps_ddl_intent_unresolved() {
    let statement = parse_sql("ALTER TABLE public.users DROP COLUMN nickname")
        .expect("ALTER TABLE DROP COLUMN should parse as DDL intent");

    assert_eq!(
        statement,
        SqlStatement::Ddl(SqlDdlStatement::AlterTableDropColumn(
            SqlAlterTableDropColumnStatement {
                entity: "public.users".to_string(),
                column_name: "nickname".to_string(),
                if_exists: false,
                schema_version_contract: SqlDdlSchemaVersionContract::default(),
            },
        )),
    );
}

#[test]
fn parse_alter_table_drop_column_if_exists_statement_keeps_ddl_intent_unresolved() {
    let statement = parse_sql("ALTER TABLE public.users DROP COLUMN IF EXISTS nickname")
        .expect("ALTER TABLE DROP COLUMN IF EXISTS should parse as DDL intent");

    assert_eq!(
        statement,
        SqlStatement::Ddl(SqlDdlStatement::AlterTableDropColumn(
            SqlAlterTableDropColumnStatement {
                entity: "public.users".to_string(),
                column_name: "nickname".to_string(),
                if_exists: true,
                schema_version_contract: SqlDdlSchemaVersionContract::default(),
            },
        )),
    );
}

#[test]
fn parse_alter_table_rename_column_statement_keeps_ddl_intent_unresolved() {
    let statement = parse_sql("ALTER TABLE public.users RENAME COLUMN nickname TO handle")
        .expect("ALTER TABLE RENAME COLUMN should parse as DDL intent");

    assert_eq!(
        statement,
        SqlStatement::Ddl(SqlDdlStatement::AlterTableRenameColumn(
            SqlAlterTableRenameColumnStatement {
                entity: "public.users".to_string(),
                old_column_name: "nickname".to_string(),
                new_column_name: "handle".to_string(),
                schema_version_contract: SqlDdlSchemaVersionContract::default(),
            },
        )),
    );
}

#[test]
fn parse_create_unique_index_statement_keeps_ddl_intent_unresolved() {
    let statement = parse_sql("CREATE UNIQUE INDEX user_age_idx ON public.users (profile.age)")
        .expect("CREATE UNIQUE INDEX statement should parse");

    assert_eq!(
        statement,
        SqlStatement::Ddl(SqlDdlStatement::CreateIndex(SqlCreateIndexStatement {
            name: "user_age_idx".to_string(),
            entity: "public.users".to_string(),
            key_items: ddl_field_paths(&["profile.age"]),
            predicate_sql: None,
            uniqueness: SqlCreateIndexUniqueness::Unique,
            if_not_exists: false,
            schema_version_contract: SqlDdlSchemaVersionContract::default(),
        })),
    );
}

#[test]
fn parse_create_index_keeps_expression_key_intent_unresolved() {
    let statement = parse_sql("CREATE INDEX user_lower_name_idx ON users (LOWER(name) ASC)")
        .expect("expression key CREATE INDEX should parse before catalog binding");

    assert_eq!(
        statement,
        SqlStatement::Ddl(SqlDdlStatement::CreateIndex(SqlCreateIndexStatement {
            name: "user_lower_name_idx".to_string(),
            entity: "users".to_string(),
            key_items: vec![ddl_expression_key(
                SqlCreateIndexExpressionFunction::Lower,
                "name",
            )],
            predicate_sql: None,
            uniqueness: SqlCreateIndexUniqueness::NonUnique,
            if_not_exists: false,
            schema_version_contract: SqlDdlSchemaVersionContract::default(),
        })),
    );
}

#[test]
fn parse_create_index_keeps_supported_expression_key_family_intent_unresolved() {
    let statement = parse_sql(
        "CREATE INDEX user_text_expr_idx ON users (LOWER(name), UPPER(code), TRIM(email))",
    )
    .expect("supported expression key family should parse before catalog binding");

    assert_eq!(
        statement,
        SqlStatement::Ddl(SqlDdlStatement::CreateIndex(SqlCreateIndexStatement {
            name: "user_text_expr_idx".to_string(),
            entity: "users".to_string(),
            key_items: vec![
                ddl_expression_key(SqlCreateIndexExpressionFunction::Lower, "name"),
                ddl_expression_key(SqlCreateIndexExpressionFunction::Upper, "code"),
                ddl_expression_key(SqlCreateIndexExpressionFunction::Trim, "email"),
            ],
            predicate_sql: None,
            uniqueness: SqlCreateIndexUniqueness::NonUnique,
            if_not_exists: false,
            schema_version_contract: SqlDdlSchemaVersionContract::default(),
        })),
    );
}

#[test]
fn parse_create_index_rejects_unknown_expression_index_functions() {
    let sql = "CREATE INDEX user_reverse_name_idx ON users (REVERSE(name))";
    let err = parse_sql(sql).expect_err("unknown expression function should fail closed");
    assert!(
        matches!(err, SqlParseError::UnsupportedFeature { .. }),
        "unknown expression key function should report a typed unsupported feature",
    );
}

#[test]
fn parse_create_index_keeps_filtered_index_predicate_sql() {
    let statement = parse_sql("CREATE INDEX user_age_idx ON users (age) WHERE active = true")
        .expect("filtered CREATE INDEX should parse");

    assert_eq!(
        statement,
        SqlStatement::Ddl(SqlDdlStatement::CreateIndex(SqlCreateIndexStatement {
            name: "user_age_idx".to_string(),
            entity: "users".to_string(),
            key_items: ddl_field_paths(&["age"]),
            predicate_sql: Some("active = TRUE".to_string()),
            uniqueness: SqlCreateIndexUniqueness::NonUnique,
            if_not_exists: false,
            schema_version_contract: SqlDdlSchemaVersionContract::default(),
        })),
    );
}

#[test]
fn parse_drop_index_statement_keeps_ddl_intent_unresolved() {
    let statement =
        parse_sql("DROP INDEX user_age_idx ON public.users").expect("DROP INDEX should parse");

    assert_eq!(
        statement,
        SqlStatement::Ddl(SqlDdlStatement::DropIndex(SqlDropIndexStatement {
            name: "user_age_idx".to_string(),
            entity: Some("public.users".to_string()),
            if_exists: false,
            schema_version_contract: SqlDdlSchemaVersionContract::default(),
        })),
    );
}

#[test]
fn parse_drop_index_statement_allows_implicit_typed_target() {
    let statement = parse_sql("DROP INDEX user_age_idx").expect("DROP INDEX should parse");

    assert_eq!(
        statement,
        SqlStatement::Ddl(SqlDdlStatement::DropIndex(SqlDropIndexStatement {
            name: "user_age_idx".to_string(),
            entity: None,
            if_exists: false,
            schema_version_contract: SqlDdlSchemaVersionContract::default(),
        })),
    );
}

#[test]
fn parse_ddl_idempotency_clauses_keep_explicit_intent() {
    let statement = parse_sql("CREATE INDEX IF NOT EXISTS user_age_idx ON users (age)")
        .expect("CREATE INDEX IF NOT EXISTS should parse");
    assert_eq!(
        statement,
        SqlStatement::Ddl(SqlDdlStatement::CreateIndex(SqlCreateIndexStatement {
            name: "user_age_idx".to_string(),
            entity: "users".to_string(),
            key_items: ddl_field_paths(&["age"]),
            predicate_sql: None,
            uniqueness: SqlCreateIndexUniqueness::NonUnique,
            if_not_exists: true,
            schema_version_contract: SqlDdlSchemaVersionContract::default(),
        })),
    );

    let statement =
        parse_sql("DROP INDEX IF EXISTS user_age_idx ON users").expect("DROP INDEX IF EXISTS");
    assert_eq!(
        statement,
        SqlStatement::Ddl(SqlDdlStatement::DropIndex(SqlDropIndexStatement {
            name: "user_age_idx".to_string(),
            entity: Some("users".to_string()),
            if_exists: true,
            schema_version_contract: SqlDdlSchemaVersionContract::default(),
        })),
    );

    let statement = parse_sql("DROP INDEX IF EXISTS user_age_idx")
        .expect("DROP INDEX IF EXISTS without ON should parse");
    assert_eq!(
        statement,
        SqlStatement::Ddl(SqlDdlStatement::DropIndex(SqlDropIndexStatement {
            name: "user_age_idx".to_string(),
            entity: None,
            if_exists: true,
            schema_version_contract: SqlDdlSchemaVersionContract::default(),
        })),
    );
}

#[test]
fn parse_ddl_schema_version_contracts_keep_transition_intent() {
    let statement = parse_sql(
        "ALTER TABLE users EXPECT SCHEMA VERSION 3 SET SCHEMA VERSION 4 ADD COLUMN nickname text",
    )
    .expect("ALTER TABLE should parse a prefix schema-version contract");
    assert_eq!(
        statement,
        SqlStatement::Ddl(SqlDdlStatement::AlterTableAddColumn(
            SqlAlterTableAddColumnStatement {
                entity: "users".to_string(),
                column_name: "nickname".to_string(),
                column_type: "text".to_string(),
                nullable: true,
                default: None,
                schema_version_contract: SqlDdlSchemaVersionContract {
                    expected_schema_version: Some(3),
                    next_schema_version: Some(4),
                },
            },
        )),
    );

    let statement = parse_sql(
        "ALTER TABLE users SET SCHEMA VERSION 4 EXPECT SCHEMA VERSION 3 ADD COLUMN nickname text",
    )
    .expect("ALTER TABLE should preserve a flexible prefix schema-version contract order");
    assert_eq!(
        statement,
        SqlStatement::Ddl(SqlDdlStatement::AlterTableAddColumn(
            SqlAlterTableAddColumnStatement {
                entity: "users".to_string(),
                column_name: "nickname".to_string(),
                column_type: "text".to_string(),
                nullable: true,
                default: None,
                schema_version_contract: SqlDdlSchemaVersionContract {
                    expected_schema_version: Some(3),
                    next_schema_version: Some(4),
                },
            },
        )),
    );

    let statement = parse_sql(
        "CREATE INDEX user_age_idx ON users (age) EXPECT SCHEMA VERSION 4 SET SCHEMA VERSION 5",
    )
    .expect("CREATE INDEX should parse a trailing schema-version contract");
    assert_eq!(
        statement,
        SqlStatement::Ddl(SqlDdlStatement::CreateIndex(SqlCreateIndexStatement {
            name: "user_age_idx".to_string(),
            entity: "users".to_string(),
            key_items: ddl_field_paths(&["age"]),
            predicate_sql: None,
            uniqueness: SqlCreateIndexUniqueness::NonUnique,
            if_not_exists: false,
            schema_version_contract: SqlDdlSchemaVersionContract {
                expected_schema_version: Some(4),
                next_schema_version: Some(5),
            },
        })),
    );
}

#[test]
fn parse_ddl_schema_version_contracts_reject_duplicates_and_non_literals() {
    for (sql, expected_feature) in [
        (
            "ALTER TABLE users EXPECT SCHEMA VERSION 3 EXPECT SCHEMA VERSION 4 ADD COLUMN nickname text",
            SqlFeatureCode::DdlSchemaVersionDuplicateExpectedClause,
        ),
        (
            "ALTER TABLE users SET SCHEMA VERSION 4 SET SCHEMA VERSION 5 ADD COLUMN nickname text",
            SqlFeatureCode::DdlSchemaVersionDuplicateSetClause,
        ),
        (
            "ALTER TABLE users EXPECT SCHEMA VERSION 3 ADD COLUMN nickname text EXPECT SCHEMA VERSION 4",
            SqlFeatureCode::DdlSchemaVersionDuplicateExpectedClause,
        ),
        (
            "CREATE INDEX user_age_idx ON users (age) SET SCHEMA VERSION 4 SET SCHEMA VERSION 5",
            SqlFeatureCode::DdlSchemaVersionDuplicateSetClause,
        ),
    ] {
        let err = parse_sql(sql).expect_err("duplicate DDL schema-version clauses should reject");
        assert_eq!(
            err,
            super::SqlParseError::UnsupportedFeature {
                feature: expected_feature
            },
            "duplicate DDL schema-version clause should keep a stable feature code: {sql}",
        );
    }

    for sql in [
        "ALTER TABLE users EXPECT SCHEMA VERSION ? SET SCHEMA VERSION 4 ADD COLUMN nickname text",
        "CREATE INDEX user_age_idx ON users (age) EXPECT SCHEMA VERSION 4 SET SCHEMA VERSION ?",
    ] {
        let err = parse_sql(sql).expect_err("non-literal DDL schema-version clauses should reject");
        assert!(
            matches!(err, super::SqlParseError::InvalidSyntax { .. }),
            "non-literal DDL schema-version clause should reject as syntax: {err:?}",
        );
    }
}

#[test]
fn parse_select_statement_with_qualified_identifiers() {
    let statement = parse_sql(
        "SELECT users.name, users.age \
         FROM public.users \
         WHERE users.age >= 21 \
         ORDER BY users.age DESC LIMIT 10 OFFSET 1",
    )
    .expect("qualified-identifier select statement should parse");

    assert_eq!(
        statement,
        SqlStatement::Select(SqlSelectStatement {
            entity: "public.users".to_string(),
            table_alias: None,
            projection: SqlProjection::Items(vec![
                SqlSelectItem::Field("users.name".to_string()),
                SqlSelectItem::Field("users.age".to_string()),
            ]),
            projection_aliases: vec![None, None],
            predicate: option_sql_pred!(Predicate::Compare(ComparePredicate::with_coercion(
                "users.age",
                CompareOp::Gte,
                Value::Int64(21),
                CoercionId::NumericWiden,
            ))),
            distinct: false,
            group_by: vec![],
            having: vec![],
            order_by: vec![SqlOrderTerm {
                field: sql_order_expr("users.age"),
                direction: SqlOrderDirection::Desc,
            }],
            limit: Some(10),
            offset: Some(1),
        }),
    );
}

#[test]
fn parse_select_statement_with_strict_like_prefix_predicate() {
    let statement = parse_sql(
        "SELECT * FROM users \
         WHERE name LIKE 'Al%' \
         ORDER BY id ASC LIMIT 1",
    )
    .expect("strict LIKE prefix select statement should parse");

    assert_eq!(
        statement,
        SqlStatement::Select(SqlSelectStatement {
            entity: "users".to_string(),
            table_alias: None,
            projection: SqlProjection::All,
            projection_aliases: Vec::default(),
            predicate: Some(sql_like_expr(
                SqlExpr::Field("name".to_string()),
                "Al%",
                false,
                false,
            )),
            distinct: false,
            group_by: vec![],
            having: vec![],
            order_by: vec![SqlOrderTerm {
                field: sql_order_expr("id"),
                direction: SqlOrderDirection::Asc,
            }],
            limit: Some(1),
            offset: None,
        }),
    );
}

#[test]
fn parse_select_statement_with_angle_bracket_not_equal_predicate() {
    let statement = parse_sql(
        "SELECT * FROM users \
         WHERE active <> true \
         ORDER BY id ASC LIMIT 1",
    )
    .expect("angle-bracket not-equal select statement should parse");

    assert_eq!(
        statement,
        SqlStatement::Select(SqlSelectStatement {
            entity: "users".to_string(),
            table_alias: None,
            projection: SqlProjection::All,
            projection_aliases: Vec::default(),
            predicate: option_sql_pred!(Predicate::Compare(ComparePredicate::with_coercion(
                "active",
                CompareOp::Ne,
                Value::Bool(true),
                CoercionId::Strict,
            ))),
            distinct: false,
            group_by: vec![],
            having: vec![],
            order_by: vec![SqlOrderTerm {
                field: sql_order_expr("id"),
                direction: SqlOrderDirection::Asc,
            }],
            limit: Some(1),
            offset: None,
        }),
    );
}

#[test]
fn parse_select_statement_with_in_trailing_comma_predicate() {
    let statement = parse_sql(
        "SELECT * FROM users \
         WHERE age IN (10, 20, 30,) \
         ORDER BY id ASC LIMIT 1",
    )
    .expect("IN with trailing comma select statement should parse");

    assert_eq!(
        statement,
        SqlStatement::Select(SqlSelectStatement {
            entity: "users".to_string(),
            table_alias: None,
            projection: SqlProjection::All,
            projection_aliases: Vec::default(),
            predicate: option_sql_pred!(Predicate::Compare(ComparePredicate::with_coercion(
                "age",
                CompareOp::In,
                Value::List(vec![Value::Int64(10), Value::Int64(20), Value::Int64(30)]),
                CoercionId::Strict,
            ))),
            distinct: false,
            group_by: vec![],
            having: vec![],
            order_by: vec![SqlOrderTerm {
                field: sql_order_expr("id"),
                direction: SqlOrderDirection::Asc,
            }],
            limit: Some(1),
            offset: None,
        }),
    );
}

#[test]
fn parse_select_statement_with_is_true_and_is_false_predicates() {
    let is_true = parse_sql(
        "SELECT * FROM users \
         WHERE active IS TRUE \
         ORDER BY id ASC LIMIT 1",
    )
    .expect("IS TRUE select statement should parse");
    let is_false = parse_sql(
        "SELECT * FROM users \
         WHERE active IS FALSE \
         ORDER BY id ASC LIMIT 1",
    )
    .expect("IS FALSE select statement should parse");

    assert_eq!(
        is_true,
        SqlStatement::Select(SqlSelectStatement {
            entity: "users".to_string(),
            table_alias: None,
            projection: SqlProjection::All,
            projection_aliases: Vec::default(),
            predicate: option_sql_pred!(Predicate::Compare(ComparePredicate::with_coercion(
                "active",
                CompareOp::Eq,
                Value::Bool(true),
                CoercionId::Strict,
            ))),
            distinct: false,
            group_by: vec![],
            having: vec![],
            order_by: vec![SqlOrderTerm {
                field: sql_order_expr("id"),
                direction: SqlOrderDirection::Asc,
            }],
            limit: Some(1),
            offset: None,
        }),
    );
    assert_eq!(
        is_false,
        SqlStatement::Select(SqlSelectStatement {
            entity: "users".to_string(),
            table_alias: None,
            projection: SqlProjection::All,
            projection_aliases: Vec::default(),
            predicate: option_sql_pred!(Predicate::Compare(ComparePredicate::with_coercion(
                "active",
                CompareOp::Eq,
                Value::Bool(false),
                CoercionId::Strict,
            ))),
            distinct: false,
            group_by: vec![],
            having: vec![],
            order_by: vec![SqlOrderTerm {
                field: sql_order_expr("id"),
                direction: SqlOrderDirection::Asc,
            }],
            limit: Some(1),
            offset: None,
        }),
    );
}

#[test]
fn parse_select_statement_with_is_not_true_and_is_not_false_predicates() {
    let is_not_true = parse_sql(
        "SELECT * FROM users \
         WHERE active IS NOT TRUE \
         ORDER BY id ASC LIMIT 1",
    )
    .expect("IS NOT TRUE select statement should parse");
    let is_not_false = parse_sql(
        "SELECT * FROM users \
         WHERE active IS NOT FALSE \
         ORDER BY id ASC LIMIT 1",
    )
    .expect("IS NOT FALSE select statement should parse");

    assert_eq!(
        is_not_true,
        SqlStatement::Select(SqlSelectStatement {
            entity: "users".to_string(),
            table_alias: None,
            projection: SqlProjection::All,
            projection_aliases: Vec::default(),
            predicate: option_sql_pred!(Predicate::Not(Box::new(Predicate::Compare(
                ComparePredicate::with_coercion(
                    "active",
                    CompareOp::Eq,
                    Value::Bool(true),
                    CoercionId::Strict,
                ),
            )))),
            distinct: false,
            group_by: vec![],
            having: vec![],
            order_by: vec![SqlOrderTerm {
                field: sql_order_expr("id"),
                direction: SqlOrderDirection::Asc,
            }],
            limit: Some(1),
            offset: None,
        }),
    );
    assert_eq!(
        is_not_false,
        SqlStatement::Select(SqlSelectStatement {
            entity: "users".to_string(),
            table_alias: None,
            projection: SqlProjection::All,
            projection_aliases: Vec::default(),
            predicate: option_sql_pred!(Predicate::Not(Box::new(Predicate::Compare(
                ComparePredicate::with_coercion(
                    "active",
                    CompareOp::Eq,
                    Value::Bool(false),
                    CoercionId::Strict,
                ),
            )))),
            distinct: false,
            group_by: vec![],
            having: vec![],
            order_by: vec![SqlOrderTerm {
                field: sql_order_expr("id"),
                direction: SqlOrderDirection::Asc,
            }],
            limit: Some(1),
            offset: None,
        }),
    );
}

#[test]
fn parse_select_statement_with_field_bound_between_and_not_between_predicates() {
    let between = parse_sql(
        "SELECT * FROM users \
         WHERE age BETWEEN min_age AND max_age \
         ORDER BY id ASC LIMIT 1",
    )
    .expect("field-bound BETWEEN select statement should parse");
    let not_between = parse_sql(
        "SELECT * FROM users \
         WHERE age NOT BETWEEN min_age AND max_age \
         ORDER BY id ASC LIMIT 1",
    )
    .expect("field-bound NOT BETWEEN select statement should parse");

    assert_eq!(
        between,
        SqlStatement::Select(SqlSelectStatement {
            entity: "users".to_string(),
            table_alias: None,
            projection: SqlProjection::All,
            projection_aliases: Vec::default(),
            predicate: option_sql_pred!(Predicate::And(vec![
                Predicate::CompareFields(CompareFieldsPredicate::with_coercion(
                    "age",
                    CompareOp::Gte,
                    "min_age",
                    CoercionId::NumericWiden,
                )),
                Predicate::CompareFields(CompareFieldsPredicate::with_coercion(
                    "age",
                    CompareOp::Lte,
                    "max_age",
                    CoercionId::NumericWiden,
                )),
            ])),
            distinct: false,
            group_by: vec![],
            having: vec![],
            order_by: vec![SqlOrderTerm {
                field: sql_order_expr("id"),
                direction: SqlOrderDirection::Asc,
            }],
            limit: Some(1),
            offset: None,
        }),
    );
    assert_eq!(
        not_between,
        SqlStatement::Select(SqlSelectStatement {
            entity: "users".to_string(),
            table_alias: None,
            projection: SqlProjection::All,
            projection_aliases: Vec::default(),
            predicate: option_sql_pred!(Predicate::Or(vec![
                Predicate::CompareFields(CompareFieldsPredicate::with_coercion(
                    "age",
                    CompareOp::Lt,
                    "min_age",
                    CoercionId::NumericWiden,
                )),
                Predicate::CompareFields(CompareFieldsPredicate::with_coercion(
                    "age",
                    CompareOp::Gt,
                    "max_age",
                    CoercionId::NumericWiden,
                )),
            ])),
            distinct: false,
            group_by: vec![],
            having: vec![],
            order_by: vec![SqlOrderTerm {
                field: sql_order_expr("id"),
                direction: SqlOrderDirection::Asc,
            }],
            limit: Some(1),
            offset: None,
        }),
    );
}

#[test]
fn parse_select_statement_with_strict_not_like_prefix_predicate() {
    let statement = parse_sql(
        "SELECT * FROM users \
         WHERE name NOT LIKE 'Al%' \
         ORDER BY id ASC LIMIT 1",
    )
    .expect("strict NOT LIKE prefix select statement should parse");

    assert_eq!(
        statement,
        SqlStatement::Select(SqlSelectStatement {
            entity: "users".to_string(),
            table_alias: None,
            projection: SqlProjection::All,
            projection_aliases: Vec::default(),
            predicate: Some(sql_like_expr(
                SqlExpr::Field("name".to_string()),
                "Al%",
                true,
                false,
            )),
            distinct: false,
            group_by: vec![],
            having: vec![],
            order_by: vec![SqlOrderTerm {
                field: sql_order_expr("id"),
                direction: SqlOrderDirection::Asc,
            }],
            limit: Some(1),
            offset: None,
        }),
    );
}

#[test]
fn parse_select_statement_with_ilike_prefix_predicate() {
    let statement = parse_sql(
        "SELECT * FROM users \
         WHERE name ILIKE 'al%' \
         ORDER BY id ASC LIMIT 1",
    )
    .expect("ILIKE prefix select statement should parse");

    assert_eq!(
        statement,
        SqlStatement::Select(SqlSelectStatement {
            entity: "users".to_string(),
            table_alias: None,
            projection: SqlProjection::All,
            projection_aliases: Vec::default(),
            predicate: Some(sql_like_expr(
                SqlExpr::Field("name".to_string()),
                "al%",
                false,
                true,
            )),
            distinct: false,
            group_by: vec![],
            having: vec![],
            order_by: vec![SqlOrderTerm {
                field: sql_order_expr("id"),
                direction: SqlOrderDirection::Asc,
            }],
            limit: Some(1),
            offset: None,
        }),
    );
}

#[test]
fn parse_select_statement_with_not_ilike_prefix_predicate() {
    let statement = parse_sql(
        "SELECT * FROM users \
         WHERE name NOT ILIKE 'al%' \
         ORDER BY id ASC LIMIT 1",
    )
    .expect("NOT ILIKE prefix select statement should parse");

    assert_eq!(
        statement,
        SqlStatement::Select(SqlSelectStatement {
            entity: "users".to_string(),
            table_alias: None,
            projection: SqlProjection::All,
            projection_aliases: Vec::default(),
            predicate: Some(sql_like_expr(
                SqlExpr::Field("name".to_string()),
                "al%",
                true,
                true,
            )),
            distinct: false,
            group_by: vec![],
            having: vec![],
            order_by: vec![SqlOrderTerm {
                field: sql_order_expr("id"),
                direction: SqlOrderDirection::Asc,
            }],
            limit: Some(1),
            offset: None,
        }),
    );
}

#[test]
fn parse_select_statement_with_strict_text_range_predicate() {
    let statement = parse_sql(
        "SELECT * FROM users \
         WHERE name >= 'Al' AND name < 'Am' \
         ORDER BY id ASC LIMIT 1",
    )
    .expect("strict text-range select statement should parse");

    assert_eq!(
        statement,
        SqlStatement::Select(SqlSelectStatement {
            entity: "users".to_string(),
            table_alias: None,
            projection: SqlProjection::All,
            projection_aliases: Vec::default(),
            predicate: option_sql_pred!(Predicate::And(vec![
                Predicate::Compare(ComparePredicate::with_coercion(
                    "name",
                    CompareOp::Gte,
                    Value::Text("Al".to_string()),
                    CoercionId::Strict,
                )),
                Predicate::Compare(ComparePredicate::with_coercion(
                    "name",
                    CompareOp::Lt,
                    Value::Text("Am".to_string()),
                    CoercionId::Strict,
                )),
            ])),
            distinct: false,
            group_by: vec![],
            having: vec![],
            order_by: vec![SqlOrderTerm {
                field: sql_order_expr("id"),
                direction: SqlOrderDirection::Asc,
            }],
            limit: Some(1),
            offset: None,
        }),
    );
}

#[test]
fn parse_select_statement_with_direct_starts_with_predicate() {
    let statement = parse_sql(
        "SELECT * FROM users \
         WHERE STARTS_WITH(name, 'Al') \
         ORDER BY id ASC LIMIT 1",
    )
    .expect("direct STARTS_WITH select statement should parse");

    assert_eq!(
        statement,
        SqlStatement::Select(SqlSelectStatement {
            entity: "users".to_string(),
            table_alias: None,
            projection: SqlProjection::All,
            projection_aliases: Vec::default(),
            predicate: option_sql_pred!(Predicate::Compare(ComparePredicate::with_coercion(
                "name",
                CompareOp::StartsWith,
                Value::Text("Al".to_string()),
                CoercionId::Strict,
            ))),
            distinct: false,
            group_by: vec![],
            having: vec![],
            order_by: vec![SqlOrderTerm {
                field: sql_order_expr("id"),
                direction: SqlOrderDirection::Asc,
            }],
            limit: Some(1),
            offset: None,
        }),
    );
}

#[test]
fn parse_select_statement_with_direct_lower_starts_with_predicate() {
    let statement = parse_sql(
        "SELECT * FROM users \
         WHERE STARTS_WITH(LOWER(name), 'Al') \
         ORDER BY id ASC LIMIT 1",
    )
    .expect("direct LOWER(field) STARTS_WITH select statement should parse");

    assert_eq!(
        statement,
        SqlStatement::Select(SqlSelectStatement {
            entity: "users".to_string(),
            table_alias: None,
            projection: SqlProjection::All,
            projection_aliases: Vec::default(),
            predicate: option_sql_pred!(Predicate::Compare(ComparePredicate::with_coercion(
                "name",
                CompareOp::StartsWith,
                Value::Text("Al".to_string()),
                CoercionId::TextCasefold,
            ))),
            distinct: false,
            group_by: vec![],
            having: vec![],
            order_by: vec![SqlOrderTerm {
                field: sql_order_expr("id"),
                direction: SqlOrderDirection::Asc,
            }],
            limit: Some(1),
            offset: None,
        }),
    );
}

#[test]
fn parse_select_statement_with_direct_upper_starts_with_predicate() {
    let statement = parse_sql(
        "SELECT * FROM users \
         WHERE STARTS_WITH(UPPER(name), 'AL') \
         ORDER BY id ASC LIMIT 1",
    )
    .expect("direct UPPER(field) STARTS_WITH select statement should parse");

    assert_eq!(
        statement,
        SqlStatement::Select(SqlSelectStatement {
            entity: "users".to_string(),
            table_alias: None,
            projection: SqlProjection::All,
            projection_aliases: Vec::default(),
            predicate: Some(SqlExpr::FunctionCall {
                function: SqlScalarFunction::StartsWith,
                args: vec![
                    sql_scalar_function_field_expr(SqlScalarFunction::Upper, "name"),
                    SqlExpr::Literal(Value::Text("AL".to_string())),
                ],
            }),
            distinct: false,
            group_by: vec![],
            having: vec![],
            order_by: vec![SqlOrderTerm {
                field: sql_order_expr("id"),
                direction: SqlOrderDirection::Asc,
            }],
            limit: Some(1),
            offset: None,
        }),
    );
}

#[test]
fn parse_select_statement_with_lower_like_prefix_predicate() {
    let statement = parse_sql(
        "SELECT * FROM users \
         WHERE LOWER(name) LIKE 'Al%' \
         ORDER BY id ASC LIMIT 1",
    )
    .expect("LOWER(field) LIKE prefix select statement should parse");

    assert_eq!(
        statement,
        SqlStatement::Select(SqlSelectStatement {
            entity: "users".to_string(),
            table_alias: None,
            projection: SqlProjection::All,
            projection_aliases: Vec::default(),
            predicate: Some(sql_like_expr(
                sql_scalar_function_field_expr(SqlScalarFunction::Lower, "name"),
                "Al%",
                false,
                false,
            )),
            distinct: false,
            group_by: vec![],
            having: vec![],
            order_by: vec![SqlOrderTerm {
                field: sql_order_expr("id"),
                direction: SqlOrderDirection::Asc,
            }],
            limit: Some(1),
            offset: None,
        }),
    );
}

#[test]
fn parse_select_statement_with_lower_not_like_prefix_predicate() {
    let statement = parse_sql(
        "SELECT * FROM users \
         WHERE LOWER(name) NOT LIKE 'Al%' \
         ORDER BY id ASC LIMIT 1",
    )
    .expect("LOWER(field) NOT LIKE prefix select statement should parse");

    assert_eq!(
        statement,
        SqlStatement::Select(SqlSelectStatement {
            entity: "users".to_string(),
            table_alias: None,
            projection: SqlProjection::All,
            projection_aliases: Vec::default(),
            predicate: Some(sql_like_expr(
                sql_scalar_function_field_expr(SqlScalarFunction::Lower, "name"),
                "Al%",
                true,
                false,
            )),
            distinct: false,
            group_by: vec![],
            having: vec![],
            order_by: vec![SqlOrderTerm {
                field: sql_order_expr("id"),
                direction: SqlOrderDirection::Asc,
            }],
            limit: Some(1),
            offset: None,
        }),
    );
}

#[test]
fn parse_select_statement_with_upper_like_prefix_predicate() {
    let statement = parse_sql(
        "SELECT * FROM users \
         WHERE UPPER(name) LIKE 'AL%' \
         ORDER BY id ASC LIMIT 1",
    )
    .expect("UPPER(field) LIKE prefix select statement should parse");

    assert_eq!(
        statement,
        SqlStatement::Select(SqlSelectStatement {
            entity: "users".to_string(),
            table_alias: None,
            projection: SqlProjection::All,
            projection_aliases: Vec::default(),
            predicate: Some(sql_like_expr(
                sql_scalar_function_field_expr(SqlScalarFunction::Upper, "name"),
                "AL%",
                false,
                false,
            )),
            distinct: false,
            group_by: vec![],
            having: vec![],
            order_by: vec![SqlOrderTerm {
                field: sql_order_expr("id"),
                direction: SqlOrderDirection::Asc,
            }],
            limit: Some(1),
            offset: None,
        }),
    );
}

#[test]
fn parse_select_statement_keeps_like_pattern_semantics_for_lowering() {
    let cases = [
        (
            "SELECT * FROM users WHERE name LIKE '%Al'",
            sql_like_expr(SqlExpr::Field("name".to_string()), "%Al", false, false),
        ),
        (
            "SELECT * FROM users WHERE LOWER(name) LIKE '%Al'",
            sql_like_expr(
                sql_scalar_function_field_expr(SqlScalarFunction::Lower, "name"),
                "%Al",
                false,
                false,
            ),
        ),
        (
            "SELECT * FROM users WHERE UPPER(name) LIKE '%Al'",
            sql_like_expr(
                sql_scalar_function_field_expr(SqlScalarFunction::Upper, "name"),
                "%Al",
                false,
                false,
            ),
        ),
    ];

    for (sql, expected_predicate) in cases {
        let SqlStatement::Select(statement) =
            parse_sql(sql).expect("parser should preserve LIKE pattern for lowering")
        else {
            panic!("LIKE pattern SQL should parse as SELECT");
        };

        assert_eq!(statement.predicate, Some(expected_predicate));
    }
}

#[test]
fn parse_select_grouped_statement_with_qualified_identifiers() {
    let statement = parse_sql(
        "SELECT users.age, COUNT(*) \
         FROM public.users \
         WHERE users.age >= 21 \
         GROUP BY users.age \
         ORDER BY users.age DESC LIMIT 5 OFFSET 1",
    )
    .expect("qualified-identifier grouped select statement should parse");

    assert_eq!(
        statement,
        SqlStatement::Select(SqlSelectStatement {
            entity: "public.users".to_string(),
            table_alias: None,
            projection: SqlProjection::Items(vec![
                SqlSelectItem::Field("users.age".to_string()),
                SqlSelectItem::Aggregate(SqlAggregateCall {
                    kind: SqlAggregateKind::Count,
                    input: None,
                    filter_expr: None,
                    distinct: false,
                }),
            ]),
            projection_aliases: vec![None, None],
            predicate: option_sql_pred!(Predicate::Compare(ComparePredicate::with_coercion(
                "users.age",
                CompareOp::Gte,
                Value::Int64(21),
                CoercionId::NumericWiden,
            ))),
            distinct: false,
            group_by: vec!["users.age".to_string()],
            having: vec![],
            order_by: vec![SqlOrderTerm {
                field: sql_order_expr("users.age"),
                direction: SqlOrderDirection::Desc,
            }],
            limit: Some(5),
            offset: Some(1),
        }),
    );
}

#[test]
#[cfg(feature = "sql-explain")]
fn parse_explain_execution_with_qualified_identifiers() {
    let statement = parse_sql(
        "EXPLAIN EXECUTION SELECT users.name FROM public.users \
         WHERE users.age >= 21 ORDER BY users.age DESC LIMIT 1",
    )
    .expect("qualified-identifier explain statement should parse");

    assert_eq!(
        statement,
        SqlStatement::Explain(SqlExplainStatement {
            mode: SqlExplainMode::Execution,
            verbose: false,
            statement: SqlExplainTarget::Select(SqlSelectStatement {
                entity: "public.users".to_string(),
                table_alias: None,
                projection: SqlProjection::Items(vec![SqlSelectItem::Field(
                    "users.name".to_string(),
                )]),
                projection_aliases: vec![None],
                predicate: option_sql_pred!(Predicate::Compare(ComparePredicate::with_coercion(
                    "users.age",
                    CompareOp::Gte,
                    Value::Int64(21),
                    CoercionId::NumericWiden,
                ))),
                distinct: false,
                group_by: vec![],
                having: vec![],
                order_by: vec![SqlOrderTerm {
                    field: sql_order_expr("users.age"),
                    direction: SqlOrderDirection::Desc,
                }],
                limit: Some(1),
                offset: None,
            }),
        }),
    );
}

#[test]
#[cfg(feature = "sql-explain")]
fn parse_explain_execution_verbose_with_qualified_identifiers() {
    let statement = parse_sql(
        "EXPLAIN EXECUTION VERBOSE SELECT users.name FROM public.users \
         WHERE users.age >= 21 ORDER BY users.age DESC LIMIT 1",
    )
    .expect("qualified-identifier verbose explain statement should parse");

    assert_eq!(
        statement,
        SqlStatement::Explain(SqlExplainStatement {
            mode: SqlExplainMode::Execution,
            verbose: true,
            statement: SqlExplainTarget::Select(SqlSelectStatement {
                entity: "public.users".to_string(),
                table_alias: None,
                projection: SqlProjection::Items(vec![SqlSelectItem::Field(
                    "users.name".to_string(),
                )]),
                projection_aliases: vec![None],
                predicate: option_sql_pred!(Predicate::Compare(ComparePredicate::with_coercion(
                    "users.age",
                    CompareOp::Gte,
                    Value::Int64(21),
                    CoercionId::NumericWiden,
                ))),
                distinct: false,
                group_by: vec![],
                having: vec![],
                order_by: vec![SqlOrderTerm {
                    field: sql_order_expr("users.age"),
                    direction: SqlOrderDirection::Desc,
                }],
                limit: Some(1),
                offset: None,
            }),
        }),
    );
}

#[test]
fn parse_select_grouped_statement_with_having_clauses() {
    let statement = parse_sql(
        "SELECT age, COUNT(*) \
         FROM users \
         GROUP BY age \
         HAVING age >= 21 AND COUNT(*) > 1 \
         ORDER BY age ASC LIMIT 10",
    )
    .expect("grouped HAVING select statement should parse");

    assert_eq!(
        statement,
        SqlStatement::Select(SqlSelectStatement {
            entity: "users".to_string(),
            table_alias: None,
            projection: SqlProjection::Items(vec![
                SqlSelectItem::Field("age".to_string()),
                SqlSelectItem::Aggregate(SqlAggregateCall {
                    kind: SqlAggregateKind::Count,
                    input: None,
                    filter_expr: None,
                    distinct: false,
                }),
            ]),
            projection_aliases: vec![None, None],
            predicate: None,
            distinct: false,
            group_by: vec!["age".to_string()],
            having: vec![SqlExpr::Binary {
                op: SqlExprBinaryOp::And,
                left: Box::new(SqlExpr::Binary {
                    op: SqlExprBinaryOp::Gte,
                    left: Box::new(SqlExpr::Field("age".to_string())),
                    right: Box::new(SqlExpr::Literal(Value::Int64(21))),
                }),
                right: Box::new(SqlExpr::Binary {
                    op: SqlExprBinaryOp::Gt,
                    left: Box::new(SqlExpr::Aggregate(SqlAggregateCall {
                        kind: SqlAggregateKind::Count,
                        input: None,
                        filter_expr: None,
                        distinct: false,
                    })),
                    right: Box::new(SqlExpr::Literal(Value::Int64(1))),
                }),
            }],
            order_by: vec![SqlOrderTerm {
                field: sql_order_expr("age"),
                direction: SqlOrderDirection::Asc,
            }],
            limit: Some(10),
            offset: None,
        }),
    );
}

#[test]
fn parse_select_grouped_statement_with_having_is_null_and_is_not_null_clauses() {
    let statement = parse_sql(
        "SELECT age, COUNT(*) \
         FROM users \
         GROUP BY age \
         HAVING age IS NOT NULL AND COUNT(*) IS NULL \
         ORDER BY age ASC LIMIT 10",
    )
    .expect("grouped HAVING IS [NOT] NULL clauses should parse");

    assert_eq!(
        statement,
        SqlStatement::Select(SqlSelectStatement {
            entity: "users".to_string(),
            table_alias: None,
            projection: SqlProjection::Items(vec![
                SqlSelectItem::Field("age".to_string()),
                SqlSelectItem::Aggregate(SqlAggregateCall {
                    kind: SqlAggregateKind::Count,
                    input: None,
                    filter_expr: None,
                    distinct: false,
                }),
            ]),
            projection_aliases: vec![None, None],
            predicate: None,
            distinct: false,
            group_by: vec!["age".to_string()],
            having: vec![SqlExpr::Binary {
                op: SqlExprBinaryOp::And,
                left: Box::new(SqlExpr::NullTest {
                    expr: Box::new(SqlExpr::Field("age".to_string())),
                    negated: true,
                }),
                right: Box::new(SqlExpr::NullTest {
                    expr: Box::new(SqlExpr::Aggregate(SqlAggregateCall {
                        kind: SqlAggregateKind::Count,
                        input: None,
                        filter_expr: None,
                        distinct: false,
                    })),
                    negated: false,
                }),
            }],
            order_by: vec![SqlOrderTerm {
                field: sql_order_expr("age"),
                direction: SqlOrderDirection::Asc,
            }],
            limit: Some(10),
            offset: None,
        }),
    );
}

#[test]
fn parse_select_grouped_statement_with_post_aggregate_having_exprs() {
    let statement = parse_sql(
        "SELECT class_name, AVG(strength) \
         FROM character \
         GROUP BY class_name \
         HAVING ROUND(AVG(strength), 2) >= 10 AND COUNT(*) + 1 > 5 \
         ORDER BY class_name ASC LIMIT 100",
    )
    .expect("grouped post-aggregate HAVING expressions should parse");

    let SqlStatement::Select(statement) = statement else {
        panic!("expected grouped SELECT statement");
    };

    assert_eq!(statement.having.len(), 1);
}

#[test]
fn parse_select_grouped_statement_with_coalesce_nullif_and_unary_numeric_having_exprs() {
    let statement = parse_sql(
        "SELECT age, COUNT(*) \
         FROM users \
         GROUP BY age \
         HAVING COALESCE(NULLIF(COUNT(*), 2), 99) = 99 AND ABS(AVG(age) - 10) = 0 \
         ORDER BY age ASC LIMIT 10",
    )
    .expect("grouped COALESCE/NULLIF and unary numeric HAVING expressions should parse");

    let SqlStatement::Select(statement) = statement else {
        panic!("expected grouped SELECT statement");
    };

    assert_eq!(statement.having.len(), 1);
}

#[test]
fn parse_select_grouped_statement_with_searched_case_having_exprs() {
    let statement = parse_sql(
        "SELECT age, COUNT(*) \
         FROM users \
         GROUP BY age \
         HAVING CASE WHEN COUNT(*) > 1 THEN 1 ELSE 0 END = 1 \
         ORDER BY age ASC LIMIT 10",
    )
    .expect("grouped searched CASE HAVING expressions should parse");

    assert_eq!(
        statement,
        SqlStatement::Select(SqlSelectStatement {
            entity: "users".to_string(),
            table_alias: None,
            projection: SqlProjection::Items(vec![
                SqlSelectItem::Field("age".to_string()),
                SqlSelectItem::Aggregate(SqlAggregateCall {
                    kind: SqlAggregateKind::Count,
                    input: None,
                    filter_expr: None,
                    distinct: false,
                }),
            ]),
            projection_aliases: vec![None, None],
            predicate: None,
            distinct: false,
            group_by: vec!["age".to_string()],
            having: vec![SqlExpr::Binary {
                op: SqlExprBinaryOp::Eq,
                left: Box::new(SqlExpr::Case {
                    arms: vec![SqlCaseArm {
                        condition: SqlExpr::Binary {
                            op: SqlExprBinaryOp::Gt,
                            left: Box::new(SqlExpr::Aggregate(SqlAggregateCall {
                                kind: SqlAggregateKind::Count,
                                input: None,
                                filter_expr: None,
                                distinct: false,
                            })),
                            right: Box::new(SqlExpr::Literal(Value::Int64(1))),
                        },
                        result: SqlExpr::Literal(Value::Int64(1)),
                    }],
                    else_expr: Some(Box::new(SqlExpr::Literal(Value::Int64(0)))),
                }),
                right: Box::new(SqlExpr::Literal(Value::Int64(1))),
            }],
            order_by: vec![SqlOrderTerm {
                field: sql_order_expr("age"),
                direction: SqlOrderDirection::Asc,
            }],
            limit: Some(10),
            offset: None,
        }),
        "searched CASE HAVING values should stay on the shared post-aggregate SQL-expression boundary",
    );
}

#[test]
fn parse_select_grouped_statement_with_aggregate_order_terms() {
    let statement = parse_sql(
        "SELECT age, AVG(score) \
         FROM users \
         GROUP BY age \
         ORDER BY AVG(score) DESC, ROUND(AVG(score), 2) ASC, age ASC \
         LIMIT 10",
    )
    .expect("grouped aggregate ORDER BY terms should parse");

    assert_eq!(
        statement,
        SqlStatement::Select(SqlSelectStatement {
            entity: "users".to_string(),
            table_alias: None,
            projection: SqlProjection::Items(vec![
                SqlSelectItem::Field("age".to_string()),
                SqlSelectItem::Aggregate(SqlAggregateCall {
                    kind: SqlAggregateKind::Avg,
                    input: Some(Box::new(SqlExpr::Field("score".to_string()))),
                    filter_expr: None,
                    distinct: false,
                }),
            ]),
            projection_aliases: vec![None, None],
            predicate: None,
            distinct: false,
            group_by: vec!["age".to_string()],
            having: vec![],
            order_by: vec![
                SqlOrderTerm {
                    field: sql_order_expr("AVG(score)"),
                    direction: SqlOrderDirection::Desc,
                },
                SqlOrderTerm {
                    field: sql_order_expr("ROUND(AVG(score), 2)"),
                    direction: SqlOrderDirection::Asc,
                },
                SqlOrderTerm {
                    field: sql_order_expr("age"),
                    direction: SqlOrderDirection::Asc,
                },
            ],
            limit: Some(10),
            offset: None,
        }),
    );
}

#[test]
fn parse_select_grouped_statement_with_wrapped_aggregate_order_terms() {
    let statement = parse_sql(
        "SELECT age, AVG(score) \
         FROM users \
         GROUP BY age \
         ORDER BY COALESCE(NULLIF(AVG(score), 40), 99) DESC, ABS(AVG(score) - 10) ASC, age ASC \
         LIMIT 10",
    )
    .expect("grouped wrapped aggregate ORDER BY terms should parse");

    let SqlStatement::Select(statement) = statement else {
        panic!("expected grouped SELECT statement");
    };

    assert_eq!(statement.order_by.len(), 3);
}

#[test]
fn parse_select_grouped_statement_accepts_having_is_true_for_post_aggregate_lowering() {
    let statement = parse_sql(
        "SELECT age, COUNT(*) \
         FROM users \
         GROUP BY age \
         HAVING COUNT(*) IS TRUE \
         ORDER BY age ASC LIMIT 10",
    )
    .expect("grouped HAVING IS TRUE should parse through the shared post-aggregate seam");

    assert!(
        matches!(
            statement,
            SqlStatement::Select(SqlSelectStatement {
                having,
                ..
            }) if matches!(
                having.as_slice(),
                [SqlExpr::Binary {
                    op: SqlExprBinaryOp::Eq,
                    left,
                    right
                }] if matches!(left.as_ref(), SqlExpr::Aggregate(_))
                    && matches!(right.as_ref(), SqlExpr::Literal(Value::Bool(true)))
            )
        ),
        "grouped HAVING IS TRUE should stay parser-owned syntax and defer semantic typing to lowering",
    );
}

#[test]
fn parse_sql_rejects_select_limit_before_order_with_actionable_message() {
    let err = parse_sql("SELECT * FROM users LIMIT 1 ORDER BY id")
        .expect_err("out-of-order LIMIT/ORDER clause should be rejected");

    assert_eq!(
        err,
        super::SqlParseError::InvalidSyntax {
            kind: SqlSyntaxErrorKind::ClauseOrder {
                rule: SqlClauseOrderRule::SelectOrderBeforeLimitOffset
            }
        }
    );
}

#[test]
fn parse_sql_rejects_select_offset_before_order_with_actionable_message() {
    let err = parse_sql("SELECT * FROM users OFFSET 1 ORDER BY id")
        .expect_err("out-of-order OFFSET/ORDER clause should be rejected");

    assert_eq!(
        err,
        super::SqlParseError::InvalidSyntax {
            kind: SqlSyntaxErrorKind::ClauseOrder {
                rule: SqlClauseOrderRule::SelectOrderBeforeLimitOffset
            }
        }
    );
}

#[test]
fn parse_sql_rejects_delete_limit_before_order_with_actionable_message() {
    let err = parse_sql("DELETE FROM users LIMIT 1 ORDER BY id")
        .expect_err("out-of-order DELETE LIMIT/ORDER clause should be rejected");

    assert_eq!(
        err,
        super::SqlParseError::InvalidSyntax {
            kind: SqlSyntaxErrorKind::ClauseOrder {
                rule: SqlClauseOrderRule::DeleteOrderBeforeLimit
            }
        }
    );
}

#[test]
fn parse_insert_statement_with_explicit_columns_and_values() {
    let statement = parse_sql("INSERT INTO users (id, name, age) VALUES (7, 'Ada', 21)")
        .expect("insert statement should parse");

    assert_eq!(
        statement,
        SqlStatement::Insert(SqlInsertStatement {
            entity: "users".to_string(),
            columns: vec!["id".to_string(), "name".to_string(), "age".to_string()],
            source: SqlInsertSource::Values(vec![vec![
                sql_write_literal(Value::Int64(7)),
                sql_write_literal(Value::Text("Ada".to_string())),
                sql_write_literal(Value::Int64(21)),
            ]]),
            returning: None,
        }),
    );
}

#[test]
fn parse_insert_statement_preserves_explicit_default_values() {
    let statement = parse_sql("INSERT INTO users (id, name) VALUES (DEFAULT, 'Ada')")
        .expect("insert DEFAULT should parse only as write intent");

    assert_eq!(
        statement,
        SqlStatement::Insert(SqlInsertStatement {
            entity: "users".to_string(),
            columns: vec!["id".to_string(), "name".to_string()],
            source: SqlInsertSource::Values(vec![vec![
                SqlWriteValue::Default,
                sql_write_literal(Value::Text("Ada".to_string())),
            ]]),
            returning: None,
        }),
    );
}

#[test]
fn parse_insert_statement_preserves_default_values_source() {
    let statement = parse_sql("INSERT INTO users DEFAULT VALUES")
        .expect("DEFAULT VALUES should parse as its own source");

    assert_eq!(
        statement,
        SqlStatement::Insert(SqlInsertStatement {
            entity: "users".to_string(),
            columns: Vec::new(),
            source: SqlInsertSource::DefaultValues,
            returning: None,
        }),
    );
}

#[test]
fn parse_insert_statement_with_hex_blob_literal_values() {
    let statement = parse_sql("INSERT INTO files (id, thumbnail) VALUES (7, X'0A0bFF')")
        .expect("insert statement with blob literal should parse");

    assert_eq!(
        statement,
        SqlStatement::Insert(SqlInsertStatement {
            entity: "files".to_string(),
            columns: vec!["id".to_string(), "thumbnail".to_string()],
            source: SqlInsertSource::Values(vec![vec![
                sql_write_literal(Value::Int64(7)),
                sql_write_literal(Value::Blob(vec![0x0A, 0x0B, 0xFF])),
            ]]),
            returning: None,
        }),
    );
}

#[test]
fn parse_insert_statement_with_multiple_values_tuples() {
    let statement =
        parse_sql("INSERT INTO users (id, name, age) VALUES (7, 'Ada', 21), (8, 'Bea', 22)")
            .expect("multi-row insert statement should parse");

    assert_eq!(
        statement,
        SqlStatement::Insert(SqlInsertStatement {
            entity: "users".to_string(),
            columns: vec!["id".to_string(), "name".to_string(), "age".to_string()],
            source: SqlInsertSource::Values(vec![
                vec![
                    sql_write_literal(Value::Int64(7)),
                    sql_write_literal(Value::Text("Ada".to_string())),
                    sql_write_literal(Value::Int64(21))
                ],
                vec![
                    sql_write_literal(Value::Int64(8)),
                    sql_write_literal(Value::Text("Bea".to_string())),
                    sql_write_literal(Value::Int64(22))
                ],
            ]),
            returning: None,
        }),
    );
}

#[test]
fn parse_update_statement_with_assignments_and_predicate() {
    let statement = parse_sql("UPDATE users SET name = 'Ada', age = 21 WHERE id = 7")
        .expect("update statement should parse");

    assert_eq!(
        statement,
        SqlStatement::Update(SqlUpdateStatement {
            entity: "users".to_string(),
            table_alias: None,
            assignments: vec![
                SqlAssignment {
                    field: "name".to_string(),
                    value: sql_write_literal(Value::Text("Ada".to_string())),
                },
                SqlAssignment {
                    field: "age".to_string(),
                    value: sql_write_literal(Value::Int64(21)),
                },
            ],
            predicate: option_sql_pred!(Predicate::Compare(ComparePredicate::with_coercion(
                "id",
                CompareOp::Eq,
                Value::Int64(7),
                CoercionId::Strict,
            ))),
            order_by: Vec::new(),
            limit: None,
            offset: None,
            returning: None,
        }),
    );
}

#[test]
fn parse_update_statement_preserves_explicit_default_assignment() {
    let statement = parse_sql("UPDATE users SET name = DEFAULT WHERE id = 7")
        .expect("update DEFAULT should parse only as assignment intent");

    let SqlStatement::Update(statement) = statement else {
        panic!("expected UPDATE statement");
    };
    assert_eq!(
        statement.assignments,
        vec![SqlAssignment {
            field: "name".to_string(),
            value: SqlWriteValue::Default,
        }]
    );
}

#[test]
fn parse_default_rejects_non_write_positions() {
    for sql in [
        "SELECT DEFAULT FROM users",
        "SELECT id FROM users WHERE id = DEFAULT",
        "INSERT INTO users (id) VALUES (DEFAULT + 1)",
        "UPDATE users SET name = COALESCE(DEFAULT, 'Ada') WHERE id = 7",
    ] {
        assert!(
            parse_sql(sql).is_err(),
            "DEFAULT outside a direct write position must reject: {sql}"
        );
    }
}

#[test]
fn parse_update_statement_accepts_single_table_alias() {
    let statement = parse_sql("UPDATE users u SET u.name = 'Ada', u.age = 21 WHERE u.id = 7")
        .expect("update statement with one table alias should parse");

    assert_eq!(
        statement,
        SqlStatement::Update(SqlUpdateStatement {
            entity: "users".to_string(),
            table_alias: Some("u".to_string()),
            assignments: vec![
                SqlAssignment {
                    field: "u.name".to_string(),
                    value: sql_write_literal(Value::Text("Ada".to_string())),
                },
                SqlAssignment {
                    field: "u.age".to_string(),
                    value: sql_write_literal(Value::Int64(21)),
                },
            ],
            predicate: option_sql_pred!(Predicate::Compare(ComparePredicate::with_coercion(
                "u.id",
                CompareOp::Eq,
                Value::Int64(7),
                CoercionId::Strict,
            ))),
            order_by: Vec::new(),
            limit: None,
            offset: None,
            returning: None,
        }),
    );
}

#[test]
fn parse_update_statement_with_order_limit_and_offset() {
    let statement = parse_sql(
        "UPDATE users SET age = 22 WHERE active = true ORDER BY age DESC, id ASC LIMIT 2 OFFSET 1",
    )
    .expect("update statement with ordered window should parse");

    assert_eq!(
        statement,
        SqlStatement::Update(SqlUpdateStatement {
            entity: "users".to_string(),
            table_alias: None,
            assignments: vec![SqlAssignment {
                field: "age".to_string(),
                value: sql_write_literal(Value::Int64(22)),
            }],
            predicate: option_sql_pred!(Predicate::Compare(ComparePredicate::with_coercion(
                "active",
                CompareOp::Eq,
                Value::Bool(true),
                CoercionId::Strict,
            ))),
            order_by: vec![
                SqlOrderTerm {
                    field: sql_order_expr("age"),
                    direction: SqlOrderDirection::Desc,
                },
                SqlOrderTerm {
                    field: sql_order_expr("id"),
                    direction: SqlOrderDirection::Asc,
                },
            ],
            limit: Some(2),
            offset: Some(1),
            returning: None,
        }),
    );
}

#[test]
fn parse_update_statement_rejects_invalid_window_clause_order() {
    let cases = [
        (
            "UPDATE users SET age = 22 WHERE id = 7 LIMIT 1 ORDER BY id",
            SqlClauseOrderRule::UpdateOrderBeforeLimitOffset,
        ),
        (
            "UPDATE users SET age = 22 WHERE id = 7 OFFSET 1 LIMIT 1",
            SqlClauseOrderRule::UpdateLimitBeforeOffset,
        ),
    ];

    for (sql, rule) in cases {
        let err = parse_sql(sql).expect_err("invalid UPDATE window clause order should fail");
        assert_eq!(
            err,
            SqlParseError::InvalidSyntax {
                kind: SqlSyntaxErrorKind::ClauseOrder { rule },
            },
            "invalid UPDATE window clause order should preserve a stable parser reason",
        );
    }
}

#[test]
fn parse_insert_statement_with_returning_field_list_parses() {
    let statement = parse_sql("INSERT INTO users (id, name) VALUES (1, 'Ada') RETURNING id, name")
        .expect("INSERT RETURNING field list should parse");

    assert_eq!(
        statement,
        SqlStatement::Insert(SqlInsertStatement {
            entity: "users".to_string(),
            columns: vec!["id".to_string(), "name".to_string()],
            source: SqlInsertSource::Values(vec![vec![
                sql_write_literal(Value::Int64(1)),
                sql_write_literal(Value::Text("Ada".to_string()))
            ]]),
            returning: Some(SqlReturningProjection::Fields(vec![
                "id".to_string(),
                "name".to_string(),
            ])),
        }),
    );
}

#[test]
fn parse_update_statement_with_returning_star_parses() {
    let statement =
        parse_sql("UPDATE users alias SET alias.name = 'Ada' WHERE alias.id = 1 RETURNING *")
            .expect("UPDATE RETURNING star should parse");

    assert_eq!(
        statement,
        SqlStatement::Update(SqlUpdateStatement {
            entity: "users".to_string(),
            table_alias: Some("alias".to_string()),
            assignments: vec![SqlAssignment {
                field: "alias.name".to_string(),
                value: sql_write_literal(Value::Text("Ada".to_string())),
            }],
            predicate: option_sql_pred!(Predicate::eq("alias.id".to_string(), Value::Int64(1))),
            order_by: vec![],
            limit: None,
            offset: None,
            returning: Some(SqlReturningProjection::All),
        }),
    );
}

#[test]
fn parse_delete_statement_with_returning_field_list_parses() {
    let statement =
        parse_sql("DELETE FROM users alias WHERE alias.id = 1 RETURNING alias.id, alias.name")
            .expect("DELETE RETURNING field list should parse");

    assert_eq!(
        statement,
        SqlStatement::Delete(SqlDeleteStatement {
            entity: "users".to_string(),
            table_alias: Some("alias".to_string()),
            predicate: option_sql_pred!(Predicate::eq("alias.id".to_string(), Value::Int64(1),)),
            order_by: vec![],
            limit: None,
            offset: None,
            returning: Some(SqlReturningProjection::Fields(vec![
                "alias.id".to_string(),
                "alias.name".to_string(),
            ])),
        }),
    );
}

#[test]
fn parse_delete_statement_with_returning_star_parses() {
    let statement = parse_sql("DELETE FROM users WHERE id = 1 RETURNING *")
        .expect("DELETE RETURNING star should parse");

    assert_eq!(
        statement,
        SqlStatement::Delete(SqlDeleteStatement {
            entity: "users".to_string(),
            table_alias: None,
            predicate: option_sql_pred!(Predicate::eq("id".to_string(), Value::Int64(1),)),
            order_by: vec![],
            limit: None,
            offset: None,
            returning: Some(SqlReturningProjection::All),
        }),
    );
}

#[test]
fn parse_insert_statement_without_column_list_parses() {
    let statement =
        parse_sql("INSERT INTO users VALUES (1)").expect("insert without column list should parse");

    assert_eq!(
        statement,
        SqlStatement::Insert(SqlInsertStatement {
            entity: "users".to_string(),
            columns: vec![],
            source: SqlInsertSource::Values(vec![vec![sql_write_literal(Value::Int64(1))]]),
            returning: None,
        }),
    );
}

#[test]
fn parse_insert_statement_with_field_only_select_source_parses() {
    let statement = parse_sql(
        "INSERT INTO users (name, age) SELECT name, age FROM users WHERE age >= 21 ORDER BY id ASC LIMIT 1",
    )
    .expect("insert-select should parse");

    assert_eq!(
        statement,
        SqlStatement::Insert(SqlInsertStatement {
            entity: "users".to_string(),
            columns: vec!["name".to_string(), "age".to_string()],
            source: SqlInsertSource::Select(Box::new(SqlSelectStatement {
                entity: "users".to_string(),
                table_alias: None,
                projection: SqlProjection::Items(vec![
                    SqlSelectItem::Field("name".to_string()),
                    SqlSelectItem::Field("age".to_string()),
                ]),
                projection_aliases: vec![None, None],
                predicate: option_sql_pred!(Predicate::Compare(ComparePredicate::with_coercion(
                    "age",
                    CompareOp::Gte,
                    Value::Int64(21),
                    CoercionId::NumericWiden,
                ))),
                distinct: false,
                group_by: Vec::new(),
                having: Vec::new(),
                order_by: vec![SqlOrderTerm {
                    field: sql_order_expr("id"),
                    direction: SqlOrderDirection::Asc,
                }],
                limit: Some(1),
                offset: None,
            })),
            returning: None,
        }),
    );
}

#[test]
fn parse_insert_statement_with_computed_select_source_parses() {
    let statement = parse_sql(
        "INSERT INTO users (name, age) \
         SELECT LOWER(name), age FROM users WHERE age >= 21 ORDER BY id ASC LIMIT 1",
    )
    .expect("insert-select with one admitted computed source projection should parse");

    assert_eq!(
        statement,
        SqlStatement::Insert(SqlInsertStatement {
            entity: "users".to_string(),
            columns: vec!["name".to_string(), "age".to_string()],
            source: SqlInsertSource::Select(Box::new(SqlSelectStatement {
                entity: "users".to_string(),
                table_alias: None,
                projection: SqlProjection::Items(vec![
                    sql_scalar_function_field_item(SqlScalarFunction::Lower, "name"),
                    SqlSelectItem::Field("age".to_string()),
                ]),
                projection_aliases: vec![None, None],
                predicate: option_sql_pred!(Predicate::Compare(ComparePredicate::with_coercion(
                    "age",
                    CompareOp::Gte,
                    Value::Int64(21),
                    CoercionId::NumericWiden,
                ))),
                distinct: false,
                group_by: Vec::new(),
                having: Vec::new(),
                order_by: vec![SqlOrderTerm {
                    field: sql_order_expr("id"),
                    direction: SqlOrderDirection::Asc,
                }],
                limit: Some(1),
                offset: None,
            })),
            returning: None,
        }),
    );
}

#[test]
fn parse_insert_statement_accepts_single_table_alias() {
    let statement = parse_sql("INSERT INTO users u (id, name) VALUES (1, 'Ada')")
        .expect("insert table alias should parse");

    assert_eq!(
        statement,
        SqlStatement::Insert(SqlInsertStatement {
            entity: "users".to_string(),
            columns: vec!["id".to_string(), "name".to_string()],
            source: SqlInsertSource::Values(vec![vec![
                sql_write_literal(Value::Int64(1)),
                sql_write_literal(Value::Text("Ada".to_string())),
            ]]),
            returning: None,
        }),
    );
}

#[test]
fn parse_insert_statement_accepts_as_table_alias_without_column_list() {
    let statement = parse_sql("INSERT INTO users AS u VALUES (1)")
        .expect("insert AS table alias without column list should parse");

    assert_eq!(
        statement,
        SqlStatement::Insert(SqlInsertStatement {
            entity: "users".to_string(),
            columns: vec![],
            source: SqlInsertSource::Values(vec![vec![sql_write_literal(Value::Int64(1))]]),
            returning: None,
        }),
    );
}

#[test]
fn parse_insert_statement_rejects_tuple_length_mismatch_in_any_values_tuple() {
    let err = parse_sql("INSERT INTO users (id, name, age) VALUES (7, 'Ada', 21), (8, 'Bea')")
        .expect_err("multi-row insert with tuple length mismatch should stay fail-closed");

    assert_eq!(
        err,
        super::SqlParseError::InvalidSyntax {
            kind: SqlSyntaxErrorKind::InsertValuesTupleLengthMismatch,
        }
    );
}

#[test]
fn parse_insert_statement_without_column_list_accepts_multiple_values_tuples() {
    let statement = parse_sql("INSERT INTO users VALUES (1, 'Ada', 21), (2, 'Bea', 22)")
        .expect("multi-row insert without column list should parse");

    assert_eq!(
        statement,
        SqlStatement::Insert(SqlInsertStatement {
            entity: "users".to_string(),
            columns: vec![],
            source: SqlInsertSource::Values(vec![
                vec![
                    sql_write_literal(Value::Int64(1)),
                    sql_write_literal(Value::Text("Ada".to_string())),
                    sql_write_literal(Value::Int64(21))
                ],
                vec![
                    sql_write_literal(Value::Int64(2)),
                    sql_write_literal(Value::Text("Bea".to_string())),
                    sql_write_literal(Value::Int64(22))
                ],
            ]),
            returning: None,
        }),
    );
}

#[test]
fn parse_sql_unsupported_feature_codes_are_stable() {
    let cases = [
        (
            "SELECT * FROM users JOIN other ON users.id = other.id",
            SqlFeatureCode::Join,
        ),
        (
            "WITH cte AS (SELECT * FROM users) SELECT * FROM cte",
            SqlFeatureCode::With,
        ),
        (
            "SELECT * FROM users UNION SELECT * FROM users",
            SqlFeatureCode::UnionIntersectExcept,
        ),
        (
            "SELECT * FROM users INTERSECT SELECT * FROM users",
            SqlFeatureCode::UnionIntersectExcept,
        ),
        (
            "SELECT * FROM users EXCEPT SELECT * FROM users",
            SqlFeatureCode::UnionIntersectExcept,
        ),
        #[cfg(feature = "sql-explain")]
        (
            "EXPLAIN INSERT INTO users VALUES (1)",
            SqlFeatureCode::Insert,
        ),
        (
            "SELECT * FROM users; SELECT * FROM users",
            SqlFeatureCode::MultiStatementSql,
        ),
        (
            "SELECT \"name\" FROM users",
            SqlFeatureCode::QuotedIdentifiers,
        ),
        (
            "SELECT len(name) FROM users",
            SqlFeatureCode::UnsupportedFunctionNamespace,
        ),
        (
            "SELECT ROW_NUMBER() OVER (ORDER BY age DESC) FROM users",
            SqlFeatureCode::WindowFunction,
        ),
        (
            "INSERT INTO users (id, name) VALUES (1, 'Ada') RETURNING LOWER(name)",
            SqlFeatureCode::UnsupportedFunctionNamespace,
        ),
        (
            "DESCRIBE users WHERE age > 1",
            SqlFeatureCode::DescribeModifier,
        ),
        #[cfg(feature = "sql-explain")]
        ("EXPLAIN DESCRIBE users", SqlFeatureCode::DescribeModifier),
        ("SHOW DATABASES", SqlFeatureCode::ShowUnsupportedCommand),
        (
            "SHOW INDEXES FROM users WHERE age > 1",
            SqlFeatureCode::ShowIndexesModifiers,
        ),
        (
            "SHOW CONSTRAINTS FROM users WHERE age > 1",
            SqlFeatureCode::ShowConstraintsModifiers,
        ),
        (
            "SHOW COLUMNS users WHERE age > 1",
            SqlFeatureCode::ShowColumnsModifiers,
        ),
        ("SHOW ENTITIES users", SqlFeatureCode::ShowEntitiesModifiers),
        ("SHOW STORES users", SqlFeatureCode::ShowStoresModifiers),
        ("SHOW MEMORY users", SqlFeatureCode::ShowMemoryModifiers),
        (
            "CREATE INDEX user_age_idx ON users (age DESC)",
            SqlFeatureCode::CreateIndexKeyOrderingModifiers,
        ),
    ];

    for (sql, expected_feature) in cases {
        let err = parse_sql(sql).expect_err("unsupported SQL feature should fail closed");
        assert_eq!(
            err,
            super::SqlParseError::UnsupportedFeature {
                feature: expected_feature
            },
            "unsupported feature code should stay stable for SQL: {sql}",
        );
    }
}

#[test]
fn parse_sql_accepts_wrapped_text_predicate_targets_in_where() {
    let statement =
        parse_sql("SELECT * FROM users WHERE STARTS_WITH(REPLACE(name, 'a', 'A'), 'Al')")
            .expect("wrapped text predicate target in WHERE should parse");

    let SqlStatement::Select(statement) = statement else {
        panic!("wrapped text predicate target should parse as SELECT");
    };

    assert_eq!(
        statement.predicate,
        Some(SqlExpr::FunctionCall {
            function: SqlScalarFunction::StartsWith,
            args: vec![
                SqlExpr::FunctionCall {
                    function: SqlScalarFunction::Replace,
                    args: vec![
                        SqlExpr::Field("name".to_string()),
                        SqlExpr::Literal(Value::Text("a".to_string())),
                        SqlExpr::Literal(Value::Text("A".to_string())),
                    ],
                },
                SqlExpr::Literal(Value::Text("Al".to_string())),
            ],
        }),
        "wrapped text predicate targets should stay on the shared WHERE expression seam",
    );
}

#[test]
fn parse_sql_accepts_text_predicate_expression_arguments_in_where() {
    let statement = parse_sql(
        "SELECT * FROM users \
         WHERE STARTS_WITH(REPLACE(name, 'a', 'A'), TRIM('Al'))",
    )
    .expect("text predicate expression arguments in WHERE should parse");

    let SqlStatement::Select(statement) = statement else {
        panic!("text predicate expression arguments should parse as SELECT");
    };

    assert_eq!(
        statement.predicate,
        Some(SqlExpr::FunctionCall {
            function: SqlScalarFunction::StartsWith,
            args: vec![
                SqlExpr::FunctionCall {
                    function: SqlScalarFunction::Replace,
                    args: vec![
                        SqlExpr::Field("name".to_string()),
                        SqlExpr::Literal(Value::Text("a".to_string())),
                        SqlExpr::Literal(Value::Text("A".to_string())),
                    ],
                },
                SqlExpr::FunctionCall {
                    function: SqlScalarFunction::Trim,
                    args: vec![SqlExpr::Literal(Value::Text("Al".to_string()))],
                },
            ],
        }),
        "text predicate expression arguments should stay on the shared WHERE expression seam",
    );
}

#[test]
fn parse_sql_accepts_wrapped_like_targets_in_where() {
    let strict_like = parse_sql("SELECT * FROM users WHERE REPLACE(name, 'a', 'A') LIKE 'Al%'")
        .expect("wrapped LIKE target in WHERE should parse");
    let casefold_like = parse_sql("SELECT * FROM users WHERE REPLACE(name, 'a', 'A') ILIKE 'al%'")
        .expect("wrapped ILIKE target in WHERE should parse");

    let SqlStatement::Select(strict_like) = strict_like else {
        panic!("wrapped LIKE target should parse as SELECT");
    };
    let SqlStatement::Select(casefold_like) = casefold_like else {
        panic!("wrapped ILIKE target should parse as SELECT");
    };

    let wrapped_replace = SqlExpr::FunctionCall {
        function: SqlScalarFunction::Replace,
        args: vec![
            SqlExpr::Field("name".to_string()),
            SqlExpr::Literal(Value::Text("a".to_string())),
            SqlExpr::Literal(Value::Text("A".to_string())),
        ],
    };

    assert_eq!(
        strict_like.predicate,
        Some(sql_like_expr(wrapped_replace.clone(), "Al%", false, false)),
        "wrapped LIKE targets should stay raw until expression lowering",
    );
    assert_eq!(
        casefold_like.predicate,
        Some(sql_like_expr(wrapped_replace, "al%", false, true)),
        "wrapped ILIKE targets should keep casefold intent as LIKE metadata",
    );
}

#[test]
fn parse_select_statement_rejects_simple_case_expressions() {
    let err = parse_sql("SELECT CASE age WHEN 21 THEN 'adult' ELSE 'minor' END FROM users")
        .expect_err("simple CASE expressions should stay fail-closed");

    assert_eq!(
        err,
        super::SqlParseError::UnsupportedFeature {
            feature: SqlFeatureCode::SimpleCaseExpression
        }
    );
}

#[test]
fn parse_sql_accepts_projection_aliases() {
    let statement = parse_sql(
        "SELECT name AS display_name, COUNT(*) total FROM users GROUP BY name ORDER BY name ASC",
    )
    .expect("projection aliases should parse");

    assert_eq!(
        statement,
        SqlStatement::Select(SqlSelectStatement {
            entity: "users".to_string(),
            table_alias: None,
            projection: SqlProjection::Items(vec![
                SqlSelectItem::Field("name".to_string()),
                SqlSelectItem::Aggregate(SqlAggregateCall {
                    kind: SqlAggregateKind::Count,
                    input: None,
                    filter_expr: None,
                    distinct: false,
                }),
            ]),
            projection_aliases: vec![Some("display_name".to_string()), Some("total".to_string())],
            predicate: None,
            distinct: false,
            group_by: vec!["name".to_string()],
            having: vec![],
            order_by: vec![SqlOrderTerm {
                field: sql_order_expr("name"),
                direction: SqlOrderDirection::Asc,
            }],
            limit: None,
            offset: None,
        }),
    );
}

#[test]
fn parse_sql_accepts_bare_projection_aliases() {
    let statement =
        parse_sql("SELECT TRIM(name) trimmed_name FROM users").expect("bare aliases should parse");

    assert_eq!(
        statement,
        SqlStatement::Select(SqlSelectStatement {
            entity: "users".to_string(),
            table_alias: None,
            projection: SqlProjection::Items(vec![sql_scalar_function_field_item(
                SqlScalarFunction::Trim,
                "name",
            )]),
            projection_aliases: vec![Some("trimmed_name".to_string())],
            predicate: None,
            distinct: false,
            group_by: vec![],
            having: vec![],
            order_by: vec![],
            limit: None,
            offset: None,
        }),
    );
}

#[test]
fn parse_sql_rejects_multi_statement_input() {
    let err = parse_sql("SELECT * FROM users; SELECT * FROM users")
        .expect_err("multi-statement SQL input should be rejected");

    assert_eq!(
        err,
        super::SqlParseError::UnsupportedFeature {
            feature: SqlFeatureCode::MultiStatementSql
        }
    );
}

#[test]
fn parse_sql_rejects_unknown_function_namespace() {
    let err = parse_sql("SELECT len(name) FROM users")
        .expect_err("unknown SQL function namespace should be rejected");

    assert_eq!(
        err,
        super::SqlParseError::UnsupportedFeature {
            feature: SqlFeatureCode::UnsupportedFunctionNamespace
        }
    );
}

#[test]
fn parse_sql_accepts_coalesce_and_nullif_in_where() {
    let statement = parse_sql("SELECT name FROM users WHERE COALESCE(NULLIF(age, 20), 99) = 99")
        .expect("COALESCE/NULLIF in WHERE should parse");

    assert_eq!(
        statement,
        SqlStatement::Select(SqlSelectStatement {
            entity: "users".to_string(),
            table_alias: None,
            projection: SqlProjection::Items(vec![SqlSelectItem::Field("name".to_string())]),
            projection_aliases: vec![None],
            predicate: Some(SqlExpr::Binary {
                op: SqlExprBinaryOp::Eq,
                left: Box::new(sql_scalar_function_expr(
                    SqlScalarFunction::Coalesce,
                    vec![
                        sql_scalar_function_expr(
                            SqlScalarFunction::NullIf,
                            vec![
                                SqlExpr::Field("age".to_string()),
                                SqlExpr::Literal(Value::Int64(20)),
                            ],
                        ),
                        SqlExpr::Literal(Value::Int64(99)),
                    ],
                )),
                right: Box::new(SqlExpr::Literal(Value::Int64(99))),
            }),
            distinct: false,
            group_by: vec![],
            having: vec![],
            order_by: vec![],
            limit: None,
            offset: None,
        }),
    );
}

#[test]
fn parse_sql_accepts_distinct_aggregate_qualifier() {
    let statement = parse_sql("SELECT COUNT(DISTINCT age) FROM users")
        .expect("aggregate DISTINCT qualifier should parse");

    assert_eq!(
        statement,
        SqlStatement::Select(SqlSelectStatement {
            entity: "users".to_string(),
            table_alias: None,
            projection: SqlProjection::Items(vec![SqlSelectItem::Aggregate(SqlAggregateCall {
                kind: SqlAggregateKind::Count,
                input: Some(Box::new(SqlExpr::Field("age".to_string()))),
                filter_expr: None,
                distinct: true,
            })]),
            projection_aliases: vec![None],
            predicate: None,
            distinct: false,
            group_by: vec![],
            having: vec![],
            order_by: vec![],
            limit: None,
            offset: None,
        }),
    );
}

#[test]
fn parse_sql_accepts_aggregate_filter_clauses() {
    let statement = parse_sql("SELECT COUNT(*) FILTER (WHERE age > 1) FROM users")
        .expect("aggregate FILTER clause should parse");

    assert_eq!(
        statement,
        SqlStatement::Select(SqlSelectStatement {
            entity: "users".to_string(),
            table_alias: None,
            projection: SqlProjection::Items(vec![SqlSelectItem::Aggregate(SqlAggregateCall {
                kind: SqlAggregateKind::Count,
                input: None,
                filter_expr: Some(Box::new(SqlExpr::Binary {
                    op: SqlExprBinaryOp::Gt,
                    left: Box::new(SqlExpr::Field("age".to_string())),
                    right: Box::new(SqlExpr::Literal(Value::Int64(1))),
                })),
                distinct: false,
            })]),
            projection_aliases: vec![None],
            predicate: None,
            distinct: false,
            group_by: vec![],
            having: vec![],
            order_by: vec![],
            limit: None,
            offset: None,
        }),
    );
}

#[test]
fn parse_sql_rejects_aggregate_filter_window_pairing() {
    let err =
        parse_sql("SELECT COUNT(*) FILTER (WHERE age > 1) OVER (ORDER BY age DESC) FROM users")
            .expect_err("aggregate FILTER + OVER should stay fail-closed");

    assert_eq!(
        err,
        super::SqlParseError::UnsupportedFeature {
            feature: SqlFeatureCode::WindowFunction
        }
    );
}

#[test]
fn parse_sql_accepts_expression_aggregate_inputs() {
    let statement = parse_sql("SELECT AVG(age + 1), COUNT(1), ROUND(AVG(age + 1), 2) FROM users")
        .expect("expression aggregate inputs should parse");

    assert_eq!(
        statement,
        SqlStatement::Select(SqlSelectStatement {
            entity: "users".to_string(),
            table_alias: None,
            projection: SqlProjection::Items(vec![
                SqlSelectItem::Aggregate(SqlAggregateCall {
                    kind: SqlAggregateKind::Avg,
                    input: Some(Box::new(SqlExpr::Binary {
                        op: SqlExprBinaryOp::Add,
                        left: Box::new(SqlExpr::Field("age".to_string())),
                        right: Box::new(SqlExpr::Literal(Value::Int64(1))),
                    })),
                    filter_expr: None,
                    distinct: false,
                }),
                SqlSelectItem::Aggregate(SqlAggregateCall {
                    kind: SqlAggregateKind::Count,
                    input: Some(Box::new(SqlExpr::Literal(Value::Int64(1)))),
                    filter_expr: None,
                    distinct: false,
                }),
                sql_round_item(
                    SqlExpr::Aggregate(SqlAggregateCall {
                        kind: SqlAggregateKind::Avg,
                        input: Some(Box::new(SqlExpr::Binary {
                            op: SqlExprBinaryOp::Add,
                            left: Box::new(SqlExpr::Field("age".to_string())),
                            right: Box::new(SqlExpr::Literal(Value::Int64(1))),
                        })),
                        filter_expr: None,
                        distinct: false,
                    },),
                    Value::Int64(2),
                ),
            ]),
            projection_aliases: vec![None, None, None],
            predicate: None,
            distinct: false,
            group_by: vec![],
            having: vec![],
            order_by: vec![],
            limit: None,
            offset: None,
        }),
    );
}

#[test]
fn parse_select_grouped_statement_with_unary_text_having_exprs() {
    let statement = parse_sql(
        "SELECT name, COUNT(*) \
         FROM users \
         GROUP BY name \
         HAVING LOWER(COALESCE(name, 'fallback')) >= 'b' \
         ORDER BY name ASC LIMIT 10",
    )
    .expect("grouped unary text-function HAVING statement should parse");

    assert_eq!(
        statement,
        SqlStatement::Select(SqlSelectStatement {
            entity: "users".to_string(),
            table_alias: None,
            projection: SqlProjection::Items(vec![
                SqlSelectItem::Field("name".to_string()),
                SqlSelectItem::Aggregate(SqlAggregateCall {
                    kind: SqlAggregateKind::Count,
                    input: None,
                    filter_expr: None,
                    distinct: false,
                }),
            ]),
            projection_aliases: vec![None, None],
            predicate: None,
            distinct: false,
            group_by: vec!["name".to_string()],
            having: vec![sql_binary_expr(
                sql_scalar_function_expr(
                    SqlScalarFunction::Lower,
                    vec![sql_scalar_function_expr(
                        SqlScalarFunction::Coalesce,
                        vec![
                            SqlExpr::Field("name".to_string()),
                            SqlExpr::Literal(Value::Text("fallback".to_string())),
                        ],
                    )],
                ),
                SqlExprBinaryOp::Gte,
                SqlExpr::Literal(Value::Text("b".to_string())),
            )],
            order_by: vec![SqlOrderTerm {
                field: sql_order_expr("name"),
                direction: SqlOrderDirection::Asc,
            }],
            limit: Some(10),
            offset: None,
        }),
    );
}

#[test]
fn parse_sql_accepts_function_wrapped_expression_aggregate_inputs() {
    let statement =
        parse_sql("SELECT SUM(ABS(age - 15)), AVG(COALESCE(NULLIF(age, 20), 0)) FROM users")
            .expect("function-wrapped expression aggregate inputs should parse");

    let SqlStatement::Select(statement) = statement else {
        panic!("expected aggregate SELECT statement");
    };

    assert_eq!(statement.projection_aliases, vec![None, None]);
    let SqlProjection::Items(items) = statement.projection else {
        panic!("expected item projection");
    };
    assert_eq!(items.len(), 2);
}

#[test]
fn parse_sql_accepts_unary_text_function_wrapped_expression_aggregate_inputs() {
    let statement = parse_sql(
        "SELECT MIN(LOWER(COALESCE(name, 'fallback'))), \
                MAX(LENGTH(TRIM(name))) \
         FROM users",
    )
    .expect("unary text-function wrapped aggregate inputs should parse");

    assert_eq!(
        statement,
        SqlStatement::Select(SqlSelectStatement {
            entity: "users".to_string(),
            table_alias: None,
            projection: SqlProjection::Items(vec![
                SqlSelectItem::Aggregate(SqlAggregateCall {
                    kind: SqlAggregateKind::Min,
                    input: Some(Box::new(sql_scalar_function_expr(
                        SqlScalarFunction::Lower,
                        vec![sql_scalar_function_expr(
                            SqlScalarFunction::Coalesce,
                            vec![
                                SqlExpr::Field("name".to_string()),
                                SqlExpr::Literal(Value::Text("fallback".to_string())),
                            ],
                        )],
                    ))),
                    filter_expr: None,
                    distinct: false,
                }),
                SqlSelectItem::Aggregate(SqlAggregateCall {
                    kind: SqlAggregateKind::Max,
                    input: Some(Box::new(sql_scalar_function_expr(
                        SqlScalarFunction::Length,
                        vec![sql_scalar_function_expr(
                            SqlScalarFunction::Trim,
                            vec![SqlExpr::Field("name".to_string())],
                        )],
                    ))),
                    filter_expr: None,
                    distinct: false,
                }),
            ]),
            projection_aliases: vec![None, None],
            predicate: None,
            distinct: false,
            group_by: vec![],
            having: vec![],
            order_by: vec![],
            limit: None,
            offset: None,
        }),
    );
}

#[test]
fn parse_sql_accepts_table_alias_identifier_form() {
    let statement = parse_sql("SELECT * FROM users u").expect("single-table alias should parse");

    assert_eq!(
        statement,
        SqlStatement::Select(SqlSelectStatement {
            entity: "users".to_string(),
            table_alias: Some("u".to_string()),
            projection: SqlProjection::All,
            projection_aliases: Vec::default(),
            predicate: None,
            distinct: false,
            group_by: vec![],
            having: vec![],
            order_by: vec![],
            limit: None,
            offset: None,
        }),
    );
}

#[test]
fn parse_sql_accepts_table_alias_as_form() {
    let statement = parse_sql(
        "SELECT u.name FROM users AS u WHERE u.age >= 21 ORDER BY LOWER(u.name) ASC LIMIT 1",
    )
    .expect("single-table AS alias should parse");

    assert_eq!(
        statement,
        SqlStatement::Select(SqlSelectStatement {
            entity: "users".to_string(),
            table_alias: Some("u".to_string()),
            projection: SqlProjection::Items(vec![SqlSelectItem::Field("u.name".to_string())]),
            projection_aliases: vec![None],
            predicate: option_sql_pred!(Predicate::Compare(ComparePredicate::with_coercion(
                "u.age",
                CompareOp::Gte,
                Value::Int64(21),
                CoercionId::NumericWiden,
            ))),
            distinct: false,
            group_by: vec![],
            having: vec![],
            order_by: vec![SqlOrderTerm {
                field: sql_order_expr("LOWER(u.name)"),
                direction: SqlOrderDirection::Asc,
            }],
            limit: Some(1),
            offset: None,
        }),
    );
}

#[test]
fn parse_sql_accepts_table_alias_for_schema_qualified_entity() {
    let statement = parse_sql(
        "SELECT u.name, u.age FROM public.users AS u WHERE u.age >= 21 ORDER BY u.age DESC",
    )
    .expect("single-table alias should parse for schema-qualified entity names");

    assert_eq!(
        statement,
        SqlStatement::Select(SqlSelectStatement {
            entity: "public.users".to_string(),
            table_alias: Some("u".to_string()),
            projection: SqlProjection::Items(vec![
                SqlSelectItem::Field("u.name".to_string()),
                SqlSelectItem::Field("u.age".to_string()),
            ]),
            projection_aliases: vec![None, None],
            predicate: option_sql_pred!(Predicate::Compare(ComparePredicate::with_coercion(
                "u.age",
                CompareOp::Gte,
                Value::Int64(21),
                CoercionId::NumericWiden,
            ))),
            distinct: false,
            group_by: vec![],
            having: vec![],
            order_by: vec![SqlOrderTerm {
                field: sql_order_expr("u.age"),
                direction: SqlOrderDirection::Desc,
            }],
            limit: None,
            offset: None,
        }),
    );
}

#[test]
fn parse_sql_rejects_quoted_identifier_syntax() {
    let err = parse_sql("SELECT \"name\" FROM users")
        .expect_err("quoted identifiers should be rejected in reduced parser");

    assert_eq!(
        err,
        super::SqlParseError::UnsupportedFeature {
            feature: SqlFeatureCode::QuotedIdentifiers
        }
    );
}

#[test]
fn parse_sql_normalization_is_case_and_whitespace_insensitive() {
    let canonical = parse_sql("SELECT name FROM users WHERE active = true ORDER BY name LIMIT 5")
        .expect("canonical statement should parse");
    let variant =
        parse_sql("  select   name  from users where active = TRUE  order by name  limit 5 ; ")
            .expect("variant statement should parse");

    assert_eq!(canonical, variant);
}

#[test]
fn parse_sql_rejects_excessive_expression_depth() {
    let mut predicate = String::new();
    for _ in 0..140 {
        predicate.push_str("NOT ");
    }
    predicate.push_str("active");
    let sql = format!("SELECT name FROM users WHERE {predicate}");

    let err = parse_sql(sql.as_str()).expect_err("deep expressions should reject before recursion");

    assert_eq!(
        err,
        SqlParseError::InvalidSyntax {
            kind: SqlSyntaxErrorKind::ExpressionDepthLimit {
                max_depth: MAX_SQL_EXPR_DEPTH
            },
        }
    );
}

#[test]
fn parse_sql_rejects_excessive_binary_chain_depth() {
    let mut predicate = String::from("active");
    for _ in 0..140 {
        predicate.push_str(" OR active");
    }
    let sql = format!("SELECT name FROM users WHERE {predicate}");

    let err = parse_sql(sql.as_str()).expect_err("deep binary chains should reject during parse");

    assert_eq!(
        err,
        SqlParseError::InvalidSyntax {
            kind: SqlSyntaxErrorKind::ExpressionDepthLimit {
                max_depth: MAX_SQL_EXPR_DEPTH
            },
        }
    );
}

#[test]
fn parse_sql_rejects_excessive_input_bytes() {
    let sql = " ".repeat(MAX_SQL_INPUT_BYTES.saturating_add(1));

    let err = parse_sql(sql.as_str()).expect_err("oversized SQL text should reject before lexing");

    assert_eq!(
        err,
        SqlParseError::InvalidSyntax {
            kind: SqlSyntaxErrorKind::InputTooLong {
                max_bytes: MAX_SQL_INPUT_BYTES
            },
        }
    );
}

#[test]
fn parse_sql_rejects_excessive_token_count() {
    let mut sql = String::from("SELECT ");
    for _ in 0..=MAX_SQL_TOKENS {
        sql.push_str("name,");
    }

    let err = parse_sql(sql.as_str()).expect_err("oversized token streams should reject in lexer");

    assert_eq!(
        err,
        SqlParseError::InvalidSyntax {
            kind: SqlSyntaxErrorKind::TokenLimit {
                max_tokens: MAX_SQL_TOKENS
            },
        }
    );
}

#[test]
fn parse_integrity_sql_accepts_only_the_four_canonical_forms() {
    assert_eq!(
        parse_integrity_sql("CHECK INTEGRITY Token QUICK")
            .expect("Quick integrity SQL should parse"),
        SqlIntegrityStatement::Quick {
            entity: "Token".to_string(),
        },
    );
    assert_eq!(
        parse_integrity_sql("check integrity Token deep start 'owner-key'")
            .expect("Deep start integrity SQL should parse"),
        SqlIntegrityStatement::DeepStart {
            entity: "Token".to_string(),
            submission_key: "owner-key".to_string(),
        },
    );
    assert_eq!(
        parse_integrity_sql("CHECK INTEGRITY DEEP CONTINUE '000102' AFTER 7;",)
            .expect("Deep continue integrity SQL should parse"),
        SqlIntegrityStatement::DeepContinue {
            job_id: "000102".to_string(),
            acknowledged_sequence: 7,
        },
    );
    assert_eq!(
        parse_integrity_sql("CHECK INTEGRITY DEEP ABORT '000102'")
            .expect("Deep abort integrity SQL should parse"),
        SqlIntegrityStatement::DeepAbort {
            job_id: "000102".to_string(),
        },
    );
}

#[test]
fn parse_integrity_sql_rejects_noncanonical_mode_and_argument_shapes() {
    for sql in [
        "CHECK INTEGRITY Token",
        "CHECK INTEGRITY Token DEEP",
        "CHECK INTEGRITY Token DEEP START",
        "CHECK INTEGRITY Token QUICK LIMIT 1",
        "CHECK INTEGRITY DEEP CONTINUE '000102'",
        "CHECK INTEGRITY DEEP CONTINUE Token '000102' AFTER 7",
        "CHECK INTEGRITY DEEP CONTINUE '000102' AFTER -1",
        "CHECK INTEGRITY DEEP ABORT Token '000102'",
        "CHECK INTEGRITY DEEP VERIFY '000102'",
        "SELECT * FROM Token",
    ] {
        assert!(
            parse_integrity_sql(sql).is_err(),
            "unsupported integrity SQL should reject: {sql}",
        );
    }
}