icydb-core 0.144.13

use super::*;

// Execute one identifier-normalization EXPLAIN pair and assert both spellings
// collapse onto the same public output.
fn assert_explain_identifier_normalization_case(
    session: &DbSession<SessionSqlCanister>,
    lhs_sql: &str,
    rhs_sql: &str,
    context: &str,
) {
    let lhs = statement_explain_sql::<SessionSqlEntity>(session, lhs_sql)
        .unwrap_or_else(|err| panic!("{context} left-hand SQL should succeed: {err}"));
    let rhs = statement_explain_sql::<SessionSqlEntity>(session, rhs_sql)
        .unwrap_or_else(|err| panic!("{context} right-hand SQL should succeed: {err}"));

    assert_eq!(
        lhs, rhs,
        "{context} identifier spelling should normalize to the same EXPLAIN output",
    );
}

// Execute one EXPLAIN equivalence pair and require the full public explain
// surface to match exactly, including semantic and residual filter rendering.
fn assert_explain_exact_equivalence_case<E>(
    session: &DbSession<SessionSqlCanister>,
    left_sql: &str,
    right_sql: &str,
    context: &str,
) where
    E: PersistedRow<Canister = SessionSqlCanister> + crate::traits::EntityValue,
{
    let left = statement_explain_sql::<E>(session, left_sql)
        .unwrap_or_else(|err| panic!("{context} left SQL should succeed: {err}"));
    let right = statement_explain_sql::<E>(session, right_sql)
        .unwrap_or_else(|err| panic!("{context} right SQL should succeed: {err}"));

    assert_eq!(
        left, right,
        "{context} should keep the exact same public explain surface",
    );
}

fn assert_explain_load_shape_case<E>(
    session: &DbSession<SessionSqlCanister>,
    sql: &str,
    context: &str,
) where
    E: PersistedRow<Canister = SessionSqlCanister> + crate::traits::EntityValue,
{
    let explain = statement_explain_sql::<E>(session, sql)
        .unwrap_or_else(|err| panic!("{context} should succeed: {err}"));

    assert!(
        explain.contains("mode=Load"),
        "{context} should still render the base load plan",
    );
    assert!(
        explain.contains("access="),
        "{context} should still render one routed access shape",
    );
}

fn assert_explain_index_range_case(
    session: &DbSession<SessionSqlCanister>,
    sql: &str,
    tokens: &[&str],
    context: &str,
    require_json_object: bool,
) {
    let explain = statement_explain_sql::<IndexedSessionSqlEntity>(session, sql)
        .unwrap_or_else(|err| panic!("{context} should succeed: {err}"));

    if require_json_object {
        assert!(
            explain.starts_with('{') && explain.ends_with('}'),
            "{context} should be one JSON object payload",
        );
    }
    assert_explain_contains_tokens(explain.as_str(), tokens, context);
    assert!(
        !explain.contains(if require_json_object {
            "\"type\":\"FullScan\""
        } else {
            "access=FullScan"
        }),
        "{context} must not fall back to full scan: {explain}",
    );
}

// Execute one table of EXPLAIN SQL surfaces and assert each surface keeps the
// expected token contract, optionally requiring one JSON object payload.
fn assert_explain_token_matrix<E>(
    session: &DbSession<SessionSqlCanister>,
    cases: &[(&str, Vec<&str>)],
    context: &str,
    require_json_object: bool,
) where
    E: PersistedRow<Canister = SessionSqlCanister> + crate::traits::EntityValue,
{
    for (sql, tokens) in cases {
        let explain = statement_explain_sql::<E>(session, sql)
            .unwrap_or_else(|err| panic!("{context} should succeed: {err}"));

        if require_json_object {
            assert!(
                explain.starts_with('{') && explain.ends_with('}'),
                "{context} should be one JSON object payload: {sql}",
            );
        }

        assert_explain_contains_tokens(explain.as_str(), tokens.as_slice(), sql);
    }
}

#[test]
fn explain_sql_plan_matrix_queries_include_expected_tokens() {
    reset_session_sql_store();
    let session = sql_session();

    // Phase 1: define table-driven EXPLAIN plan SQL cases.
    let cases = vec![
        (
            "EXPLAIN SELECT * FROM SessionSqlEntity ORDER BY age LIMIT 1",
            vec!["mode=Load", "access="],
        ),
        (
            "EXPLAIN SELECT DISTINCT * FROM SessionSqlEntity ORDER BY id ASC",
            vec!["mode=Load", "distinct=true"],
        ),
        (
            "EXPLAIN SELECT age, COUNT(*) \
             FROM SessionSqlEntity \
             GROUP BY age \
             ORDER BY age ASC LIMIT 10",
            vec!["mode=Load", "grouping=Grouped"],
        ),
        (
            "EXPLAIN DELETE FROM SessionSqlEntity ORDER BY age LIMIT 1",
            vec!["mode=Delete", "access="],
        ),
        (
            "EXPLAIN SELECT COUNT(*) FROM SessionSqlEntity",
            vec!["mode=Load", "access="],
        ),
    ];

    // Phase 2: execute each EXPLAIN plan query and assert stable output tokens.
    assert_explain_token_matrix::<SessionSqlEntity>(
        &session,
        cases.as_slice(),
        "EXPLAIN plan matrix query",
        false,
    );
}

#[test]
fn explain_sql_execution_matrix_queries_include_expected_tokens() {
    reset_session_sql_store();
    let session = sql_session();

    // Phase 1: define table-driven EXPLAIN EXECUTION SQL cases.
    let cases = vec![
        (
            "EXPLAIN EXECUTION SELECT * FROM SessionSqlEntity ORDER BY age LIMIT 1",
            vec!["phases:", "execution:", "node_id=0", "node_properties:"],
        ),
        (
            "EXPLAIN EXECUTION SELECT age, COUNT(*) \
             FROM SessionSqlEntity \
             GROUP BY age \
             ORDER BY age ASC LIMIT 10",
            vec!["phases:", "execution:", "node_id=0", "execution_mode="],
        ),
        (
            "EXPLAIN EXECUTION SELECT COUNT(*) FROM SessionSqlEntity",
            vec![
                "phases:",
                "execution:",
                "AggregateCount execution_mode=",
                "node_id=0",
            ],
        ),
        (
            "EXPLAIN EXECUTION DELETE FROM SessionSqlEntity ORDER BY age LIMIT 1",
            vec!["phases:", "execution:", "node_id=0", "layer="],
        ),
    ];

    // Phase 2: execute each EXPLAIN EXECUTION query and assert stable output tokens.
    assert_explain_token_matrix::<SessionSqlEntity>(
        &session,
        cases.as_slice(),
        "EXPLAIN EXECUTION matrix query",
        false,
    );
}

#[test]
fn explain_sql_global_aggregate_reuses_runtime_shared_query_plan_cache() {
    reset_session_sql_store();
    let session = sql_session();
    seed_session_sql_entities(
        &session,
        &[
            ("aggregate-cache-a", 10),
            ("aggregate-cache-b", 20),
            ("aggregate-cache-c", 30),
        ],
    );
    let aggregate_sql = "SELECT COUNT(*) \
                         FROM SessionSqlEntity \
                         WHERE age >= 20 \
                         ORDER BY age DESC LIMIT 2";

    assert_eq!(
        session.query_plan_cache_len(),
        0,
        "global aggregate cache test should start from an empty shared query-plan cache",
    );
    let runtime = execute_sql_statement_for_tests::<SessionSqlEntity>(&session, aggregate_sql)
        .expect("runtime global aggregate should execute through the shared cache");
    assert!(
        matches!(runtime, SqlStatementResult::Projection { .. }),
        "runtime global aggregate should preserve projection result shape",
    );
    assert_eq!(
        session.query_plan_cache_len(),
        1,
        "runtime global aggregate should populate one shared prepared-plan entry",
    );

    for (sql, context) in [
        (
            "EXPLAIN SELECT COUNT(*) \
             FROM SessionSqlEntity \
             WHERE age >= 20 \
             ORDER BY age DESC LIMIT 2",
            "global aggregate EXPLAIN PLAN",
        ),
        (
            "EXPLAIN JSON SELECT COUNT(*) \
             FROM SessionSqlEntity \
             WHERE age >= 20 \
             ORDER BY age DESC LIMIT 2",
            "global aggregate EXPLAIN JSON",
        ),
        (
            "EXPLAIN EXECUTION SELECT COUNT(*) \
             FROM SessionSqlEntity \
             WHERE age >= 20 \
             ORDER BY age DESC LIMIT 2",
            "global aggregate EXPLAIN EXECUTION",
        ),
    ] {
        let explain = statement_explain_sql::<SessionSqlEntity>(&session, sql)
            .unwrap_or_else(|err| panic!("{context} should succeed: {err}"));
        assert!(
            explain.contains("access=")
                || explain.contains("\"access\"")
                || explain.contains("access_strategy="),
            "{context} should still render routed access facts: {explain}",
        );
        assert_eq!(
            session.query_plan_cache_len(),
            1,
            "{context} should reuse the runtime aggregate shared prepared-plan cache entry",
        );
    }
}

#[test]
fn explain_sql_global_aggregate_terminal_matrix_keeps_cached_descriptors() {
    reset_session_sql_store();
    let session = sql_session();
    seed_session_sql_entities(
        &session,
        &[
            ("aggregate-descriptor-a", 10),
            ("aggregate-descriptor-b", 20),
        ],
    );

    let explain = statement_explain_sql::<SessionSqlEntity>(
        &session,
        "EXPLAIN EXECUTION SELECT COUNT(*), SUM(age), MIN(age), MAX(age) \
         FROM SessionSqlEntity \
         WHERE age >= 10",
    )
    .expect("global aggregate terminal matrix EXPLAIN EXECUTION should succeed");

    assert_explain_contains_tokens(
        explain.as_str(),
        &[
            "AggregateCount execution_mode=",
            "AggregateSum execution_mode=",
            "AggregateMin execution_mode=",
            "AggregateMax execution_mode=",
        ],
        "global aggregate terminal matrix",
    );
    assert_eq!(
        session.query_plan_cache_len(),
        1,
        "all global aggregate terminals over the same base query should share one prepared-plan cache entry",
    );
}

#[test]
fn explain_sql_global_aggregate_indexed_input_uses_cached_index_route() {
    reset_session_sql_store();
    let session = indexed_sql_session();
    seed_indexed_session_sql_entities(&session, &[("Sam", 30), ("Sasha", 24), ("Mira", 40)]);

    for (sql, context) in [
        (
            "EXPLAIN SELECT COUNT(*) \
             FROM IndexedSessionSqlEntity \
             WHERE name = 'Sam'",
            "indexed global aggregate EXPLAIN PLAN",
        ),
        (
            "EXPLAIN EXECUTION SELECT COUNT(*) \
             FROM IndexedSessionSqlEntity \
             WHERE name = 'Sam'",
            "indexed global aggregate EXPLAIN EXECUTION",
        ),
    ] {
        let explain = statement_explain_sql::<IndexedSessionSqlEntity>(&session, sql)
            .unwrap_or_else(|err| panic!("{context} should succeed: {err}"));
        assert!(
            !explain.contains("access=FullScan")
                && !explain.contains("\"type\":\"FullScan\"")
                && !explain.contains("FullScan execution_mode="),
            "{context} should derive cached indexed route facts instead of full scan: {explain}",
        );
        assert_eq!(
            session.query_plan_cache_len(),
            1,
            "{context} should use the shared prepared-plan cache for the indexed base query",
        );
    }
}

#[test]
fn explain_sql_execution_surfaces_direct_slot_row_projection_materialization() {
    reset_session_sql_store();
    let session = sql_session();

    let explain = statement_explain_sql::<SessionSqlEntity>(
        &session,
        "EXPLAIN EXECUTION SELECT name FROM SessionSqlEntity ORDER BY id ASC LIMIT 1",
    )
    .expect("direct-slot-row EXPLAIN EXECUTION should succeed");

    assert!(
        explain.contains("proj_materialization=Text(\"direct_slot_row\")"),
        "scalar SQL EXPLAIN EXECUTION should expose direct slot-row projection materialization: {explain}",
    );
}

#[test]
fn explain_sql_execution_surfaces_covering_read_projection_materialization() {
    reset_session_sql_store();
    let session = indexed_sql_session();

    seed_indexed_session_sql_entities(&session, &[("Sam", 30), ("Sasha", 24), ("Mira", 40)]);
    let explain = statement_explain_sql::<IndexedSessionSqlEntity>(
        &session,
        "EXPLAIN EXECUTION SELECT name FROM IndexedSessionSqlEntity WHERE name = 'Sam' ORDER BY id ASC LIMIT 1",
    )
    .expect("covering-read EXPLAIN EXECUTION should succeed");

    assert!(
        explain.contains("proj_materialization=Text(\"covering_read\")"),
        "covering SQL EXPLAIN EXECUTION should expose covering-read projection materialization: {explain}",
    );
}

#[test]
fn explain_sql_execution_expression_owned_where_surfaces_explicit_residual_filter_expr() {
    reset_session_sql_store();
    let session = sql_session();

    let explain = statement_explain_sql::<SessionSqlEntity>(
        &session,
        "EXPLAIN EXECUTION SELECT name \
         FROM SessionSqlEntity \
         WHERE COALESCE(NULLIF(age, 20), 99) = 99 \
         ORDER BY age ASC",
    )
    .expect("expression-owned WHERE EXPLAIN EXECUTION should succeed");

    assert!(
        explain.contains("ResidualFilter execution_mode="),
        "expression-owned WHERE EXPLAIN EXECUTION should still surface one residual filter stage: {explain}",
    );
    assert!(
        explain.contains("filter_expr=COALESCE(NULLIF(age, 20), 99) = 99"),
        "expression-owned WHERE EXPLAIN EXECUTION should still surface the semantic filter expression: {explain}",
    );
    assert!(
        explain.contains("residual_filter_expr=COALESCE(NULLIF(age, 20), 99) = 99"),
        "expression-owned WHERE EXPLAIN EXECUTION should surface the explicit residual filter expression even without one derived predicate: {explain}",
    );
    assert!(
        !explain.contains("residual_filter_predicate="),
        "expression-owned WHERE EXPLAIN EXECUTION should not invent one residual predicate when the explicit residual is expression-only: {explain}",
    );
}

#[test]
fn explain_sql_execution_verbose_keyword_is_accepted_in_sql_surface() {
    reset_session_sql_store();
    let session = sql_session();

    let explain = statement_explain_sql::<SessionSqlEntity>(
        &session,
        "EXPLAIN EXECUTION VERBOSE SELECT name \
         FROM SessionSqlEntity \
         ORDER BY id ASC LIMIT 1",
    )
    .expect("EXPLAIN EXECUTION VERBOSE should succeed");

    assert!(
        explain.contains("execution:"),
        "EXPLAIN EXECUTION VERBOSE should keep the execution explain payload: {explain}",
    );
    assert!(
        explain.contains("Access choice:"),
        "EXPLAIN EXECUTION VERBOSE should surface the human-readable access-choice section: {explain}",
    );
    assert!(
        explain.contains("diag.s.semantic_reuse_artifact=shared_prepared_query_plan")
            && explain.contains("diag.s.semantic_reuse=miss"),
        "EXPLAIN EXECUTION VERBOSE should surface session-owned semantic reuse diagnostics: {explain}",
    );
}

#[test]
fn explain_sql_execution_verbose_reports_shared_query_plan_reuse_after_first_build() {
    reset_session_sql_store();
    let session = sql_session();
    let sql = "EXPLAIN EXECUTION VERBOSE SELECT name \
               FROM SessionSqlEntity \
               WHERE age >= 20 \
               ORDER BY age ASC, id ASC LIMIT 2";

    let first = statement_explain_sql::<SessionSqlEntity>(&session, sql)
        .expect("first SQL verbose execution explain should succeed");
    let second = statement_explain_sql::<SessionSqlEntity>(&session, sql)
        .expect("second SQL verbose execution explain should succeed");

    assert!(
        first.contains("diag.s.semantic_reuse_artifact=shared_prepared_query_plan")
            && first.contains("diag.s.semantic_reuse=miss"),
        "the first SQL verbose execution explain should miss the shared prepared query-plan cache: {first}",
    );
    assert!(
        second.contains("diag.s.semantic_reuse_artifact=shared_prepared_query_plan")
            && second.contains("diag.s.semantic_reuse=hit"),
        "the second SQL verbose execution explain should hit the shared prepared query-plan cache: {second}",
    );
}

#[test]
fn explain_sql_execution_verbose_keeps_distinct_semantic_identity_on_reuse_miss() {
    reset_session_sql_store();
    let session = sql_session();
    let left = statement_explain_sql::<SessionSqlEntity>(
        &session,
        "EXPLAIN EXECUTION VERBOSE SELECT name \
         FROM SessionSqlEntity \
         WHERE age >= 20 \
         ORDER BY age ASC, id ASC LIMIT 2",
    )
    .expect("left SQL verbose execution explain should succeed");
    let right = statement_explain_sql::<SessionSqlEntity>(
        &session,
        "EXPLAIN EXECUTION VERBOSE SELECT name \
         FROM SessionSqlEntity \
         WHERE age >= 20 \
         ORDER BY age DESC, id DESC LIMIT 1",
    )
    .expect("right SQL verbose execution explain should succeed");

    for (label, explain) in [
        ("left SQL verbose explain", left),
        ("right SQL verbose explain", right),
    ] {
        assert!(
            explain.contains("diag.s.semantic_reuse_artifact=shared_prepared_query_plan")
                && explain.contains("diag.s.semantic_reuse=miss"),
            "{label} should miss reuse when semantic ordering or limit identity differs: {explain}",
        );
    }
}

#[test]
fn explain_sql_execution_verbose_searched_case_where_uses_canonical_residual_predicate_surface() {
    reset_session_sql_store();
    let session = sql_session();
    seed_nullable_session_sql_entities(
        &session,
        &[
            ("alpha", Some("ally")),
            ("mira", None),
            ("nora", Some("north")),
            ("zed", None),
        ],
    );

    let left = statement_explain_sql::<SessionNullableSqlEntity>(
        &session,
        "EXPLAIN EXECUTION VERBOSE SELECT name, nickname \
         FROM SessionNullableSqlEntity \
         WHERE CASE WHEN nickname IS NULL THEN name >= 'm' ELSE name < 'm' END \
         ORDER BY name ASC LIMIT 5",
    )
    .expect("searched CASE verbose execution explain should succeed");
    let right = statement_explain_sql::<SessionNullableSqlEntity>(
        &session,
        "EXPLAIN EXECUTION VERBOSE SELECT name, nickname \
         FROM SessionNullableSqlEntity \
         WHERE (nickname IS NULL AND name >= 'm') \
            OR (NOT (nickname IS NULL) AND name < 'm') \
         ORDER BY name ASC LIMIT 5",
    )
    .expect("canonical boolean verbose execution explain should succeed");

    for (label, explain) in [
        ("searched CASE", left.as_str()),
        ("canonical boolean", right.as_str()),
    ] {
        assert_explain_contains_tokens(
            explain,
            &[
                "residual_filter_predicate=Or([",
                "diag.r.predicate_stage=residual_post_access",
            ],
            label,
        );
        assert!(
            !explain.contains("residual_filter_expr="),
            "{label} verbose execution explain should not keep a separate residual_filter_expr once the residual boolean tree canonicalizes back onto the shared predicate surface: {explain}",
        );
    }

    assert_explain_contains_tokens(
        left.as_str(),
        &[
            "diag.r.predicate_index_capability=requires_full_scan",
            "pred_idx_cap=Text(\"requires_full_scan\")",
        ],
        "searched CASE",
    );
    for (label, explain) in [
        ("searched CASE", left.as_str()),
        ("canonical boolean", right.as_str()),
    ] {
        assert!(
            explain.contains("diag.s.semantic_reuse_artifact=shared_prepared_query_plan")
                && (explain.contains("diag.s.semantic_reuse=miss")
                    || explain.contains("diag.s.semantic_reuse=hit")),
            "{label} verbose SQL explain should project the session-owned semantic reuse diagnostics separately from the canonical planner/executor payload: {explain}",
        );
    }
}

#[test]
fn explain_sql_execution_equivalent_and_shapes_preserve_exact_filter_surface() {
    reset_session_sql_store();
    let session = sql_session();

    assert_explain_exact_equivalence_case::<SessionSqlEntity>(
        &session,
        "EXPLAIN EXECUTION SELECT name \
         FROM SessionSqlEntity \
         WHERE age = 20 AND name = 'alpha' \
         ORDER BY id ASC LIMIT 5",
        "EXPLAIN EXECUTION SELECT name \
         FROM SessionSqlEntity \
         WHERE name = 'alpha' AND (age = 20) \
         ORDER BY id ASC LIMIT 5",
        "equivalent extractable AND WHERE EXPLAIN EXECUTION",
    );
}

#[test]
fn explain_sql_execution_equivalent_residual_and_shapes_preserve_exact_filter_surface() {
    reset_session_sql_store();
    let session = sql_session();

    assert_explain_exact_equivalence_case::<SessionSqlEntity>(
        &session,
        "EXPLAIN EXECUTION SELECT name \
         FROM SessionSqlEntity \
         WHERE STARTS_WITH(REPLACE(name, 'a', 'A'), TRIM('Al')) AND age = 20 \
         ORDER BY id ASC LIMIT 5",
        "EXPLAIN EXECUTION SELECT name \
         FROM SessionSqlEntity \
         WHERE age = 20 AND (STARTS_WITH(REPLACE(name, 'a', 'A'), TRIM('Al'))) \
         ORDER BY id ASC LIMIT 5",
        "equivalent residual AND WHERE EXPLAIN EXECUTION",
    );
}

#[test]
fn explain_sql_execution_equivalent_or_shapes_preserve_exact_filter_surface() {
    reset_session_sql_store();
    let session = sql_session();

    assert_explain_exact_equivalence_case::<SessionSqlEntity>(
        &session,
        "EXPLAIN EXECUTION SELECT name \
         FROM SessionSqlEntity \
         WHERE age = 20 OR name = 'alpha' \
         ORDER BY id ASC LIMIT 5",
        "EXPLAIN EXECUTION SELECT name \
         FROM SessionSqlEntity \
         WHERE name = 'alpha' OR (age = 20) \
         ORDER BY id ASC LIMIT 5",
        "equivalent extractable OR WHERE EXPLAIN EXECUTION",
    );
}

#[test]
fn explain_sql_execution_equivalent_residual_or_shapes_preserve_exact_filter_surface() {
    reset_session_sql_store();
    let session = sql_session();

    assert_explain_exact_equivalence_case::<SessionSqlEntity>(
        &session,
        "EXPLAIN EXECUTION SELECT name \
         FROM SessionSqlEntity \
         WHERE STARTS_WITH(REPLACE(name, 'a', 'A'), TRIM('Al')) OR age = 20 \
         ORDER BY id ASC LIMIT 5",
        "EXPLAIN EXECUTION SELECT name \
         FROM SessionSqlEntity \
         WHERE age = 20 OR (STARTS_WITH(REPLACE(name, 'a', 'A'), TRIM('Al'))) \
         ORDER BY id ASC LIMIT 5",
        "equivalent residual OR WHERE EXPLAIN EXECUTION",
    );
}

#[test]
fn explain_sql_execution_equivalent_mixed_extractable_shapes_preserve_exact_filter_surface() {
    reset_session_sql_store();
    let session = sql_session();

    assert_explain_exact_equivalence_case::<SessionSqlEntity>(
        &session,
        "EXPLAIN EXECUTION SELECT name \
         FROM SessionSqlEntity \
         WHERE (age = 20 AND name = 'alpha') OR age = 30 \
         ORDER BY id ASC LIMIT 5",
        "EXPLAIN EXECUTION SELECT name \
         FROM SessionSqlEntity \
         WHERE age = 30 OR (name = 'alpha' AND age = 20) \
         ORDER BY id ASC LIMIT 5",
        "equivalent mixed extractable WHERE EXPLAIN EXECUTION",
    );
}

#[test]
fn explain_sql_execution_equivalent_mixed_residual_shapes_preserve_exact_filter_surface() {
    reset_session_sql_store();
    let session = sql_session();

    assert_explain_exact_equivalence_case::<SessionSqlEntity>(
        &session,
        "EXPLAIN EXECUTION SELECT name \
         FROM SessionSqlEntity \
         WHERE (age = 20 AND STARTS_WITH(REPLACE(name, 'a', 'A'), TRIM('Al'))) OR name = 'alpha' \
         ORDER BY id ASC LIMIT 5",
        "EXPLAIN EXECUTION SELECT name \
         FROM SessionSqlEntity \
         WHERE name = 'alpha' OR (STARTS_WITH(REPLACE(name, 'a', 'A'), TRIM('Al')) AND age = 20) \
         ORDER BY id ASC LIMIT 5",
        "equivalent mixed residual WHERE EXPLAIN EXECUTION",
    );
}

#[test]
fn explain_sql_execution_duplicate_extractable_boolean_children_collapse() {
    reset_session_sql_store();
    let session = sql_session();

    assert_explain_exact_equivalence_case::<SessionSqlEntity>(
        &session,
        "EXPLAIN EXECUTION SELECT name \
         FROM SessionSqlEntity \
         WHERE age = 20 AND age = 20 \
         ORDER BY id ASC LIMIT 5",
        "EXPLAIN EXECUTION SELECT name \
         FROM SessionSqlEntity \
         WHERE age = 20 \
         ORDER BY id ASC LIMIT 5",
        "duplicate extractable boolean children EXPLAIN EXECUTION",
    );
}

#[test]
fn explain_sql_execution_duplicate_residual_boolean_children_collapse() {
    reset_session_sql_store();
    let session = sql_session();

    assert_explain_exact_equivalence_case::<SessionSqlEntity>(
        &session,
        "EXPLAIN EXECUTION SELECT name \
         FROM SessionSqlEntity \
         WHERE STARTS_WITH(REPLACE(name, 'a', 'A'), TRIM('Al')) \
           OR STARTS_WITH(REPLACE(name, 'a', 'A'), TRIM('Al')) \
         ORDER BY id ASC LIMIT 5",
        "EXPLAIN EXECUTION SELECT name \
         FROM SessionSqlEntity \
         WHERE STARTS_WITH(REPLACE(name, 'a', 'A'), TRIM('Al')) \
         ORDER BY id ASC LIMIT 5",
        "duplicate residual boolean children EXPLAIN EXECUTION",
    );
}

#[test]
fn explain_sql_execution_equivalent_extractable_compare_orientations_preserve_exact_filter_surface()
{
    reset_session_sql_store();
    let session = sql_session();

    assert_explain_exact_equivalence_case::<SessionSqlEntity>(
        &session,
        "EXPLAIN EXECUTION SELECT name \
         FROM SessionSqlEntity \
         WHERE 20 = age \
         ORDER BY id ASC LIMIT 5",
        "EXPLAIN EXECUTION SELECT name \
         FROM SessionSqlEntity \
         WHERE age = 20 \
         ORDER BY id ASC LIMIT 5",
        "equivalent extractable compare orientations EXPLAIN EXECUTION",
    );
}

#[test]
fn explain_sql_execution_equivalent_residual_compare_orientations_preserve_exact_filter_surface() {
    reset_session_sql_store();
    let session = sql_session();

    assert_explain_exact_equivalence_case::<SessionSqlEntity>(
        &session,
        "EXPLAIN EXECUTION SELECT name \
         FROM SessionSqlEntity \
         WHERE 'AlphA' = REPLACE(name, 'a', 'A') \
         ORDER BY id ASC LIMIT 5",
        "EXPLAIN EXECUTION SELECT name \
         FROM SessionSqlEntity \
         WHERE REPLACE(name, 'a', 'A') = 'AlphA' \
         ORDER BY id ASC LIMIT 5",
        "equivalent residual compare orientations EXPLAIN EXECUTION",
    );
}

#[test]
fn explain_sql_json_matrix_queries_include_expected_tokens() {
    reset_session_sql_store();
    let session = sql_session();

    // Phase 1: define table-driven EXPLAIN JSON SQL cases.
    let cases = vec![
        (
            "EXPLAIN JSON SELECT * FROM SessionSqlEntity ORDER BY age LIMIT 1",
            vec!["\"mode\":{\"type\":\"Load\"", "\"access\":"],
        ),
        (
            "EXPLAIN JSON SELECT DISTINCT * FROM SessionSqlEntity ORDER BY id ASC",
            vec!["\"mode\":{\"type\":\"Load\"", "\"distinct\":true"],
        ),
        (
            "EXPLAIN JSON SELECT age, COUNT(*) \
             FROM SessionSqlEntity \
             GROUP BY age \
             ORDER BY age ASC LIMIT 10",
            vec!["\"mode\":{\"type\":\"Load\"", "\"grouping\""],
        ),
        (
            "EXPLAIN JSON DELETE FROM SessionSqlEntity ORDER BY age LIMIT 1",
            vec!["\"mode\":{\"type\":\"Delete\"", "\"access\":"],
        ),
        (
            "EXPLAIN JSON SELECT COUNT(*) FROM SessionSqlEntity",
            vec!["\"mode\":{\"type\":\"Load\"", "\"access\":"],
        ),
    ];

    // Phase 2: execute each EXPLAIN JSON query and assert stable output tokens.
    assert_explain_token_matrix::<SessionSqlEntity>(
        &session,
        cases.as_slice(),
        "EXPLAIN JSON matrix query",
        true,
    );
}

#[test]
fn explain_sql_delete_rejection_matrix_preserves_unsupported_feature_detail() {
    reset_session_sql_store();
    let session = sql_session();

    for (sql, feature, context) in [
        (
            "EXPLAIN SELECT * FROM SessionSqlEntity JOIN other ON SessionSqlEntity.id = other.id",
            "JOIN",
            "EXPLAIN JOIN",
        ),
        (
            "EXPLAIN JSON SELECT * FROM SessionSqlEntity JOIN other ON SessionSqlEntity.id = other.id",
            "JOIN",
            "EXPLAIN JSON JOIN",
        ),
    ] {
        let err = statement_explain_sql::<SessionSqlEntity>(&session, sql)
            .expect_err("unsupported EXPLAIN feature should stay fail-closed");

        assert!(
            matches!(
                err,
                QueryError::Execute(crate::db::query::intent::QueryExecutionError::Unsupported(
                    _
                ))
            ),
            "{context} should fail through the unsupported SQL boundary",
        );
        assert_sql_unsupported_feature_detail(err, feature);
    }
}

#[test]
fn explain_sql_delete_direct_starts_with_family_matches_like_output() {
    reset_indexed_session_sql_store();
    let session = indexed_sql_session();

    // Phase 1: compare the accepted direct family against the established LIKE
    // family so EXPLAIN stays honest and surface-coherent for delete routes too.
    let cases = [
        (
            "EXPLAIN DELETE FROM IndexedSessionSqlEntity WHERE STARTS_WITH(name, 'S') ORDER BY name ASC LIMIT 2",
            "EXPLAIN DELETE FROM IndexedSessionSqlEntity WHERE name LIKE 'S%' ORDER BY name ASC LIMIT 2",
            "strict direct STARTS_WITH delete explain",
        ),
        (
            "EXPLAIN DELETE FROM IndexedSessionSqlEntity WHERE STARTS_WITH(LOWER(name), 's') ORDER BY name ASC LIMIT 2",
            "EXPLAIN DELETE FROM IndexedSessionSqlEntity WHERE LOWER(name) LIKE 's%' ORDER BY name ASC LIMIT 2",
            "direct LOWER(field) STARTS_WITH delete explain",
        ),
        (
            "EXPLAIN DELETE FROM IndexedSessionSqlEntity WHERE STARTS_WITH(UPPER(name), 'S') ORDER BY name ASC LIMIT 2",
            "EXPLAIN DELETE FROM IndexedSessionSqlEntity WHERE UPPER(name) LIKE 'S%' ORDER BY name ASC LIMIT 2",
            "direct UPPER(field) STARTS_WITH delete explain",
        ),
    ];

    // Phase 2: assert the logical plan text remains the same across both
    // spellings, proving the accepted direct family reuses the same delete path.
    for (direct_sql, like_sql, context) in cases {
        let direct = statement_explain_sql::<IndexedSessionSqlEntity>(&session, direct_sql)
            .expect("direct STARTS_WITH delete EXPLAIN should succeed");
        let like = statement_explain_sql::<IndexedSessionSqlEntity>(&session, like_sql)
            .expect("LIKE delete EXPLAIN should succeed");

        assert_eq!(
            direct, like,
            "bounded direct STARTS_WITH delete EXPLAIN should match the established LIKE path: {context}",
        );
    }
}

#[test]
fn explain_sql_delete_direct_text_range_matrix_preserves_index_range_route() {
    reset_indexed_session_sql_store();
    let session = indexed_sql_session();

    for (sql, tokens, context) in [
        (
            "EXPLAIN DELETE FROM IndexedSessionSqlEntity WHERE UPPER(name) >= 'S' AND UPPER(name) < 'T' ORDER BY name ASC LIMIT 2",
            &[
                "mode=Delete",
                "access=IndexRange",
                "predicate=And([Compare",
                "op: Lt, value: Text(\"T\")",
                "op: Gte, value: Text(\"S\")",
                "id: TextCasefold",
            ][..],
            "direct UPPER(field) ordered text-range delete EXPLAIN",
        ),
        (
            "EXPLAIN DELETE FROM IndexedSessionSqlEntity WHERE LOWER(name) >= 's' AND LOWER(name) < 't' ORDER BY name ASC LIMIT 2",
            &[
                "mode=Delete",
                "access=IndexRange",
                "predicate=And([Compare",
                "op: Lt, value: Text(\"t\")",
                "op: Gte, value: Text(\"s\")",
                "id: TextCasefold",
            ][..],
            "direct LOWER(field) ordered text-range delete EXPLAIN",
        ),
    ] {
        assert_explain_index_range_case(
            &session,
            sql,
            tokens,
            &format!("{context} should preserve the shared expression index-range route"),
            false,
        );
    }
}

#[test]
fn explain_json_sql_direct_text_range_matrix_preserves_index_range_route() {
    reset_indexed_session_sql_store();
    let session = indexed_sql_session();

    for (sql, tokens, context) in [
        (
            "EXPLAIN JSON SELECT name FROM IndexedSessionSqlEntity WHERE UPPER(name) >= 'S' AND UPPER(name) < 'T' ORDER BY name ASC",
            &[
                "\"mode\":{\"type\":\"Load\"",
                "\"access\":{\"type\":\"IndexRange\"",
                "\"predicate\":\"And([Compare",
                "id: TextCasefold",
            ][..],
            "direct UPPER(field) ordered text-range JSON EXPLAIN",
        ),
        (
            "EXPLAIN JSON SELECT name FROM IndexedSessionSqlEntity WHERE LOWER(name) >= 's' AND LOWER(name) < 't' ORDER BY name ASC",
            &[
                "\"mode\":{\"type\":\"Load\"",
                "\"access\":{\"type\":\"IndexRange\"",
                "\"predicate\":\"And([Compare",
                "id: TextCasefold",
            ][..],
            "direct LOWER(field) ordered text-range JSON EXPLAIN",
        ),
    ] {
        assert_explain_index_range_case(
            &session,
            sql,
            tokens,
            &format!("{context} should preserve the shared expression index-range route"),
            true,
        );
    }
}

#[test]
fn explain_json_sql_direct_equivalent_prefix_matrix_preserves_index_range_route() {
    reset_indexed_session_sql_store();
    let session = indexed_sql_session();

    for (sql, context) in [
        (
            "EXPLAIN JSON SELECT name FROM IndexedSessionSqlEntity WHERE UPPER(name) LIKE 'S%' ORDER BY name ASC",
            "direct UPPER(field) LIKE JSON explain route",
        ),
        (
            "EXPLAIN JSON SELECT name FROM IndexedSessionSqlEntity WHERE STARTS_WITH(UPPER(name), 'S') ORDER BY name ASC",
            "direct UPPER(field) STARTS_WITH JSON explain route",
        ),
        (
            "EXPLAIN JSON SELECT name FROM IndexedSessionSqlEntity WHERE UPPER(name) >= 'S' AND UPPER(name) < 'T' ORDER BY name ASC",
            "direct UPPER(field) ordered text-range JSON explain route",
        ),
        (
            "EXPLAIN JSON SELECT name FROM IndexedSessionSqlEntity WHERE LOWER(name) LIKE 's%' ORDER BY name ASC",
            "direct LOWER(field) LIKE JSON explain route",
        ),
        (
            "EXPLAIN JSON SELECT name FROM IndexedSessionSqlEntity WHERE STARTS_WITH(LOWER(name), 's') ORDER BY name ASC",
            "direct LOWER(field) STARTS_WITH JSON explain route",
        ),
        (
            "EXPLAIN JSON SELECT name FROM IndexedSessionSqlEntity WHERE LOWER(name) >= 's' AND LOWER(name) < 't' ORDER BY name ASC",
            "direct LOWER(field) ordered text-range JSON explain route",
        ),
    ] {
        assert_explain_index_range_case(
            &session,
            sql,
            &[
                "\"mode\":{\"type\":\"Load\"",
                "\"access\":{\"type\":\"IndexRange\"",
            ],
            context,
            true,
        );
    }
}

#[test]
fn explain_json_sql_delete_direct_text_range_matrix_preserves_index_range_route() {
    reset_indexed_session_sql_store();
    let session = indexed_sql_session();

    for (sql, tokens, context) in [
        (
            "EXPLAIN JSON DELETE FROM IndexedSessionSqlEntity WHERE UPPER(name) >= 'S' AND UPPER(name) < 'T' ORDER BY name ASC LIMIT 2",
            &[
                "\"mode\":{\"type\":\"Delete\"",
                "\"access\":{\"type\":\"IndexRange\"",
                "\"predicate\":\"And([Compare",
                "id: TextCasefold",
            ][..],
            "direct UPPER(field) ordered text-range JSON delete EXPLAIN",
        ),
        (
            "EXPLAIN JSON DELETE FROM IndexedSessionSqlEntity WHERE LOWER(name) >= 's' AND LOWER(name) < 't' ORDER BY name ASC LIMIT 2",
            &[
                "\"mode\":{\"type\":\"Delete\"",
                "\"access\":{\"type\":\"IndexRange\"",
                "\"predicate\":\"And([Compare",
                "id: TextCasefold",
            ][..],
            "direct LOWER(field) ordered text-range JSON delete EXPLAIN",
        ),
    ] {
        assert_explain_index_range_case(
            &session,
            sql,
            tokens,
            &format!("{context} should preserve the shared expression index-range route"),
            true,
        );
    }
}

#[test]
fn explain_json_sql_delete_direct_equivalent_prefix_matrix_preserves_index_range_route() {
    reset_indexed_session_sql_store();
    let session = indexed_sql_session();

    for (sql, context) in [
        (
            "EXPLAIN JSON DELETE FROM IndexedSessionSqlEntity WHERE UPPER(name) LIKE 'S%' ORDER BY name ASC LIMIT 2",
            "direct UPPER(field) LIKE JSON delete explain route",
        ),
        (
            "EXPLAIN JSON DELETE FROM IndexedSessionSqlEntity WHERE STARTS_WITH(UPPER(name), 'S') ORDER BY name ASC LIMIT 2",
            "direct UPPER(field) STARTS_WITH JSON delete explain route",
        ),
        (
            "EXPLAIN JSON DELETE FROM IndexedSessionSqlEntity WHERE UPPER(name) >= 'S' AND UPPER(name) < 'T' ORDER BY name ASC LIMIT 2",
            "direct UPPER(field) ordered text-range JSON delete explain route",
        ),
        (
            "EXPLAIN JSON DELETE FROM IndexedSessionSqlEntity WHERE LOWER(name) LIKE 's%' ORDER BY name ASC LIMIT 2",
            "direct LOWER(field) LIKE JSON delete explain route",
        ),
        (
            "EXPLAIN JSON DELETE FROM IndexedSessionSqlEntity WHERE STARTS_WITH(LOWER(name), 's') ORDER BY name ASC LIMIT 2",
            "direct LOWER(field) STARTS_WITH JSON delete explain route",
        ),
        (
            "EXPLAIN JSON DELETE FROM IndexedSessionSqlEntity WHERE LOWER(name) >= 's' AND LOWER(name) < 't' ORDER BY name ASC LIMIT 2",
            "direct LOWER(field) ordered text-range JSON delete explain route",
        ),
    ] {
        assert_explain_index_range_case(
            &session,
            sql,
            &[
                "\"mode\":{\"type\":\"Delete\"",
                "\"access\":{\"type\":\"IndexRange\"",
            ],
            context,
            true,
        );
    }
}

#[test]
fn explain_sql_identifier_normalization_matrix_matches_unqualified_output() {
    reset_session_sql_store();
    let session = sql_session();

    let cases = [
        (
            "logical explain qualified identifiers",
            "EXPLAIN SELECT * \
             FROM public.SessionSqlEntity \
             WHERE SessionSqlEntity.age >= 21 \
             ORDER BY SessionSqlEntity.age DESC LIMIT 1",
            "EXPLAIN SELECT * \
             FROM SessionSqlEntity \
             WHERE age >= 21 \
             ORDER BY age DESC LIMIT 1",
        ),
        (
            "execution explain qualified identifiers",
            "EXPLAIN EXECUTION SELECT SessionSqlEntity.name \
             FROM SessionSqlEntity \
             WHERE SessionSqlEntity.age >= 21 \
             ORDER BY SessionSqlEntity.age DESC LIMIT 1",
            "EXPLAIN EXECUTION SELECT name \
             FROM SessionSqlEntity \
             WHERE age >= 21 \
             ORDER BY age DESC LIMIT 1",
        ),
        (
            "execution explain table alias identifiers",
            "EXPLAIN EXECUTION SELECT alias.name \
             FROM SessionSqlEntity alias \
             WHERE alias.age >= 21 \
             ORDER BY alias.age DESC LIMIT 1",
            "EXPLAIN EXECUTION SELECT name \
             FROM SessionSqlEntity \
             WHERE age >= 21 \
             ORDER BY age DESC LIMIT 1",
        ),
    ];

    for (context, lhs_sql, rhs_sql) in cases {
        assert_explain_identifier_normalization_case(&session, lhs_sql, rhs_sql, context);
    }
}

#[test]
fn explain_sql_distinct_surface_matrix_returns_expected_tokens() {
    reset_session_sql_store();
    let session = sql_session();

    for (sql, tokens, context) in [
        (
            "EXPLAIN EXECUTION SELECT DISTINCT * FROM SessionSqlEntity ORDER BY id ASC LIMIT 1",
            &["node_id=0"][..],
            "execution explain distinct star",
        ),
        (
            "EXPLAIN SELECT DISTINCT age FROM SessionSqlEntity",
            &["distinct=true"][..],
            "logical explain distinct scalar projection",
        ),
    ] {
        let explain = statement_explain_sql::<SessionSqlEntity>(&session, sql)
            .unwrap_or_else(|err| panic!("{context} should succeed: {err}"));
        assert_explain_contains_tokens(explain.as_str(), tokens, context);
    }
}

#[test]
fn explain_sql_distinct_rejects_order_by_non_projected_field() {
    reset_session_sql_store();
    let session = sql_session();

    for (sql, context) in [
        (
            "EXPLAIN SELECT DISTINCT name FROM SessionSqlEntity ORDER BY age ASC",
            "logical EXPLAIN DISTINCT ORDER BY non-projected field",
        ),
        (
            "EXPLAIN JSON SELECT DISTINCT name FROM SessionSqlEntity ORDER BY age ASC",
            "JSON EXPLAIN DISTINCT ORDER BY non-projected field",
        ),
        (
            "EXPLAIN EXECUTION SELECT DISTINCT name FROM SessionSqlEntity ORDER BY age ASC",
            "execution EXPLAIN DISTINCT ORDER BY non-projected field",
        ),
    ] {
        let err = statement_explain_sql::<SessionSqlEntity>(&session, sql)
            .expect_err("EXPLAIN DISTINCT ORDER BY on a non-projected field should fail closed");

        assert!(
            err.to_string().contains(
                "SELECT DISTINCT ORDER BY terms must be derivable from the projected distinct tuple"
            ),
            "{context} should preserve the DISTINCT projected-tuple boundary message: {err}",
        );
    }
}

#[test]
fn explain_sql_alias_normalization_matrix_matches_canonical_plan_output() {
    reset_session_sql_store();
    let session = sql_session();
    for (aliased_sql, canonical_sql, context) in [
        (
            "EXPLAIN SELECT name AS display_name FROM SessionSqlEntity ORDER BY age LIMIT 1",
            "EXPLAIN SELECT name FROM SessionSqlEntity ORDER BY age LIMIT 1",
            "projection aliases",
        ),
        (
            "EXPLAIN SELECT name AS display_name FROM SessionSqlEntity ORDER BY display_name ASC LIMIT 1",
            "EXPLAIN SELECT name FROM SessionSqlEntity ORDER BY name ASC LIMIT 1",
            "ORDER BY field aliases",
        ),
    ] {
        assert_session_sql_alias_matches_canonical::<String>(
            &session,
            statement_explain_sql::<SessionSqlEntity>,
            aliased_sql,
            canonical_sql,
            context,
        );
    }
    reset_indexed_session_sql_store();
    let indexed_session = indexed_sql_session();
    assert_session_sql_alias_matches_canonical::<String>(
        &indexed_session,
        statement_explain_sql::<ExpressionIndexedSessionSqlEntity>,
        "EXPLAIN SELECT LOWER(name) AS normalized_name FROM ExpressionIndexedSessionSqlEntity ORDER BY normalized_name ASC LIMIT 1",
        "EXPLAIN SELECT LOWER(name) FROM ExpressionIndexedSessionSqlEntity ORDER BY LOWER(name) ASC LIMIT 1",
        "ORDER BY LOWER(field) aliases",
    );
}

#[test]
fn explain_sql_where_searched_case_matches_canonical_boolean_output() {
    reset_session_sql_store();
    let session = sql_session();

    let left = statement_explain_sql::<SessionSqlEntity>(
        &session,
        "EXPLAIN SELECT name \
         FROM SessionSqlEntity \
         WHERE CASE WHEN age >= 30 THEN TRUE ELSE age = 20 END \
         ORDER BY age ASC",
    )
    .expect("searched CASE explain parity left SQL should succeed");
    let right = statement_explain_sql::<SessionSqlEntity>(
        &session,
        "EXPLAIN SELECT name \
         FROM SessionSqlEntity \
         WHERE COALESCE(age >= 30, FALSE) \
            OR (NOT COALESCE(age >= 30, FALSE) AND age = 20) \
         ORDER BY age ASC",
    )
    .expect("searched CASE explain parity right SQL should succeed");

    assert_eq!(
        left, right,
        "searched CASE WHERE explain parity should now preserve the exact same public explain surface",
    );
}

#[test]
fn explain_sql_where_coalesce_and_nullif_surfaces_filter_expr_with_fallback_predicate() {
    reset_session_sql_store();
    let session = sql_session();

    let explain = statement_explain_sql::<SessionSqlEntity>(
        &session,
        "EXPLAIN SELECT name \
         FROM SessionSqlEntity \
         WHERE COALESCE(NULLIF(age, 20), 99) = 99 \
         ORDER BY age ASC",
    )
    .expect("COALESCE/NULLIF WHERE EXPLAIN should succeed");

    assert_explain_contains_tokens(
        explain.as_str(),
        &[
            "filter_expr=Some(\"COALESCE(NULLIF(age, 20), 99) = 99\")",
            "predicate=None",
        ],
        "COALESCE/NULLIF WHERE explain should expose semantic filter ownership without claiming one derived predicate shape",
    );
}

#[test]
fn explain_sql_where_text_predicate_constant_arguments_surface_filter_expr_and_predicate() {
    reset_session_sql_store();
    let session = sql_session();

    let explain = statement_explain_sql::<SessionSqlEntity>(
        &session,
        "EXPLAIN SELECT name \
         FROM SessionSqlEntity \
         WHERE STARTS_WITH(name, TRIM('Al')) \
         ORDER BY age ASC",
    )
    .expect("text predicate constant arguments WHERE EXPLAIN should succeed");

    assert_explain_contains_tokens(
        explain.as_str(),
        &[
            "filter_expr=Some(\"STARTS_WITH(name, ",
            "predicate=Compare { field: \"name\"",
            "op: StartsWith",
            "Text(\"Al\")",
        ],
        "text predicate constant arguments WHERE explain should expose both semantic filter ownership and the folded derived predicate",
    );
}

#[test]
fn explain_sql_where_compare_constant_arguments_surface_filter_expr_and_predicate() {
    reset_session_sql_store();
    let session = sql_session();

    let explain = statement_explain_sql::<SessionSqlEntity>(
        &session,
        "EXPLAIN SELECT name \
         FROM SessionSqlEntity \
         WHERE name = TRIM('alpha') \
         ORDER BY age ASC",
    )
    .expect("compare constant arguments WHERE EXPLAIN should succeed");

    assert_explain_contains_tokens(
        explain.as_str(),
        &[
            "filter_expr=Some(\"name = ",
            "predicate=Compare { field: \"name\"",
            "op: Eq",
            "Text(\"alpha\")",
        ],
        "compare constant arguments WHERE explain should expose both semantic filter ownership and the folded derived predicate",
    );
}

#[test]
fn explain_sql_where_casefold_compare_constant_arguments_surface_filter_expr_and_predicate() {
    reset_session_sql_store();
    let session = sql_session();

    let explain = statement_explain_sql::<SessionSqlEntity>(
        &session,
        "EXPLAIN SELECT name \
         FROM SessionSqlEntity \
         WHERE LOWER(name) = TRIM('ALPHA') \
         ORDER BY age ASC",
    )
    .expect("casefold compare constant arguments WHERE EXPLAIN should succeed");

    assert_explain_contains_tokens(
        explain.as_str(),
        &[
            "filter_expr=Some(\"LOWER(name) = ",
            "predicate=Compare { field: \"name\"",
            "op: Eq",
            "Text(\"ALPHA\")",
            "TextCasefold",
        ],
        "casefold compare constant arguments WHERE explain should expose both semantic filter ownership and the folded casefold predicate",
    );
}

#[test]
fn explain_sql_where_compare_and_true_constant_arguments_surface_filter_expr_and_predicate() {
    reset_session_sql_store();
    let session = sql_session();

    let explain = statement_explain_sql::<SessionSqlEntity>(
        &session,
        "EXPLAIN SELECT name \
         FROM SessionSqlEntity \
         WHERE name = TRIM('alpha') AND NULLIF('alpha', 'alpha') IS NULL \
         ORDER BY age ASC",
    )
    .expect("compare and true constant arguments WHERE EXPLAIN should succeed");

    assert_explain_contains_tokens(
        explain.as_str(),
        &[
            "filter_expr=Some(\"name = ",
            "predicate=Compare { field: \"name\"",
            "op: Eq",
            "Text(\"alpha\")",
        ],
        "compare and true constant arguments WHERE explain should expose the simplified semantic filter and recovered derived predicate",
    );
}

#[test]
fn explain_sql_where_compare_and_false_constant_arguments_surface_folded_false_predicate() {
    reset_session_sql_store();
    let session = sql_session();

    let explain = statement_explain_sql::<SessionSqlEntity>(
        &session,
        "EXPLAIN SELECT name \
         FROM SessionSqlEntity \
         WHERE name = TRIM('alpha') AND NULLIF('alpha', 'alpha') IS NOT NULL \
         ORDER BY age ASC",
    )
    .expect("compare and false constant arguments WHERE EXPLAIN should succeed");

    assert_explain_contains_tokens(
        explain.as_str(),
        &["filter_expr=Some(\"FALSE\")", "predicate=False"],
        "compare and false constant arguments WHERE explain should expose the folded FALSE filter expression and recovered FALSE predicate",
    );
}

#[test]
fn explain_sql_where_compare_or_false_constant_arguments_surface_filter_expr_and_predicate() {
    reset_session_sql_store();
    let session = sql_session();

    let explain = statement_explain_sql::<SessionSqlEntity>(
        &session,
        "EXPLAIN SELECT name \
         FROM SessionSqlEntity \
         WHERE name = TRIM('alpha') OR NULLIF('alpha', 'alpha') IS NOT NULL \
         ORDER BY age ASC",
    )
    .expect("compare or false constant arguments WHERE EXPLAIN should succeed");

    assert_explain_contains_tokens(
        explain.as_str(),
        &[
            "filter_expr=Some(\"name = ",
            "predicate=Compare { field: \"name\"",
            "op: Eq",
            "Text(\"alpha\")",
        ],
        "compare or false constant arguments WHERE explain should expose the simplified semantic filter and recovered derived predicate",
    );
}

#[test]
fn explain_sql_where_compare_or_true_constant_arguments_surface_folded_true_filter_expr() {
    reset_session_sql_store();
    let session = sql_session();

    let explain = statement_explain_sql::<SessionSqlEntity>(
        &session,
        "EXPLAIN SELECT name \
         FROM SessionSqlEntity \
         WHERE name = TRIM('alpha') OR NULLIF('alpha', 'alpha') IS NULL \
         ORDER BY age ASC",
    )
    .expect("compare or true constant arguments WHERE EXPLAIN should succeed");

    assert_explain_contains_tokens(
        explain.as_str(),
        &["filter_expr=Some(\"TRUE\")", "predicate=None"],
        "compare or true constant arguments WHERE explain should expose the folded TRUE filter expression and current TRUE predicate storage behavior",
    );
}

#[test]
fn explain_sql_where_constant_null_test_arguments_surface_folded_boolean_predicates() {
    reset_session_sql_store();
    let session = sql_session();

    for (sql, filter_token, predicate_token, context) in [
        (
            "EXPLAIN SELECT name \
             FROM SessionSqlEntity \
             WHERE NULLIF('alpha', 'alpha') IS NULL \
             ORDER BY age ASC",
            "filter_expr=Some(\"TRUE\")",
            "predicate=None",
            "constant null-test WHERE that folds to TRUE",
        ),
        (
            "EXPLAIN SELECT name \
             FROM SessionSqlEntity \
             WHERE NULLIF('alpha', 'alpha') IS NOT NULL \
             ORDER BY age ASC",
            "filter_expr=Some(\"FALSE\")",
            "predicate=False",
            "constant null-test WHERE that folds to FALSE",
        ),
    ] {
        let explain = statement_explain_sql::<SessionSqlEntity>(&session, sql)
            .unwrap_or_else(|err| panic!("{context} explain should succeed: {err:?}"));

        assert_explain_contains_tokens(
            explain.as_str(),
            &[filter_token, predicate_token],
            "constant null-test WHERE explain should expose the folded boolean filter expression and the current derived predicate shape",
        );
    }
}

#[test]
fn explain_sql_where_casefold_text_predicate_and_true_constant_arguments_surface_filter_expr_and_predicate()
 {
    reset_session_sql_store();
    let session = sql_session();

    let explain = statement_explain_sql::<SessionSqlEntity>(
        &session,
        "EXPLAIN SELECT name \
         FROM SessionSqlEntity \
         WHERE STARTS_WITH(LOWER(name), TRIM('AL')) \
           AND NULLIF('alpha', 'alpha') IS NULL \
         ORDER BY age ASC",
    )
    .expect("casefold text predicate and true constant arguments WHERE EXPLAIN should succeed");

    assert_explain_contains_tokens(
        explain.as_str(),
        &[
            "filter_expr=Some(\"STARTS_WITH(LOWER(name), ",
            "predicate=Compare { field: \"name\"",
            "op: StartsWith",
            "Text(\"AL\")",
            "TextCasefold",
        ],
        "casefold text predicate and true constant arguments WHERE explain should expose the simplified semantic filter and recovered casefold starts-with predicate",
    );
}

#[test]
fn explain_sql_where_casefold_text_predicate_or_false_constant_arguments_surface_filter_expr_and_predicate()
 {
    reset_session_sql_store();
    let session = sql_session();

    let explain = statement_explain_sql::<SessionSqlEntity>(
        &session,
        "EXPLAIN SELECT name \
         FROM SessionSqlEntity \
         WHERE STARTS_WITH(LOWER(name), TRIM('AL')) \
           OR NULLIF('alpha', 'alpha') IS NOT NULL \
         ORDER BY age ASC",
    )
    .expect("casefold text predicate or false constant arguments WHERE EXPLAIN should succeed");

    assert_explain_contains_tokens(
        explain.as_str(),
        &[
            "filter_expr=Some(\"STARTS_WITH(LOWER(name), ",
            "predicate=Compare { field: \"name\"",
            "op: StartsWith",
            "Text(\"AL\")",
            "TextCasefold",
        ],
        "casefold text predicate or false constant arguments WHERE explain should expose the simplified semantic filter and recovered casefold starts-with predicate",
    );
}

#[test]
fn explain_sql_where_casefold_text_predicate_constant_arguments_surface_filter_expr_and_predicate()
{
    reset_session_sql_store();
    let session = sql_session();

    let explain = statement_explain_sql::<SessionSqlEntity>(
        &session,
        "EXPLAIN SELECT name \
         FROM SessionSqlEntity \
         WHERE STARTS_WITH(LOWER(name), TRIM('AL')) \
         ORDER BY age ASC",
    )
    .expect("casefold text predicate constant arguments WHERE EXPLAIN should succeed");

    assert_explain_contains_tokens(
        explain.as_str(),
        &[
            "filter_expr=Some(\"STARTS_WITH(LOWER(name), ",
            "predicate=Compare { field: \"name\"",
            "op: StartsWith",
            "Text(\"AL\")",
            "TextCasefold",
        ],
        "casefold text predicate constant arguments WHERE explain should expose both semantic filter ownership and the folded casefold starts-with predicate",
    );
}

#[test]
fn explain_sql_delete_wrapped_like_and_ilike_surface_filter_expr_with_fallback_predicate() {
    reset_session_sql_store();
    let session = sql_session();

    for (sql, context) in [
        (
            "EXPLAIN DELETE FROM SessionSqlEntity \
             WHERE REPLACE(name, 'a', 'A') LIKE 'Al%' \
             ORDER BY age ASC LIMIT 1",
            "wrapped LIKE delete explain",
        ),
        (
            "EXPLAIN DELETE FROM SessionSqlEntity \
             WHERE REPLACE(name, 'a', 'A') ILIKE 'al%' \
             ORDER BY age ASC LIMIT 1",
            "wrapped ILIKE delete explain",
        ),
    ] {
        let explain = statement_explain_sql::<SessionSqlEntity>(&session, sql)
            .unwrap_or_else(|err| panic!("{context} should succeed: {err:?}"));

        assert_explain_contains_tokens(
            explain.as_str(),
            &[
                "mode=Delete",
                "filter_expr=Some(\"STARTS_WITH(",
                "predicate=None",
            ],
            "wrapped LIKE/ILIKE delete explain should expose semantic filter ownership without claiming one derived predicate shape",
        );
    }
}

#[test]
fn explain_sql_delete_text_predicate_expression_arguments_surface_filter_expr_with_fallback_predicate()
 {
    reset_session_sql_store();
    let session = sql_session();

    let explain = statement_explain_sql::<SessionSqlEntity>(
        &session,
        "EXPLAIN DELETE FROM SessionSqlEntity \
         WHERE STARTS_WITH(REPLACE(name, 'a', 'A'), TRIM('Al')) \
         ORDER BY age ASC LIMIT 1",
    )
    .expect("text predicate expression arguments delete explain should succeed");

    assert_explain_contains_tokens(
        explain.as_str(),
        &[
            "mode=Delete",
            "filter_expr=Some(\"STARTS_WITH(",
            "predicate=None",
        ],
        "text predicate expression arguments delete explain should expose semantic filter ownership without claiming one derived predicate shape",
    );
}

#[test]
fn explain_sql_delete_compare_boolean_constant_arguments_surface_recovered_predicates() {
    reset_session_sql_store();
    let session = sql_session();

    for (sql, expected_tokens, context) in [
        (
            "EXPLAIN DELETE FROM SessionSqlEntity \
             WHERE name = TRIM('alpha') OR NULLIF('alpha', 'alpha') IS NOT NULL \
             ORDER BY age ASC LIMIT 1",
            vec![
                "mode=Delete",
                "filter_expr=Some(\"name = ",
                "predicate=Compare { field: \"name\"",
                "op: Eq",
                "Text(\"alpha\")",
            ],
            "compare OR FALSE delete explain",
        ),
        (
            "EXPLAIN DELETE FROM SessionSqlEntity \
             WHERE name = TRIM('alpha') OR NULLIF('alpha', 'alpha') IS NULL \
             ORDER BY age ASC LIMIT 1",
            vec![
                "mode=Delete",
                "filter_expr=Some(\"TRUE\")",
                "predicate=None",
            ],
            "compare OR TRUE delete explain",
        ),
    ] {
        let explain = statement_explain_sql::<SessionSqlEntity>(&session, sql)
            .unwrap_or_else(|err| panic!("{context} should succeed: {err:?}"));

        assert_explain_contains_tokens(
            explain.as_str(),
            &expected_tokens,
            "delete explain should expose the recovered boolean-simplified filter and current predicate storage behavior",
        );
    }
}

#[test]
fn explain_sql_delete_casefold_text_predicate_boolean_constant_arguments_surface_recovered_predicates()
 {
    reset_session_sql_store();
    let session = sql_session();

    for (sql, expected_tokens, context) in [
        (
            "EXPLAIN DELETE FROM SessionSqlEntity \
             WHERE STARTS_WITH(LOWER(name), TRIM('AL')) \
               OR NULLIF('alpha', 'alpha') IS NOT NULL \
             ORDER BY age ASC LIMIT 1",
            vec![
                "mode=Delete",
                "filter_expr=Some(\"STARTS_WITH(LOWER(name), ",
                "predicate=Compare { field: \"name\"",
                "op: StartsWith",
                "Text(\"AL\")",
                "TextCasefold",
            ],
            "casefold text predicate OR FALSE delete explain",
        ),
        (
            "EXPLAIN DELETE FROM SessionSqlEntity \
             WHERE STARTS_WITH(LOWER(name), TRIM('AL')) \
               OR NULLIF('alpha', 'alpha') IS NULL \
             ORDER BY age ASC LIMIT 1",
            vec![
                "mode=Delete",
                "filter_expr=Some(\"TRUE\")",
                "predicate=None",
            ],
            "casefold text predicate OR TRUE delete explain",
        ),
    ] {
        let explain = statement_explain_sql::<SessionSqlEntity>(&session, sql)
            .unwrap_or_else(|err| panic!("{context} should succeed: {err:?}"));

        assert_explain_contains_tokens(
            explain.as_str(),
            &expected_tokens,
            "delete explain should expose the recovered casefold boolean-simplified filter and current predicate storage behavior",
        );
    }
}

#[test]
fn explain_sql_order_by_supported_scalar_text_aliases_match_canonical_plan_output() {
    reset_session_sql_store();
    let session = sql_session();

    for (alias_sql, canonical_sql, context) in [
        (
            "EXPLAIN SELECT TRIM(name) AS trimmed_name FROM SessionSqlEntity ORDER BY trimmed_name ASC LIMIT 1",
            "EXPLAIN SELECT TRIM(name) FROM SessionSqlEntity ORDER BY TRIM(name) ASC LIMIT 1",
            "ORDER BY TRIM alias",
        ),
        (
            "EXPLAIN SELECT LTRIM(name) AS left_trimmed_name FROM SessionSqlEntity ORDER BY left_trimmed_name ASC LIMIT 1",
            "EXPLAIN SELECT LTRIM(name) FROM SessionSqlEntity ORDER BY LTRIM(name) ASC LIMIT 1",
            "ORDER BY LTRIM alias",
        ),
        (
            "EXPLAIN SELECT RTRIM(name) AS right_trimmed_name FROM SessionSqlEntity ORDER BY right_trimmed_name ASC LIMIT 1",
            "EXPLAIN SELECT RTRIM(name) FROM SessionSqlEntity ORDER BY RTRIM(name) ASC LIMIT 1",
            "ORDER BY RTRIM alias",
        ),
        (
            "EXPLAIN SELECT LENGTH(name) AS name_len FROM SessionSqlEntity ORDER BY name_len DESC LIMIT 1",
            "EXPLAIN SELECT LENGTH(name) FROM SessionSqlEntity ORDER BY LENGTH(name) DESC LIMIT 1",
            "ORDER BY LENGTH alias",
        ),
        (
            "EXPLAIN SELECT LEFT(name, 2) AS short_name FROM SessionSqlEntity ORDER BY short_name ASC LIMIT 1",
            "EXPLAIN SELECT LEFT(name, 2) FROM SessionSqlEntity ORDER BY LEFT(name, 2) ASC LIMIT 1",
            "ORDER BY LEFT alias",
        ),
        (
            "EXPLAIN SELECT TRIM(name) AS trimmed_name, ROUND((age + age) / (age + 1), 2) AS normalized_age FROM SessionSqlEntity ORDER BY trimmed_name ASC, normalized_age DESC LIMIT 1",
            "EXPLAIN SELECT TRIM(name), ROUND((age + age) / (age + 1), 2) FROM SessionSqlEntity ORDER BY TRIM(name) ASC, ROUND((age + age) / (age + 1), 2) DESC LIMIT 1",
            "mixed TRIM plus nested ROUND alias ordering",
        ),
    ] {
        assert_session_sql_alias_matches_canonical::<String>(
            &session,
            statement_explain_sql::<SessionSqlEntity>,
            alias_sql,
            canonical_sql,
            context,
        );
    }
}

#[test]
fn explain_sql_accepts_order_by_bounded_numeric_aliases() {
    reset_session_sql_store();
    let session = sql_session();

    for sql in [
        "EXPLAIN SELECT age + 1 AS next_age FROM SessionSqlEntity ORDER BY next_age ASC LIMIT 1",
        "EXPLAIN SELECT ROUND(age / 3, 2) AS rounded_age FROM SessionSqlEntity ORDER BY rounded_age DESC LIMIT 1",
    ] {
        assert_explain_load_shape_case::<SessionSqlEntity>(
            &session,
            sql,
            "bounded numeric ORDER BY alias explain",
        );
    }

    for sql in [
        "EXPLAIN SELECT rank + rank AS total FROM SessionAggregateEntity ORDER BY total ASC LIMIT 1",
        "EXPLAIN SELECT ROUND(rank + rank, 2) AS rounded_total FROM SessionAggregateEntity ORDER BY rounded_total DESC LIMIT 1",
    ] {
        assert_explain_load_shape_case::<SessionAggregateEntity>(
            &session,
            sql,
            "bounded numeric ORDER BY alias explain",
        );
    }
}

#[test]
fn explain_sql_accepts_direct_bounded_numeric_order_terms() {
    reset_session_sql_store();
    let session = sql_session();

    for sql in [
        "EXPLAIN SELECT age FROM SessionSqlEntity ORDER BY age + 1 ASC LIMIT 1",
        "EXPLAIN SELECT age FROM SessionSqlEntity ORDER BY ROUND(age / 3, 2) DESC LIMIT 1",
    ] {
        assert_explain_load_shape_case::<SessionSqlEntity>(
            &session,
            sql,
            "direct bounded numeric ORDER BY explain",
        );
    }
}

#[test]
fn explain_sql_text_specific_computed_projection_matrix_preserves_surface_contracts() {
    reset_session_sql_store();
    let session = sql_session();

    let scalar_explain = statement_explain_sql::<SessionSqlEntity>(
        &session,
        "EXPLAIN SELECT TRIM(name) FROM SessionSqlEntity ORDER BY age LIMIT 1",
    )
    .expect(
        "EXPLAIN should support text-specific computed projection on the narrowed statement lane",
    );
    assert!(
        scalar_explain.contains("mode=Load"),
        "text-specific computed projection explain should still render the base load plan",
    );
    assert!(
        scalar_explain.contains("access="),
        "text-specific computed projection explain should still expose the routed access shape",
    );

    let grouped_explain = statement_explain_sql::<SessionSqlEntity>(
        &session,
        "EXPLAIN SELECT TRIM(name), COUNT(*) \
         FROM SessionSqlEntity \
         GROUP BY name \
         ORDER BY name ASC LIMIT 10",
    )
    .expect("EXPLAIN should support grouped text-specific computed projection over grouped fields");
    assert!(
        grouped_explain.contains("grouping="),
        "grouped computed SQL projection explain should still expose grouped planning",
    );
    assert!(
        grouped_explain.contains("mode=Load"),
        "grouped computed SQL projection explain should still render the base load plan",
    );

    let (left_sql, right_sql, context) = (
        "EXPLAIN SELECT DISTINCT age, COUNT(*) FROM SessionSqlEntity GROUP BY age",
        "EXPLAIN SELECT age, COUNT(*) FROM SessionSqlEntity GROUP BY age",
        "top-level grouped SELECT DISTINCT explain",
    );
    assert_explain_exact_equivalence_case::<SessionSqlEntity>(
        &session, left_sql, right_sql, context,
    );
}

#[test]
fn explain_sql_grouped_additive_order_terms_preserve_surface_contracts() {
    reset_session_sql_store();
    let session = sql_session();

    for sql in [
        "EXPLAIN SELECT age, COUNT(*) \
         FROM SessionSqlEntity \
         GROUP BY age \
         ORDER BY age + 1 ASC LIMIT 10",
        "EXPLAIN SELECT age + 1 AS next_age, COUNT(*) \
         FROM SessionSqlEntity \
         GROUP BY age \
         ORDER BY next_age ASC LIMIT 10",
    ] {
        let explain = statement_explain_sql::<SessionSqlEntity>(&session, sql)
            .expect("grouped additive ORDER BY explain should succeed");
        assert!(
            explain.contains("grouping="),
            "grouped additive ORDER BY explain should still expose grouped planning",
        );
        assert!(
            explain.contains("age + 1") || explain.contains("next_age"),
            "grouped additive ORDER BY explain should preserve the requested computed order surface",
        );
    }
}

#[test]
fn explain_sql_grouped_filter_aggregate_surfaces_filter_shape_across_plan_and_json() {
    reset_session_sql_store();
    let session = sql_session();

    // Phase 1: run one grouped filtered aggregate explain through the plain
    // text surface and require the grouped aggregate filter shape to stay
    // visible to the public explain contract.
    let explain = statement_explain_sql::<SessionSqlEntity>(
        &session,
        "EXPLAIN SELECT age, COUNT(*) FILTER (WHERE age >= 20), COUNT(*) \
         FROM SessionSqlEntity \
         GROUP BY age \
         HAVING COUNT(*) FILTER (WHERE age >= 20) > 0 \
         ORDER BY COUNT(*) FILTER (WHERE age >= 20) DESC, age ASC LIMIT 10",
    )
    .expect("grouped aggregate FILTER EXPLAIN should succeed");
    assert_explain_contains_tokens(
        explain.as_str(),
        &[
            "grouping=Grouped",
            "filter_expr: Some(\"age >= 20\")",
            "having: Some(",
        ],
        "grouped aggregate FILTER explain should keep filter and HAVING shape visible",
    );

    // Phase 2: require the JSON explain surface to expose the same filtered
    // aggregate shape instead of dropping it during serialization.
    let explain_json = statement_explain_sql::<SessionSqlEntity>(
        &session,
        "EXPLAIN JSON SELECT age, COUNT(*) FILTER (WHERE age >= 20), COUNT(*) \
         FROM SessionSqlEntity \
         GROUP BY age \
         HAVING COUNT(*) FILTER (WHERE age >= 20) > 0 \
         ORDER BY COUNT(*) FILTER (WHERE age >= 20) DESC, age ASC LIMIT 10",
    )
    .expect("grouped aggregate FILTER EXPLAIN JSON should succeed");
    assert_explain_contains_tokens(
        explain_json.as_str(),
        &[
            "\"grouping\"",
            "filter_expr: Some(\\\"age >= 20\\\")",
            "having: Some(",
        ],
        "grouped aggregate FILTER explain JSON should keep filter and HAVING shape visible",
    );
}

#[test]
fn explain_sql_grouped_boolean_searched_case_having_uses_canonical_semantic_shape() {
    reset_session_sql_store();
    let session = sql_session();

    let explain = statement_explain_sql::<SessionSqlEntity>(
        &session,
        "EXPLAIN SELECT age, COUNT(*) \
         FROM SessionSqlEntity \
         GROUP BY age \
         HAVING CASE WHEN COUNT(*) > 1 THEN TRUE ELSE FALSE END \
         ORDER BY age ASC LIMIT 10",
    )
    .expect("grouped boolean searched CASE HAVING EXPLAIN should succeed");

    assert_explain_contains_tokens(
        explain.as_str(),
        &[
            "grouping=Grouped",
            "having: Some(",
            "FunctionCall { function: Coalesce",
            "Literal(Bool(false))",
        ],
        "grouped boolean searched CASE HAVING explain should surface the canonical grouped semantic form",
    );
    assert!(
        !explain.contains("Expr::Case") && !explain.contains("Case {"),
        "grouped boolean searched CASE HAVING explain should not fall back to the original CASE spelling once canonicalized",
    );
}

#[test]
fn explain_sql_grouped_boolean_searched_case_truth_wrapper_matches_canonical_output() {
    reset_session_sql_store();
    let session = sql_session();

    assert_explain_exact_equivalence_case::<SessionSqlEntity>(
        &session,
        "EXPLAIN SELECT age, COUNT(*) \
         FROM SessionSqlEntity \
         GROUP BY age \
         HAVING CASE WHEN COUNT(*) > 1 THEN TRUE ELSE FALSE END \
         ORDER BY age ASC LIMIT 10",
        "EXPLAIN SELECT age, COUNT(*) \
         FROM SessionSqlEntity \
         GROUP BY age \
         HAVING CASE WHEN (COUNT(*) > 1) = TRUE THEN TRUE ELSE FALSE END \
         ORDER BY age ASC LIMIT 10",
        "grouped boolean searched CASE truth wrappers should keep the exact same EXPLAIN output once canonicalized",
    );
}

#[test]
fn explain_sql_grouped_boolean_searched_case_without_else_matches_explicit_null_output() {
    reset_session_sql_store();
    let session = sql_session();

    assert_explain_exact_equivalence_case::<SessionSqlEntity>(
        &session,
        "EXPLAIN SELECT age, COUNT(*) \
         FROM SessionSqlEntity \
         GROUP BY age \
         HAVING CASE WHEN COUNT(*) > 1 THEN TRUE END \
         ORDER BY age ASC LIMIT 10",
        "EXPLAIN SELECT age, COUNT(*) \
         FROM SessionSqlEntity \
         GROUP BY age \
         HAVING CASE WHEN COUNT(*) > 1 THEN TRUE ELSE NULL END \
         ORDER BY age ASC LIMIT 10",
        "grouped boolean searched CASE without ELSE should keep the same EXPLAIN output as the explicit ELSE NULL grouped boolean family",
    );
}

#[test]
fn explain_sql_grouped_boolean_searched_case_without_else_truth_wrapper_matches_explicit_null_output()
 {
    reset_session_sql_store();
    let session = sql_session();

    assert_explain_exact_equivalence_case::<SessionSqlEntity>(
        &session,
        "EXPLAIN SELECT age, COUNT(*) \
         FROM SessionSqlEntity \
         GROUP BY age \
         HAVING CASE WHEN COUNT(*) > 1 THEN TRUE ELSE NULL END \
         ORDER BY age ASC LIMIT 10",
        "EXPLAIN SELECT age, COUNT(*) \
         FROM SessionSqlEntity \
         GROUP BY age \
         HAVING CASE WHEN (COUNT(*) > 1) = TRUE THEN TRUE END \
         ORDER BY age ASC LIMIT 10",
        "grouped boolean searched CASE without ELSE should keep the same EXPLAIN output as the explicit ELSE NULL grouped boolean family even when the admitted WHEN condition carries a redundant truth wrapper",
    );
}

#[test]
fn explain_sql_grouped_boolean_searched_case_without_else_uses_canonical_null_family_shape() {
    reset_session_sql_store();
    let session = sql_session();

    let explain = statement_explain_sql::<SessionSqlEntity>(
        &session,
        "EXPLAIN SELECT age, COUNT(*) \
         FROM SessionSqlEntity \
         GROUP BY age \
         HAVING CASE WHEN COUNT(*) > 1 THEN TRUE END \
         ORDER BY age ASC LIMIT 10",
    )
    .expect("grouped searched CASE HAVING without ELSE EXPLAIN should succeed");

    assert_explain_contains_tokens(
        explain.as_str(),
        &["grouping=Grouped", "having: Some(", "Literal(Null)"],
        "grouped searched CASE HAVING without ELSE should surface the grouped null-family canonical explain shape",
    );
    assert!(
        !explain.contains("Expr::Case") && !explain.contains("Case {"),
        "grouped searched CASE HAVING without ELSE should not keep the raw CASE surface once it joins the grouped null-family canonical form: {explain}",
    );
}

#[test]
fn explain_sql_global_aggregate_having_without_else_matches_explicit_null_output() {
    reset_session_sql_store();
    let session = sql_session();

    assert_explain_exact_equivalence_case::<SessionSqlEntity>(
        &session,
        "EXPLAIN SELECT COUNT(*) \
         FROM SessionSqlEntity \
         HAVING CASE WHEN COUNT(*) > 1 THEN TRUE END",
        "EXPLAIN SELECT COUNT(*) \
         FROM SessionSqlEntity \
         HAVING CASE WHEN COUNT(*) > 1 THEN TRUE ELSE NULL END",
        "global aggregate omitted-ELSE grouped boolean HAVING should keep the same EXPLAIN output as the explicit ELSE NULL family",
    );
}

#[test]
fn explain_sql_global_aggregate_having_without_else_truth_wrapper_matches_explicit_null_output() {
    reset_session_sql_store();
    let session = sql_session();

    assert_explain_exact_equivalence_case::<SessionSqlEntity>(
        &session,
        "EXPLAIN SELECT COUNT(*) \
         FROM SessionSqlEntity \
         HAVING CASE WHEN COUNT(*) > 1 THEN TRUE ELSE NULL END",
        "EXPLAIN SELECT COUNT(*) \
         FROM SessionSqlEntity \
         HAVING CASE WHEN (COUNT(*) > 1) = TRUE THEN TRUE END",
        "global aggregate omitted-ELSE grouped boolean HAVING should keep the same EXPLAIN output as the explicit ELSE NULL family even when the admitted WHEN condition carries a redundant truth wrapper",
    );
}

#[test]
fn explain_sql_scalar_where_surfaces_filter_expr_and_predicate_across_plan_and_json() {
    reset_session_sql_store();
    let session = sql_session();

    let explain = statement_explain_sql::<SessionSqlEntity>(
        &session,
        "EXPLAIN SELECT * FROM SessionSqlEntity WHERE age >= 20 ORDER BY id ASC LIMIT 5",
    )
    .expect("scalar WHERE EXPLAIN should succeed");
    assert_explain_contains_tokens(
        explain.as_str(),
        &[
            "filter_expr=Some(\"age >= 20\")",
            "predicate=Compare { field: \"age\"",
            "op: Gte",
        ],
        "scalar WHERE explain should expose semantic filter expression and derived predicate separately",
    );

    let explain_json = statement_explain_sql::<SessionSqlEntity>(
        &session,
        "EXPLAIN JSON SELECT * FROM SessionSqlEntity WHERE age >= 20 ORDER BY id ASC LIMIT 5",
    )
    .expect("scalar WHERE EXPLAIN JSON should succeed");
    assert_explain_contains_tokens(
        explain_json.as_str(),
        &[
            "\"filter_expr\":\"age >= 20\"",
            "\"predicate\":\"Compare { field: \\\"age\\\"",
            "op: Gte",
        ],
        "scalar WHERE explain JSON should expose semantic filter expression and derived predicate separately",
    );
}

#[test]
fn explain_sql_grouped_where_surfaces_filter_expr_and_predicate_across_plan_and_json() {
    reset_session_sql_store();
    let session = sql_session();

    let explain = statement_explain_sql::<SessionSqlEntity>(
        &session,
        "EXPLAIN SELECT age, COUNT(*) \
         FROM SessionSqlEntity \
         WHERE CASE WHEN age >= 30 THEN TRUE ELSE age = 20 END \
         GROUP BY age \
         ORDER BY age ASC LIMIT 5",
    )
    .expect("grouped WHERE EXPLAIN should succeed");
    assert_explain_contains_tokens(
        explain.as_str(),
        &[
            "grouping=Grouped",
            "filter_expr=Some(\"CASE WHEN age >= 30 THEN TRUE ELSE age = 20 END\")",
            "predicate=Or([",
        ],
        "grouped WHERE explain should expose semantic filter expression and derived predicate separately",
    );

    let explain_json = statement_explain_sql::<SessionSqlEntity>(
        &session,
        "EXPLAIN JSON SELECT age, COUNT(*) \
         FROM SessionSqlEntity \
         WHERE CASE WHEN age >= 30 THEN TRUE ELSE age = 20 END \
         GROUP BY age \
         ORDER BY age ASC LIMIT 5",
    )
    .expect("grouped WHERE EXPLAIN JSON should succeed");
    assert_explain_contains_tokens(
        explain_json.as_str(),
        &[
            "\"grouping\"",
            "\"filter_expr\":\"CASE WHEN age >= 30 THEN TRUE ELSE age = 20 END\"",
            "\"predicate\":\"Or([",
        ],
        "grouped WHERE explain JSON should expose semantic filter expression and derived predicate separately",
    );
}

#[test]
fn explain_sql_grouped_where_coalesce_and_nullif_surfaces_filter_expr_with_fallback_predicate() {
    reset_session_sql_store();
    let session = sql_session();

    let explain = statement_explain_sql::<SessionSqlEntity>(
        &session,
        "EXPLAIN SELECT age, COUNT(*) \
         FROM SessionSqlEntity \
         WHERE COALESCE(NULLIF(age, 20), 99) = 99 \
         GROUP BY age \
         ORDER BY age ASC LIMIT 5",
    )
    .expect("grouped COALESCE/NULLIF WHERE EXPLAIN should succeed");

    assert_explain_contains_tokens(
        explain.as_str(),
        &[
            "grouping=Grouped",
            "filter_expr=Some(\"COALESCE(NULLIF(age, 20), 99) = 99\")",
            "predicate=None",
        ],
        "grouped COALESCE/NULLIF WHERE explain should expose semantic filter ownership without claiming one derived predicate shape",
    );
}

#[test]
fn explain_sql_grouped_where_compare_boolean_constant_arguments_surface_recovered_predicates() {
    reset_session_sql_store();
    let session = sql_session();

    for (sql, expected_filter_token, expected_predicate_token, context) in [
        (
            "EXPLAIN SELECT age, COUNT(*) \
             FROM SessionSqlEntity \
             WHERE name = TRIM('alpha') OR NULLIF('alpha', 'alpha') IS NOT NULL \
             GROUP BY age \
             ORDER BY age ASC LIMIT 5",
            "filter_expr=Some(\"name = ",
            "predicate=Compare { field: \"name\"",
            "grouped compare OR FALSE explain",
        ),
        (
            "EXPLAIN SELECT age, COUNT(*) \
             FROM SessionSqlEntity \
             WHERE name = TRIM('alpha') OR NULLIF('alpha', 'alpha') IS NULL \
             GROUP BY age \
             ORDER BY age ASC LIMIT 5",
            "filter_expr=Some(\"TRUE\")",
            "predicate=None",
            "grouped compare OR TRUE explain",
        ),
    ] {
        let explain = statement_explain_sql::<SessionSqlEntity>(&session, sql)
            .unwrap_or_else(|err| panic!("{context} should succeed: {err:?}"));

        assert_explain_contains_tokens(
            explain.as_str(),
            &[
                "grouping=Grouped",
                expected_filter_token,
                expected_predicate_token,
            ],
            "grouped explain should expose the recovered boolean-simplified filter and predicate state",
        );
    }

    let explain_json = statement_explain_sql::<SessionSqlEntity>(
        &session,
        "EXPLAIN JSON SELECT age, COUNT(*) \
         FROM SessionSqlEntity \
         WHERE name = TRIM('alpha') OR NULLIF('alpha', 'alpha') IS NOT NULL \
         GROUP BY age \
         ORDER BY age ASC LIMIT 5",
    )
    .expect("grouped compare OR FALSE EXPLAIN JSON should succeed");
    assert_explain_contains_tokens(
        explain_json.as_str(),
        &[
            "\"grouping\"",
            "\"filter_expr\":\"name = ",
            "\"predicate\":\"Compare { field: \\\"name\\\"",
        ],
        "grouped compare OR FALSE explain JSON should expose the recovered boolean-simplified filter and predicate state",
    );
}

#[test]
fn explain_sql_grouped_where_casefold_text_predicate_boolean_constant_arguments_surface_recovered_predicates()
 {
    reset_session_sql_store();
    let session = sql_session();

    for (sql, expected_filter_token, expected_predicate_token, context) in [
        (
            "EXPLAIN SELECT age, COUNT(*) \
             FROM SessionSqlEntity \
             WHERE STARTS_WITH(LOWER(name), TRIM('AL')) \
               OR NULLIF('alpha', 'alpha') IS NOT NULL \
             GROUP BY age \
             ORDER BY age ASC LIMIT 5",
            "filter_expr=Some(\"STARTS_WITH(LOWER(name), ",
            "predicate=Compare { field: \"name\"",
            "grouped casefold text predicate OR FALSE explain",
        ),
        (
            "EXPLAIN SELECT age, COUNT(*) \
             FROM SessionSqlEntity \
             WHERE STARTS_WITH(LOWER(name), TRIM('AL')) \
               OR NULLIF('alpha', 'alpha') IS NULL \
             GROUP BY age \
             ORDER BY age ASC LIMIT 5",
            "filter_expr=Some(\"TRUE\")",
            "predicate=None",
            "grouped casefold text predicate OR TRUE explain",
        ),
    ] {
        let explain = statement_explain_sql::<SessionSqlEntity>(&session, sql)
            .unwrap_or_else(|err| panic!("{context} should succeed: {err:?}"));

        assert_explain_contains_tokens(
            explain.as_str(),
            &[
                "grouping=Grouped",
                expected_filter_token,
                expected_predicate_token,
            ],
            "grouped explain should expose the recovered casefold boolean-simplified filter and predicate state",
        );
    }
}

#[test]
fn explain_sql_grouped_where_casefold_compare_boolean_constant_arguments_surface_recovered_predicates()
 {
    reset_session_sql_store();
    let session = sql_session();

    let explain = statement_explain_sql::<SessionSqlEntity>(
        &session,
        "EXPLAIN SELECT age, COUNT(*) \
         FROM SessionSqlEntity \
         WHERE LOWER(name) = TRIM('ALPHA') \
           OR NULLIF('alpha', 'alpha') IS NOT NULL \
         GROUP BY age \
         ORDER BY age ASC LIMIT 5",
    )
    .expect("grouped casefold compare OR FALSE explain should succeed");

    assert_explain_contains_tokens(
        explain.as_str(),
        &[
            "grouping=Grouped",
            "filter_expr=Some(\"LOWER(name) = ",
            "predicate=Compare { field: \"name\"",
            "op: Eq",
            "Text(\"ALPHA\")",
            "TextCasefold",
        ],
        "grouped casefold compare OR FALSE explain should expose the recovered casefold compare filter and predicate state",
    );
}

#[test]
fn explain_sql_grouped_aggregate_order_alias_matches_canonical_plan_output() {
    reset_indexed_session_sql_store();
    let session = indexed_sql_session();

    assert_session_sql_alias_matches_canonical::<String>(
        &session,
        statement_explain_sql::<IndexedSessionSqlEntity>,
        "EXPLAIN SELECT name, AVG(age) AS avg_age \
         FROM IndexedSessionSqlEntity \
         GROUP BY name \
         ORDER BY avg_age DESC, name ASC LIMIT 2",
        "EXPLAIN SELECT name, AVG(age) \
         FROM IndexedSessionSqlEntity \
         GROUP BY name \
         ORDER BY AVG(age) DESC, name ASC LIMIT 2",
        "grouped aggregate ORDER BY aliases",
    );
}

#[test]
fn explain_sql_grouped_aggregate_input_order_alias_matches_canonical_plan_output() {
    reset_indexed_session_sql_store();
    let session = indexed_sql_session();

    assert_session_sql_alias_matches_canonical::<String>(
        &session,
        statement_explain_sql::<IndexedSessionSqlEntity>,
        "EXPLAIN SELECT name, AVG(age + 1) AS avg_plus_one \
         FROM IndexedSessionSqlEntity \
         GROUP BY name \
         ORDER BY avg_plus_one DESC, name ASC LIMIT 2",
        "EXPLAIN SELECT name, AVG(age + 1) \
         FROM IndexedSessionSqlEntity \
         GROUP BY name \
         ORDER BY AVG(age + 1) DESC, name ASC LIMIT 2",
        "grouped aggregate input ORDER BY aliases",
    );
}

#[test]
fn explain_sql_rejects_non_explain_statements() {
    reset_session_sql_store();
    let session = sql_session();

    let err = statement_explain_sql::<SessionSqlEntity>(&session, "SELECT * FROM SessionSqlEntity")
        .expect_err("explain_sql must reject non-EXPLAIN statements");

    assert!(
        matches!(
            err,
            QueryError::Execute(crate::db::query::intent::QueryExecutionError::Unsupported(
                _
            ))
        ),
        "non-EXPLAIN input must fail as unsupported explain usage",
    );
}

#[test]
fn explain_sql_field_to_field_predicate_stays_visible_in_predicate_tree() {
    reset_indexed_session_sql_store();
    let session = indexed_sql_session();

    let explain = statement_explain_sql::<SessionDeterministicRangeEntity>(
        &session,
        "EXPLAIN JSON SELECT label \
         FROM SessionDeterministicRangeEntity \
         WHERE tier = 'gold' AND score > 18 AND handle > label \
         ORDER BY score ASC, id ASC",
    )
    .expect("mixed literal and field-to-field EXPLAIN JSON should succeed");

    assert_explain_contains_tokens(
        explain.as_str(),
        &[
            "\"predicate\":\"And([Compare",
            "CompareFields { left_field: \\\"handle\\\", op: Gt, right_field: \\\"label\\\"",
        ],
        "field-to-field explain should keep the compare-fields predicate shape visible",
    );
}