icydb-core 0.252.3

use crate::{
    db::{
        predicate::{CompareOp, Predicate},
        query::plan::expr::{
            BinaryOp, Expr, FieldId, FieldPath, Function, UnaryOp,
            compile_normalized_bool_expr_to_predicate,
            derive_normalized_bool_expr_predicate_subset,
        },
        sql::{
            lowering::predicate::{
                derive_sql_where_expr_predicate_subset, lower_sql_scalar_where_bool_expr,
                lower_sql_where_bool_expr, lower_sql_where_expr,
            },
            parser::parse_sql,
        },
    },
    value::Value,
};

fn parse_where_expr(sql: &str) -> crate::db::sql::parser::SqlExpr {
    let statement = parse_sql(sql).expect("SQL WHERE test statement should parse");
    let crate::db::sql::parser::SqlStatement::Select(select) = statement else {
        panic!("expected SELECT statement");
    };

    select
        .predicate
        .expect("SQL WHERE test statement should carry one predicate")
}

#[test]
fn lower_sql_where_bool_expr_preserves_upper_prefix_semantics() {
    let expr = parse_where_expr(
        "SELECT * FROM users WHERE UPPER(name) LIKE 'AL%' ORDER BY id ASC LIMIT 1",
    );

    let lowered =
        lower_sql_where_bool_expr(&expr).expect("UPPER(...) prefix LIKE should be admitted");
    let Expr::FunctionCall {
        function: Function::StartsWith,
        args,
    } = lowered
    else {
        panic!("UPPER(...) prefix LIKE should normalize onto STARTS_WITH(...)");
    };
    let [left, right] = args.as_slice() else {
        panic!("normalized STARTS_WITH(...) should keep two arguments");
    };
    let Expr::FunctionCall {
        function: Function::Upper,
        args,
    } = left
    else {
        panic!("UPPER target must remain an UPPER expression");
    };
    let [Expr::Field(field)] = args.as_slice() else {
        panic!("normalized UPPER(...) should keep the original field");
    };

    assert_eq!(field, &FieldId::new("name"));
    assert_eq!(right, &Expr::Literal(Value::Text("AL".to_string())));
}

#[test]
fn lower_sql_where_bool_expr_preserves_upper_ordering_semantics() {
    let expr =
        parse_where_expr("SELECT * FROM users WHERE UPPER(name) < '[' ORDER BY id ASC LIMIT 1");

    let lowered =
        lower_sql_where_bool_expr(&expr).expect("UPPER(...) ordered comparison should be admitted");
    let Expr::Binary {
        op: BinaryOp::Lt,
        left,
        right,
    } = lowered
    else {
        panic!("UPPER(...) ordered comparison should remain structural");
    };
    let Expr::FunctionCall {
        function: Function::Upper,
        args,
    } = left.as_ref()
    else {
        panic!("ordered UPPER target must remain an UPPER expression");
    };

    assert_eq!(args.as_slice(), &[Expr::Field(FieldId::new("name"))]);
    assert_eq!(right.as_ref(), &Expr::Literal(Value::Text("[".to_string())),);
}

#[test]
fn derive_where_predicate_subset_returns_none_for_admitted_expression_only_shapes() {
    let expr = parse_where_expr(
        "SELECT * FROM users WHERE STARTS_WITH(REPLACE(name, 'a', 'A'), TRIM('Al'))",
    );
    let lowered = lower_sql_where_bool_expr(&expr)
        .expect("admitted expression-only WHERE shape should lower successfully");

    assert!(
        derive_normalized_bool_expr_predicate_subset(&lowered).is_none(),
        "predicate extraction should stay subset-only for admitted expression-owned WHERE shapes",
    );
}

#[test]
fn lower_sql_where_expr_rejects_expression_only_shapes_on_strict_predicate_path() {
    let expr = parse_where_expr(
        "SELECT * FROM users WHERE STARTS_WITH(REPLACE(name, 'a', 'A'), TRIM('Al'))",
    );

    let err = lower_sql_where_expr(&expr)
        .expect_err("strict predicate-only WHERE lowering should reject expression-only shapes");

    std::assert_matches!(
        err,
        crate::db::sql::lowering::SqlLoweringError::UnsupportedWhereExpression
    );
}

#[test]
fn lower_sql_where_expr_recovers_membership_after_bool_lowering() {
    let expr = parse_where_expr("SELECT * FROM users WHERE age IN (10, 20, 10)");

    let predicate =
        lower_sql_where_expr(&expr).expect("strict membership WHERE predicate should lower");
    let Predicate::Compare(compare) = predicate else {
        panic!("membership should extract one compact compare predicate after bool lowering");
    };

    assert_eq!(compare.field(), "age");
    assert_eq!(compare.op(), CompareOp::In);
    let Value::List(values) = compare.value() else {
        panic!("compact membership predicate should carry a list literal");
    };
    assert_eq!(
        values.len(),
        2,
        "compact membership predicate should canonicalize duplicate members",
    );
}

#[test]
fn derive_where_predicate_subset_recovers_plain_compare_after_bool_lowering() {
    let expr = parse_where_expr("SELECT * FROM users WHERE age >= 21");
    let lowered = lower_sql_where_bool_expr(&expr).expect("plain compare WHERE shape should lower");
    let predicate = derive_sql_where_expr_predicate_subset(&lowered)
        .expect("plain compare WHERE shape should recover one predicate subset");

    assert!(
        matches!(
            predicate,
            Predicate::Compare(ref compare)
                if compare.field() == "age"
                    && compare.op() == CompareOp::Gte
                    && compare.value() == &Value::Int64(21)
        ),
        "plain compares should derive from the lowered boolean expression",
    );
}

#[test]
fn derive_where_predicate_subset_keeps_canonical_nested_path_identity() {
    let expr = Expr::Binary {
        op: BinaryOp::Gte,
        left: Box::new(Expr::FieldPath(FieldPath::new(
            "profile",
            vec!["rank".to_string()],
        ))),
        right: Box::new(Expr::Literal(Value::Int64(0))),
    };
    let predicate = derive_normalized_bool_expr_predicate_subset(&expr)
        .expect("nested scalar-path compare should retain a planner predicate");

    assert!(matches!(
        predicate,
        Predicate::Compare(ref compare)
            if compare.field() == "profile.rank"
                && compare.op() == CompareOp::Gte
                && compare.value() == &Value::Int64(0)
    ));
}

#[test]
fn derive_where_predicate_subset_recovers_compare_and_membership_after_bool_lowering() {
    let expr = parse_where_expr(
        "SELECT * FROM users \
         WHERE collection_id = '01KV5N439P0000000000000000' \
           AND stage IN ('Draft', 'Review', 'Draft')",
    );
    let lowered = lower_sql_where_bool_expr(&expr)
        .expect("simple compare plus membership WHERE shape should lower");

    let predicate = derive_sql_where_expr_predicate_subset(&lowered)
        .expect("simple compare plus membership WHERE should recover one predicate subset");
    let Predicate::And(children) = predicate else {
        panic!("lowered conjunction should recover one AND predicate");
    };

    assert!(
        children.iter().any(|child| matches!(
            child,
            Predicate::Compare(compare)
                if compare.field() == "collection_id"
                    && compare.op() == CompareOp::Eq
                    && compare.value()
                        == &Value::Text("01KV5N439P0000000000000000".to_string())
        )),
        "lowered conjunction should retain the fixed equality prefix",
    );
    assert!(
        children.iter().any(|child| matches!(
            child,
            Predicate::Compare(compare)
                if compare.field() == "stage"
                    && compare.op() == CompareOp::In
                    && matches!(
                        compare.value(),
                        Value::List(values)
                            if values.as_slice()
                                == [
                                    Value::Text("Draft".to_string()),
                                    Value::Text("Review".to_string())
                                ]
                    )
        )),
        "lowered conjunction should keep membership compact and deduplicated",
    );
}

#[test]
fn derive_where_predicate_subset_recovers_wide_membership_after_scalar_bool_lowering() {
    let expr = parse_where_expr(
        "SELECT * FROM users \
         WHERE collection_id = '01KV5N439P0000000000000000' \
           AND stage IN (\
             'Draft', 'Review', 'Published', 'Archived', 'Queued', \
             'Rejected', 'Minted', 'Burned', 'Frozen'\
           )",
    );
    let lowered = lower_sql_scalar_where_bool_expr(&expr)
        .expect("scalar compare plus wide membership WHERE shape should lower");
    let predicate = derive_sql_where_expr_predicate_subset(&lowered)
        .expect("scalar compare plus wide membership WHERE should recover one predicate subset");
    let Predicate::And(children) = predicate else {
        panic!("scalar lowered conjunction should recover one AND predicate: {predicate:?}");
    };

    assert!(
        children.iter().any(|child| matches!(
            child,
            Predicate::Compare(compare)
                if compare.field() == "stage"
                    && compare.op() == CompareOp::In
                    && matches!(compare.value(), Value::List(values) if values.len() == 9)
        )),
        "scalar lowered conjunction should keep wide membership compact: {children:?}",
    );
}

#[test]
fn lower_sql_where_bool_expr_keeps_membership_compact() {
    let expr = parse_where_expr(
        "SELECT * FROM users \
         WHERE collection_id IN (\
             '01KV5N439P0000000000000000', \
             'missing-collection-000', \
             'missing-collection-001', \
             'missing-collection-002', \
             'missing-collection-003'\
         )",
    );
    let lowered =
        lower_sql_where_bool_expr(&expr).expect("wide SQL membership WHERE shape should lower");
    let Expr::FunctionCall { function, args } = lowered else {
        panic!("SQL membership should lower to compact IN_LIST function");
    };

    assert_eq!(function, Function::InList);
    let [Expr::Field(field), Expr::Literal(Value::List(values))] = args.as_slice() else {
        panic!("compact SQL membership should retain target plus list literal");
    };

    assert_eq!(field, &FieldId::new("collection_id"));
    assert_eq!(values.len(), 5);
}

#[test]
fn derive_where_predicate_subset_recovers_folded_constant_compare_shapes() {
    let expr = parse_where_expr(
        "SELECT * FROM users WHERE name = TRIM('alpha') AND NULLIF('alpha', 'alpha') IS NULL",
    );
    let lowered = lower_sql_where_bool_expr(&expr)
        .expect("foldable compare WHERE shape should lower successfully");
    let subset = derive_normalized_bool_expr_predicate_subset(&lowered)
        .expect("foldable compare WHERE shape should recover one predicate subset");

    assert!(
        matches!(
            subset,
            crate::db::predicate::Predicate::Compare(ref compare)
                if compare.field() == "name"
                    && compare.op() == crate::db::predicate::CompareOp::Eq
                    && compare.value() == &Value::Text("alpha".to_string())
        ),
        "predicate subset derivation should stay available after legality is decided earlier",
    );
}

#[test]
fn compile_where_bool_expr_requires_normalized_shape() {
    let expr = Expr::Binary {
        op: crate::db::query::plan::expr::BinaryOp::Eq,
        left: Box::new(Expr::Literal(Value::Int64(5))),
        right: Box::new(Expr::Field(FieldId::new("age"))),
    };

    assert!(
        std::panic::catch_unwind(|| {
            let _ = compile_normalized_bool_expr_to_predicate(&expr);
        })
        .is_err(),
        "non-normalized predicate shape should panic",
    );
}

#[test]
fn compile_where_bool_expr_keeps_bare_bool_fields_structural() {
    let expr = Expr::Field(FieldId::new("active"));

    let Predicate::Compare(compare) = compile_normalized_bool_expr_to_predicate(&expr) else {
        panic!("bare bool field should compile to compare predicate");
    };

    assert_eq!(compare.field(), "active");
    assert_eq!(compare.op(), CompareOp::Eq);
    assert_eq!(compare.value(), &Value::Bool(true));
}

#[test]
fn compile_where_bool_expr_keeps_bool_not_false_branch_structural() {
    let expr = Expr::Unary {
        op: UnaryOp::Not,
        expr: Box::new(Expr::Field(FieldId::new("active"))),
    };

    let Predicate::Compare(compare) = compile_normalized_bool_expr_to_predicate(&expr) else {
        panic!("NOT bool field should compile to compare predicate");
    };

    assert_eq!(compare.field(), "active");
    assert_eq!(compare.op(), CompareOp::Eq);
    assert_eq!(compare.value(), &Value::Bool(false));
}

#[test]
fn compile_where_bool_expr_keeps_lowered_casefold_compare_structural() {
    let expr = Expr::Binary {
        op: crate::db::query::plan::expr::BinaryOp::Eq,
        left: Box::new(Expr::FunctionCall {
            function: Function::Lower,
            args: vec![Expr::Field(FieldId::new("name"))],
        }),
        right: Box::new(Expr::Literal(Value::Text("alice".into()))),
    };

    let Predicate::Compare(compare) = compile_normalized_bool_expr_to_predicate(&expr) else {
        panic!("LOWER(field) compare should compile to compare predicate");
    };

    assert_eq!(compare.field(), "name");
    assert_eq!(compare.op(), CompareOp::Eq);
    assert_eq!(compare.value(), &Value::Text("alice".into()));
    assert_eq!(
        compare.coercion().id(),
        crate::db::predicate::CoercionId::TextCasefold
    );
}

#[test]
fn compile_where_bool_expr_supports_missing_empty_and_collection_contains_functions() {
    let missing = Expr::FunctionCall {
        function: Function::IsMissing,
        args: vec![Expr::Field(FieldId::new("nickname"))],
    };
    let empty = Expr::FunctionCall {
        function: Function::IsEmpty,
        args: vec![Expr::Field(FieldId::new("tags"))],
    };
    let contains = Expr::FunctionCall {
        function: Function::CollectionContains,
        args: vec![
            Expr::Field(FieldId::new("tags")),
            Expr::Literal(Value::Text("mage".into())),
        ],
    };

    std::assert_matches!(
        compile_normalized_bool_expr_to_predicate(&missing),
        Predicate::IsMissing { field } if field == "nickname"
    );
    std::assert_matches!(
        compile_normalized_bool_expr_to_predicate(&empty),
        Predicate::IsEmpty { field } if field == "tags"
    );
    std::assert_matches!(
        compile_normalized_bool_expr_to_predicate(&contains),
        Predicate::Compare(_)
    );
}