kglite 0.16.4

use super::*;
use crate::graph::core::pattern_matching::pattern::{PropOp, RelEdgePredicate};
use crate::graph::core::pattern_matching::PropertyMatcher;
use crate::graph::languages::cypher::parser::parse_cypher;

#[test]
fn test_predicate_pushdown_simple() {
    let mut query = parse_cypher("MATCH (n:Person) WHERE n.age = 30 RETURN n").unwrap();

    let graph = DirGraph::new();
    let params = HashMap::new();
    optimize(&mut query, &graph, &params);

    // WHERE is kept as a safety net even when all predicates are pushed
    assert_eq!(query.clauses.len(), 3); // MATCH + WHERE + RETURN
    assert!(matches!(&query.clauses[0], Clause::Match(_)));
    assert!(matches!(&query.clauses[2], Clause::Return(_)));

    // The MATCH pattern should now have {age: 30} as a property
    if let Clause::Match(m) = &query.clauses[0] {
        if let PatternElement::Node(np) = &m.patterns[0].elements[0] {
            assert!(np.properties.is_some());
            let props = np.properties.as_ref().unwrap();
            assert!(props.contains_key("age"));
        } else {
            panic!("Expected node pattern");
        }
    }
}

#[test]
fn test_predicate_pushdown_partial() {
    let mut query =
        parse_cypher("MATCH (n:Person) WHERE n.age = 30 AND n.score > 100 RETURN n").unwrap();

    let graph = DirGraph::new();
    let params = HashMap::new();
    optimize(&mut query, &graph, &params);

    // Both n.age = 30 and n.score > 100 should be pushed into MATCH
    // WHERE is kept as a safety net
    assert_eq!(query.clauses.len(), 3); // MATCH + WHERE + RETURN

    if let Clause::Match(m) = &query.clauses[0] {
        if let PatternElement::Node(np) = &m.patterns[0].elements[0] {
            let props = np.properties.as_ref().unwrap();
            assert!(matches!(
                props.get("age"),
                Some(PropertyMatcher::Equals(Value::Int64(30)))
            ));
            assert!(matches!(
                props.get("score"),
                Some(PropertyMatcher::GreaterThan(Value::Int64(100)))
            ));
        }
    }
}

#[test]
fn test_predicate_pushdown_keeps_inline_property_collision_in_where() {
    let mut query = parse_cypher(
        "MATCH (n:Person {name: 'Alice'}) \
         WHERE n.name = 'Bob' AND size(n.name) > 0 RETURN n",
    )
    .unwrap();

    push_where_into_match(&mut query, &HashMap::new());

    let Clause::Match(m) = &query.clauses[0] else {
        panic!("expected MATCH clause");
    };
    let PatternElement::Node(node) = &m.patterns[0].elements[0] else {
        panic!("expected node pattern");
    };
    assert!(matches!(
        node.properties.as_ref().and_then(|props| props.get("name")),
        Some(PropertyMatcher::Equals(Value::String(value))) if value == "Alice"
    ));

    let Clause::Where(where_clause) = &query.clauses[1] else {
        panic!("expected residual WHERE clause");
    };
    let Predicate::And(left, _) = &where_clause.predicate else {
        panic!("expected both the collision and non-pushable residual");
    };
    assert!(matches!(
        left.as_ref(),
        Predicate::Comparison {
            left: Expression::PropertyAccess { property, .. },
            operator: ComparisonOp::Equals,
            right: Expression::Literal(Value::String(value)),
        } if property == "name" && value == "Bob"
    ));
}

#[test]
fn test_predicate_pushdown_into_a_scoped_optional_match_where() {
    // The predicate lives inside the OPTIONAL MATCH, so it pushes into that
    // clause's own pattern — and stays in the clause as the safety net every
    // non-pattern-matcher path relies on. It must never reappear as a
    // pipeline-level `Clause::Where`, which is what deleted null-extended
    // rows before scoping.
    let mut query = parse_cypher(
        "MATCH (p:Person) OPTIONAL MATCH (p)-[:KNOWS]->(f:Person) WHERE f.age > 35 RETURN p, f",
    )
    .unwrap();

    push_where_into_match(&mut query, &HashMap::new());

    assert!(
        !query.clauses.iter().any(|c| matches!(c, Clause::Where(_))),
        "the scoped predicate must not be lifted into a standalone WHERE"
    );
    let Clause::OptionalMatch(m) = &query.clauses[1] else {
        panic!("expected OPTIONAL MATCH clause");
    };
    let PatternElement::Node(node) = &m.patterns[0].elements[2] else {
        panic!("expected the optional target node");
    };
    assert!(matches!(
        node.properties.as_ref().and_then(|props| props.get("age")),
        Some(PropertyMatcher::GreaterThan(Value::Int64(35)))
    ));
    assert!(
        m.where_clause.is_some(),
        "fully-pushed predicate stays as the safety net"
    );
}

#[test]
fn test_scoped_optional_where_narrows_to_the_unpushable_remainder() {
    let mut query = parse_cypher(
        "MATCH (p:Person) OPTIONAL MATCH (p)-[:KNOWS]->(f:Person) \
         WHERE f.age > 35 AND size(f.name) > 0 RETURN p, f",
    )
    .unwrap();

    push_where_into_match(&mut query, &HashMap::new());

    let Clause::OptionalMatch(m) = &query.clauses[1] else {
        panic!("expected OPTIONAL MATCH clause");
    };
    let where_clause = m.where_clause.as_ref().expect("remainder survives");
    assert!(
        !matches!(where_clause.predicate, Predicate::And(_, _)),
        "the pushed comparison should be gone from the remainder"
    );
}

#[test]
fn test_scoped_optional_where_blocks_the_aggregate_fusion() {
    // The fused counter cannot evaluate a per-candidate predicate, so the
    // shape must stay on the materialized executor.
    let mut query = parse_cypher(
        "MATCH (p:Person) OPTIONAL MATCH (p)-[:KNOWS]->(f:Person) WHERE f.age > 35 \
         RETURN p.name AS n, count(f) AS k",
    )
    .unwrap();

    fuse_optional_match_aggregate(&mut query);

    assert!(
        !query
            .clauses
            .iter()
            .any(|c| matches!(c, Clause::FusedOptionalMatchAggregate { .. })),
        "a clause-owned WHERE must block the fused counter"
    );
}

#[test]
fn test_predicate_pushdown_keeps_second_same_direction_bound_in_where() {
    let mut query = parse_cypher(
        "MATCH (n:Person) \
         WHERE n.age > 35 AND n.age > 38 AND size(n.name) > 0 RETURN n",
    )
    .unwrap();

    push_where_into_match(&mut query, &HashMap::new());

    let Clause::Match(m) = &query.clauses[0] else {
        panic!("expected MATCH clause");
    };
    let PatternElement::Node(node) = &m.patterns[0].elements[0] else {
        panic!("expected node pattern");
    };
    assert!(matches!(
        node.properties.as_ref().and_then(|props| props.get("age")),
        Some(PropertyMatcher::GreaterThan(Value::Int64(35)))
    ));

    let Clause::Where(where_clause) = &query.clauses[1] else {
        panic!("expected residual WHERE clause");
    };
    assert!(matches!(where_clause.predicate, Predicate::And(_, _)));
}

#[test]
fn test_text_predicate_pushdown_literals_and_parameters() {
    let cases = [
        (
            "MATCH (n:Person) WHERE n.name STARTS WITH 'Ali' RETURN n",
            "starts",
        ),
        (
            "MATCH (n:Person) WHERE n.name CONTAINS 'lic' RETURN n",
            "contains",
        ),
        (
            "MATCH (n:Person) WHERE n.name ENDS WITH $suffix RETURN n",
            "ends",
        ),
    ];
    let mut params = HashMap::new();
    params.insert("suffix".to_string(), Value::String("ice".to_string()));

    for (cypher, expected) in cases {
        let mut query = parse_cypher(cypher).unwrap();
        push_where_into_match(&mut query, &params);

        let Clause::Match(m) = &query.clauses[0] else {
            panic!("expected MATCH clause");
        };
        let PatternElement::Node(node) = &m.patterns[0].elements[0] else {
            panic!("expected node pattern");
        };
        let matcher = node
            .properties
            .as_ref()
            .and_then(|properties| properties.get("name"));
        assert!(match expected {
            "starts" =>
                matches!(matcher, Some(PropertyMatcher::StartsWith(value)) if value == "Ali"),
            "contains" =>
                matches!(matcher, Some(PropertyMatcher::Contains(value)) if value == "lic"),
            "ends" => matches!(matcher, Some(PropertyMatcher::EndsWith(value)) if value == "ice"),
            _ => unreachable!(),
        });
        assert!(matches!(query.clauses[1], Clause::Where(_)));
    }
}

#[test]
fn test_text_predicate_pushdown_rejects_unsafe_shapes() {
    for cypher in [
        "MATCH (n:Person) WHERE n.name CONTAINS '' RETURN n",
        "MATCH (n:Person) WHERE n.name ENDS WITH $missing RETURN n",
        "MATCH (n:Person) WHERE NOT (n.name STARTS WITH 'Ali') RETURN n",
    ] {
        let mut query = parse_cypher(cypher).unwrap();
        push_where_into_match(&mut query, &HashMap::new());

        let Clause::Match(m) = &query.clauses[0] else {
            panic!("expected MATCH clause");
        };
        let PatternElement::Node(node) = &m.patterns[0].elements[0] else {
            panic!("expected node pattern");
        };
        assert!(node.properties.is_none());
    }
}

#[test]
fn test_relationship_text_and_parameter_pushdown() {
    let cases = [
        (
            "MATCH (a:A)-[r:R]->(b:B) WHERE r.tag STARTS WITH 'pre' RETURN b",
            PropOp::StartsWith,
            Value::String("pre".to_string()),
        ),
        (
            "MATCH (a:A)-[r:R]->(b:B) WHERE r.tag CONTAINS $needle RETURN b",
            PropOp::Contains,
            Value::String("mid".to_string()),
        ),
        (
            "MATCH (a:A)-[r:R]->(b:B) WHERE r.tag ENDS WITH 'end' RETURN b",
            PropOp::EndsWith,
            Value::String("end".to_string()),
        ),
        (
            "MATCH (a:A)-[r:R]->(b:B) WHERE r.score = $score RETURN b",
            PropOp::Eq,
            Value::Int64(7),
        ),
    ];
    let mut params = HashMap::new();
    params.insert("needle".to_string(), Value::String("mid".to_string()));
    params.insert("score".to_string(), Value::Int64(7));

    for (cypher, expected_op, expected_value) in cases {
        let mut query = parse_cypher(cypher).unwrap();
        optimize(&mut query, &DirGraph::new(), &params);

        let filter = query
            .clauses
            .iter()
            .filter_map(|clause| match clause {
                Clause::Match(m) => Some(m),
                _ => None,
            })
            .flat_map(|m| &m.patterns)
            .flat_map(|pattern| &pattern.elements)
            .find_map(|element| match element {
                PatternElement::Edge(edge) => edge.edge_filter.as_ref(),
                _ => None,
            })
            .expect("expected pushed relationship filter");
        assert!(matches!(
            &filter.predicate,
            RelEdgePredicate::Property { op, value, .. }
                if *op == expected_op && *value == expected_value
        ));
    }
}

#[test]
fn test_relationship_text_pushdown_rejects_missing_or_wrong_typed_params() {
    for (cypher, params) in [
        (
            "MATCH (a:A)-[r:R]->(b:B) WHERE r.tag CONTAINS $missing RETURN b",
            HashMap::new(),
        ),
        (
            "MATCH (a:A)-[r:R]->(b:B) WHERE r.tag ENDS WITH $suffix RETURN b",
            HashMap::from([("suffix".to_string(), Value::Int64(7))]),
        ),
    ] {
        let mut query = parse_cypher(cypher).unwrap();
        optimize(&mut query, &DirGraph::new(), &params);
        assert!(query.clauses.iter().all(|clause| match clause {
            Clause::Match(m) => m.patterns.iter().all(|pattern| {
                pattern.elements.iter().all(|element| match element {
                    PatternElement::Edge(edge) => edge.edge_filter.is_none(),
                    _ => true,
                })
            }),
            _ => true,
        }));
        assert!(query
            .clauses
            .iter()
            .any(|clause| matches!(clause, Clause::Where(_))));
    }
}

#[test]
fn test_comparison_pushdown() {
    let mut query = parse_cypher("MATCH (n:Person) WHERE n.age > 30 RETURN n").unwrap();

    let graph = DirGraph::new();
    let params = HashMap::new();
    optimize(&mut query, &graph, &params);

    // Comparison should be pushed into MATCH, WHERE kept as safety net
    assert_eq!(query.clauses.len(), 3); // MATCH + WHERE + RETURN

    if let Clause::Match(m) = &query.clauses[0] {
        if let PatternElement::Node(np) = &m.patterns[0].elements[0] {
            let props = np.properties.as_ref().unwrap();
            assert!(matches!(
                props.get("age"),
                Some(PropertyMatcher::GreaterThan(Value::Int64(30)))
            ));
        }
    }
}

#[test]
fn test_no_pushdown_for_not_equals() {
    let mut query = parse_cypher("MATCH (n:Person) WHERE n.age <> 30 RETURN n").unwrap();

    let graph = DirGraph::new();
    let params = HashMap::new();
    optimize(&mut query, &graph, &params);

    // NotEquals should NOT be pushed - WHERE should remain
    assert_eq!(query.clauses.len(), 3); // MATCH + WHERE + RETURN
}

#[test]
fn test_predicate_pushdown_parameter() {
    let mut query = parse_cypher("MATCH (n:Person) WHERE n.name = $name RETURN n").unwrap();

    let graph = DirGraph::new();
    let mut params = HashMap::new();
    params.insert("name".to_string(), Value::String("Alice".to_string()));
    optimize(&mut query, &graph, &params);

    // Parameter resolved and pushed; WHERE kept as safety net
    assert_eq!(query.clauses.len(), 3); // MATCH + WHERE + RETURN

    // The MATCH pattern should now have {name: 'Alice'} as a property
    if let Clause::Match(m) = &query.clauses[0] {
        if let PatternElement::Node(np) = &m.patterns[0].elements[0] {
            assert!(np.properties.is_some());
            let props = np.properties.as_ref().unwrap();
            assert!(props.contains_key("name"));
            assert!(matches!(
                props.get("name"),
                Some(PropertyMatcher::Equals(Value::String(s))) if s == "Alice"
            ));
        } else {
            panic!("Expected node pattern");
        }
    }
}

#[test]
fn test_predicate_pushdown_parameter_partial() {
    let mut query =
        parse_cypher("MATCH (n:Person) WHERE n.name = $name AND n.age > $min_age RETURN n")
            .unwrap();

    let graph = DirGraph::new();
    let mut params = HashMap::new();
    params.insert("name".to_string(), Value::String("Alice".to_string()));
    params.insert("min_age".to_string(), Value::Int64(25));
    optimize(&mut query, &graph, &params);

    // Both should be pushed: n.name = $name (equality) and n.age > $min_age (comparison)
    // WHERE kept as safety net
    assert_eq!(query.clauses.len(), 3); // MATCH + WHERE + RETURN

    if let Clause::Match(m) = &query.clauses[0] {
        if let PatternElement::Node(np) = &m.patterns[0].elements[0] {
            let props = np.properties.as_ref().unwrap();
            assert!(matches!(
                props.get("name"),
                Some(PropertyMatcher::Equals(Value::String(s))) if s == "Alice"
            ));
            assert!(matches!(
                props.get("age"),
                Some(PropertyMatcher::GreaterThan(Value::Int64(25)))
            ));
        }
    }
}

#[test]
fn test_comparison_range_merge() {
    let mut query =
        parse_cypher("MATCH (n:Paper) WHERE n.year >= 2015 AND n.year <= 2022 RETURN n").unwrap();

    let graph = DirGraph::new();
    let params = HashMap::new();
    optimize(&mut query, &graph, &params);

    // Both comparisons should be merged into a Range matcher; WHERE kept
    assert_eq!(query.clauses.len(), 3); // MATCH + WHERE + RETURN

    if let Clause::Match(m) = &query.clauses[0] {
        if let PatternElement::Node(np) = &m.patterns[0].elements[0] {
            let props = np.properties.as_ref().unwrap();
            assert!(matches!(
                props.get("year"),
                Some(PropertyMatcher::Range {
                    lower: Value::Int64(2015),
                    lower_inclusive: true,
                    upper: Value::Int64(2022),
                    upper_inclusive: true,
                })
            ));
        }
    }
}

#[test]
fn test_correlated_nodeprop_pushdown() {
    // The classic shape from sodir-prospect build:
    //   MATCH (a:A) MATCH (b:B) WHERE b.x = a.y
    // should push { x: EqualsNodeProp { var: "a", prop: "y" } } onto B.
    let mut query =
        parse_cypher("MATCH (a:A) MATCH (b:B) WHERE b.x = a.y RETURN a.id, b.id").unwrap();

    let graph = DirGraph::new();
    let params = HashMap::new();
    optimize(&mut query, &graph, &params);

    // Locate the second MATCH (matching on B)
    let b_match = query
        .clauses
        .iter()
        .filter_map(|c| match c {
            Clause::Match(m) => Some(m),
            _ => None,
        })
        .find(|m| {
            matches!(
                &m.patterns[0].elements[0],
                PatternElement::Node(np) if np.node_type.as_deref() == Some("B")
            )
        })
        .expect("expected second MATCH on B");

    if let PatternElement::Node(np) = &b_match.patterns[0].elements[0] {
        let props = np.properties.as_ref().expect("expected props on b");
        match props.get("x") {
            Some(PropertyMatcher::EqualsNodeProp { var, prop }) => {
                assert_eq!(var, "a");
                assert_eq!(prop, "y");
            }
            other => panic!("expected EqualsNodeProp on b.x, got {:?}", other),
        }
    }
}

#[test]
fn test_correlated_nodeprop_reversed_sides() {
    // Reversed: a.y = b.x (the cur_var b appears on the right).
    let mut query =
        parse_cypher("MATCH (a:A) MATCH (b:B) WHERE a.y = b.x RETURN a.id, b.id").unwrap();

    let graph = DirGraph::new();
    let params = HashMap::new();
    optimize(&mut query, &graph, &params);

    let b_match = query
        .clauses
        .iter()
        .filter_map(|c| match c {
            Clause::Match(m) => Some(m),
            _ => None,
        })
        .find(|m| {
            matches!(
                &m.patterns[0].elements[0],
                PatternElement::Node(np) if np.node_type.as_deref() == Some("B")
            )
        })
        .unwrap();

    if let PatternElement::Node(np) = &b_match.patterns[0].elements[0] {
        let props = np.properties.as_ref().unwrap();
        assert!(matches!(
            props.get("x"),
            Some(PropertyMatcher::EqualsNodeProp { var, prop })
                if var == "a" && prop == "y"
        ));
    }
}

#[test]
fn test_scalar_var_pushdown_from_unwind() {
    // WHERE s.title = fname where fname comes from UNWIND should become
    // an EqualsVar matcher on the pattern.
    let mut query =
        parse_cypher("UNWIND ['x','y'] AS fname MATCH (s:Strat) WHERE s.title = fname RETURN s.id")
            .unwrap();

    let graph = DirGraph::new();
    let params = HashMap::new();
    optimize(&mut query, &graph, &params);

    let s_match = query
        .clauses
        .iter()
        .filter_map(|c| match c {
            Clause::Match(m) => Some(m),
            _ => None,
        })
        .next()
        .unwrap();

    if let PatternElement::Node(np) = &s_match.patterns[0].elements[0] {
        let props = np.properties.as_ref().unwrap();
        assert!(matches!(
            props.get("title"),
            Some(PropertyMatcher::EqualsVar(n)) if n == "fname"
        ));
    }
}

#[test]
fn test_no_pushdown_when_both_vars_in_same_match() {
    // Within the same MATCH, a.y = b.x is handled by the pattern
    // executor's shared-var join. We must NOT rewrite it as an
    // EqualsNodeProp (which assumes prior-bound node).
    let mut query = parse_cypher("MATCH (a:A), (b:B) WHERE a.y = b.x RETURN a.id, b.id").unwrap();

    let graph = DirGraph::new();
    let params = HashMap::new();
    optimize(&mut query, &graph, &params);

    for clause in &query.clauses {
        if let Clause::Match(m) = clause {
            for pat in &m.patterns {
                for el in &pat.elements {
                    if let PatternElement::Node(np) = el {
                        if let Some(props) = &np.properties {
                            for m in props.values() {
                                assert!(
                                    !matches!(m, PropertyMatcher::EqualsNodeProp { .. }),
                                    "same-MATCH correlated equality must not be rewritten"
                                );
                            }
                        }
                    }
                }
            }
        }
    }
}

#[test]
fn test_undirected_pattern_reversed_by_selectivity() {
    // Regression: `(other)-[r]-(p {title: 'X'})` was not reversed because
    // the planner bailed out on undirected edges, leaving `other` (no
    // constraints, full graph scan) as the start node. The selective
    // anchor `p {title: 'X'}` must now be picked as start.
    let mut query =
        parse_cypher("MATCH (other)-[r]-(p {title: 'X'}) RETURN type(r), count(other)").unwrap();

    let graph = DirGraph::new();
    let params = HashMap::new();
    optimize(&mut query, &graph, &params);

    let m = query
        .clauses
        .iter()
        .find_map(|c| match c {
            Clause::Match(m) => Some(m),
            _ => None,
        })
        .unwrap();

    let first = match &m.patterns[0].elements[0] {
        PatternElement::Node(np) => np,
        _ => panic!("expected node"),
    };
    assert_eq!(
        first.variable.as_deref(),
        Some("p"),
        "selective anchor `p` should be the start after reversal"
    );
    assert!(
        first.properties.is_some(),
        "start node must carry the title property after reversal"
    );
}

#[test]
fn test_nonindexed_in_does_not_tie_id_anchor() {
    let mut query = parse_cypher(
        "MATCH (a:Broad)-[:R]->(b:Anchor {id: 7}) \
         WHERE a.code IN ['code_7'] RETURN a, b",
    )
    .unwrap();
    let mut graph = DirGraph::new();
    graph
        .type_indices
        .entry_or_default("Broad".to_string())
        .extend((0..100).map(petgraph::graph::NodeIndex::new));
    graph
        .type_indices
        .entry_or_default("Anchor".to_string())
        .extend((100..200).map(petgraph::graph::NodeIndex::new));

    optimize(&mut query, &graph, &HashMap::new());

    let match_clause = query
        .clauses
        .iter()
        .find_map(|clause| match clause {
            Clause::Match(m) => Some(m),
            _ => None,
        })
        .expect("expected MATCH clause");
    let PatternElement::Node(first) = &match_clause.patterns[0].elements[0] else {
        panic!("expected start node");
    };
    assert_eq!(first.variable.as_deref(), Some("b"));
}

#[test]
fn test_empty_in_parameter_pushes_known_empty_matcher() {
    let mut query = parse_cypher("MATCH (n:Item) WHERE n.code IN $codes RETURN n").unwrap();
    let params = HashMap::from([("codes".to_string(), Value::List(Vec::new()))]);

    push_where_into_match(&mut query, &params);

    let Clause::Match(match_clause) = &query.clauses[0] else {
        panic!("expected MATCH clause");
    };
    let PatternElement::Node(node) = &match_clause.patterns[0].elements[0] else {
        panic!("expected node pattern");
    };
    assert!(matches!(
        node.properties
            .as_ref()
            .and_then(|properties| properties.get("code")),
        Some(PropertyMatcher::In(values)) if values.is_empty()
    ));
}

#[test]
fn test_label_cardinality_includes_secondary_carriers() {
    let query = parse_cypher("MATCH (n:Item) RETURN n").unwrap();
    let Clause::Match(match_clause) = &query.clauses[0] else {
        panic!("expected MATCH clause");
    };
    let PatternElement::Node(node) = &match_clause.patterns[0].elements[0] else {
        panic!("expected node pattern");
    };
    let mut graph = DirGraph::new();
    graph
        .type_indices
        .entry_or_default("Item".to_string())
        .extend((0..3).map(petgraph::graph::NodeIndex::new));
    graph.secondary_label_index.insert(
        crate::graph::schema::InternedKey::from_str("Item"),
        vec![
            petgraph::graph::NodeIndex::new(3),
            petgraph::graph::NodeIndex::new(4),
        ],
    );
    graph.has_secondary_labels = true;

    assert_eq!(join_order::estimate_node_selectivity(node, &graph), 5);
}

#[test]
fn test_undirected_pattern_no_reverse_when_first_is_anchor() {
    // Reverse case: anchor is already first — no reversal expected.
    let mut query =
        parse_cypher("MATCH (p {title: 'X'})-[r]-(other) RETURN type(r), count(other)").unwrap();

    let graph = DirGraph::new();
    let params = HashMap::new();
    optimize(&mut query, &graph, &params);

    let m = query
        .clauses
        .iter()
        .find_map(|c| match c {
            Clause::Match(m) => Some(m),
            _ => None,
        })
        .unwrap();

    let first = match &m.patterns[0].elements[0] {
        PatternElement::Node(np) => np,
        _ => panic!("expected node"),
    };
    assert_eq!(first.variable.as_deref(), Some("p"));
}

#[test]
fn test_var_length_pattern_reversed_by_selectivity() {
    // `(other)-[*1..3]-(p {id: 1})` — the anchor is selective. Reversal
    // is safe when the pattern has no path assignment (the `path_assignments`
    // guard already protects path-bound patterns).
    let mut query = parse_cypher("MATCH (other)-[*1..3]-(p {id: 1}) RETURN p, other").unwrap();

    let graph = DirGraph::new();
    let params = HashMap::new();
    optimize(&mut query, &graph, &params);

    let m = query
        .clauses
        .iter()
        .find_map(|c| match c {
            Clause::Match(m) => Some(m),
            _ => None,
        })
        .unwrap();

    let first = match &m.patterns[0].elements[0] {
        PatternElement::Node(np) => np,
        _ => panic!("expected node"),
    };
    assert_eq!(
        first.variable.as_deref(),
        Some("p"),
        "var-length patterns should still get start-node optimization"
    );
}

#[test]
fn test_var_length_with_path_assignment_not_reversed() {
    // Path assignments must block reversal (path semantics depend on direction).
    let mut query = parse_cypher("MATCH path = (other)-[*1..3]-(p {id: 1}) RETURN path").unwrap();

    let graph = DirGraph::new();
    let params = HashMap::new();
    optimize(&mut query, &graph, &params);

    let m = query
        .clauses
        .iter()
        .find_map(|c| match c {
            Clause::Match(m) => Some(m),
            _ => None,
        })
        .unwrap();

    let first = match &m.patterns[0].elements[0] {
        PatternElement::Node(np) => np,
        _ => panic!("expected node"),
    };
    assert_eq!(
        first.variable.as_deref(),
        Some("other"),
        "path-bound patterns must not be reversed"
    );
}

#[test]
fn test_limit_pushdown_single_match_with_where() {
    // Single MATCH + WHERE + RETURN + LIMIT — pushdown is safe; the LIMIT
    // clause should be removed and the MATCH should carry limit_hint.
    let mut query =
        parse_cypher("MATCH (n:Person) WHERE n.age > 25 RETURN n.name LIMIT 10").unwrap();

    let graph = DirGraph::new();
    let params = HashMap::new();
    optimize(&mut query, &graph, &params);

    // Limit clause should be gone (absorbed into the MATCH hint)
    let has_limit = query.clauses.iter().any(|c| matches!(c, Clause::Limit(_)));
    assert!(
        !has_limit,
        "single-MATCH query should have LIMIT pushed into MATCH"
    );

    let m = query
        .clauses
        .iter()
        .find_map(|c| match c {
            Clause::Match(m) => Some(m),
            _ => None,
        })
        .expect("expected a MATCH clause");
    assert_eq!(m.limit_hint, Some(10));
}

#[test]
fn test_limit_pushdown_unfiltered_node_cartesian() {
    let mut query =
        parse_cypher("MATCH (a:Person), (b:Organization) RETURN a.name, b.name LIMIT 20").unwrap();

    let graph = DirGraph::new();
    let params = HashMap::new();
    optimize(&mut query, &graph, &params);

    assert!(
        !query.clauses.iter().any(|c| matches!(c, Clause::Limit(_))),
        "pure node cartesian should absorb LIMIT"
    );
    let match_clause = query
        .clauses
        .iter()
        .find_map(|clause| match clause {
            Clause::Match(m) => Some(m),
            _ => None,
        })
        .unwrap();
    assert_eq!(match_clause.limit_hint, Some(20));
}

#[test]
fn test_limit_pushdown_keeps_filtered_node_cartesian_conservative() {
    let mut query = parse_cypher(
        "MATCH (a:Person), (b:Organization) WHERE a.city = b.city \
         RETURN a.name, b.name LIMIT 20",
    )
    .unwrap();

    let graph = DirGraph::new();
    let params = HashMap::new();
    optimize(&mut query, &graph, &params);

    assert!(query.clauses.iter().any(|c| matches!(c, Clause::Limit(_))));
    let match_clause = query
        .clauses
        .iter()
        .find_map(|clause| match clause {
            Clause::Match(m) => Some(m),
            _ => None,
        })
        .unwrap();
    assert_eq!(match_clause.limit_hint, None);
}

#[test]
fn test_multi_match_no_reverse_when_bound_var_first() {
    // Regression for the user's "(p)-[:P31]->(:human) gets reversed to scan
    // 13M humans" report: in the second MATCH, `p` is already bound by the
    // first MATCH. The planner must not reverse the pattern to start from
    // `(:human)` (a 13M-row scan) just because `p` looks unconstrained
    // statically.
    let mut query =
        parse_cypher("MATCH (p:Person) MATCH (p)-[:KNOWS]->(c:Company) RETURN p, c").unwrap();

    let graph = DirGraph::new();
    let params = HashMap::new();
    optimize(&mut query, &graph, &params);

    // The second MATCH should still start with `p`, not `c`.
    let matches: Vec<_> = query
        .clauses
        .iter()
        .filter_map(|c| match c {
            Clause::Match(m) => Some(m),
            _ => None,
        })
        .collect();
    assert!(matches.len() >= 2, "expected two MATCH clauses");
    let second = matches[1];
    let first_var = match &second.patterns[0].elements[0] {
        PatternElement::Node(np) => np.variable.as_deref(),
        _ => None,
    };
    assert_eq!(
        first_var,
        Some("p"),
        "second MATCH must keep pre-bound `p` as start node, not reverse to `c`"
    );
}

#[test]
fn test_multi_match_reorder_prefers_anchored_pattern() {
    // When a single MATCH has two patterns sharing a pre-bound variable,
    // the more selective pattern should run first. The planner already
    // does intra-clause reordering — this test pins the cross-clause
    // bound-vars logic so it doesn't regress.
    let mut query = parse_cypher(
        "MATCH (p {id: 1}) \
         MATCH (p)-[:R1]->(:T1), (p)-[:R2]->({id: 99}) \
         RETURN p",
    )
    .unwrap();

    let graph = DirGraph::new();
    let params = HashMap::new();
    optimize(&mut query, &graph, &params);

    // Just ensure optimization doesn't crash and patterns survive.
    let m2 = query
        .clauses
        .iter()
        .filter_map(|c| match c {
            Clause::Match(m) => Some(m),
            _ => None,
        })
        .nth(1)
        .expect("expected second MATCH");
    assert_eq!(m2.patterns.len(), 2);
}

#[test]
fn test_limit_pushdown_multi_match_safety() {
    // Regression: 3-MATCH + WHERE on last-bound variable + LIMIT N produced
    // fewer rows than expected because push_limit_into_match was setting
    // `limit_hint` on the last MATCH, which interacts incorrectly with the
    // per-row pattern executor in the subsequent-MATCH path. The planner
    // must NOT push LIMIT into MATCH when there are multiple MATCH clauses.
    let mut query = parse_cypher(
        "MATCH (a)-[:R1]->(:T1) \
         MATCH (a)-[:R2]->(b) \
         MATCH (b)-[:R3]->(c) \
         WHERE c.id = 7318 \
         RETURN a.id, b.id LIMIT 50",
    )
    .unwrap();

    let graph = DirGraph::new();
    let params = HashMap::new();
    optimize(&mut query, &graph, &params);

    // Limit clause must remain — pushdown is unsafe for multi-MATCH
    let has_limit = query.clauses.iter().any(|c| matches!(c, Clause::Limit(_)));
    assert!(has_limit, "multi-MATCH query must retain its LIMIT clause");

    // No MATCH clause should have a limit_hint set
    for clause in &query.clauses {
        if let Clause::Match(m) = clause {
            assert_eq!(
                m.limit_hint, None,
                "multi-MATCH clauses must not receive a limit_hint"
            );
        }
    }
}

#[test]
fn test_reorder_match_clauses_picks_rare_edge_first() {
    // Two MATCH clauses share `p`, both id-anchored on the other end. The
    // planner should drive the smaller-edge-type clause first so the
    // executor enumerates fewer rows before joining.
    //
    // Motivating real-world case (Wikidata):
    //   MATCH (p)-[:P31]->({id:5})    -- 80M instance-of edges
    //   MATCH (p)-[:P27]->({id:183})  -- 3M citizenship edges
    // → swap so P27 drives, then per-row check P31. ~25× cheaper.
    let mut query = parse_cypher(
        "MATCH (p)-[:VERY_COMMON]->({id: 1}) \
         MATCH (p)-[:RARE]->({id: 2}) \
         RETURN p",
    )
    .unwrap();

    let graph = DirGraph::new();
    {
        let mut cache = graph.edge_type_counts_cache.write().unwrap();
        let mut counts = HashMap::new();
        counts.insert("VERY_COMMON".to_string(), 1_000_000);
        counts.insert("RARE".to_string(), 1_000);
        *cache = Some(std::sync::Arc::new(counts));
    }

    let params = HashMap::new();
    optimize(&mut query, &graph, &params);

    let matches: Vec<_> = query
        .clauses
        .iter()
        .filter_map(|c| match c {
            Clause::Match(m) => Some(m),
            _ => None,
        })
        .collect();
    assert_eq!(matches.len(), 2, "expected two MATCH clauses preserved");

    // First MATCH after reorder must be the RARE one.
    let first_edge_type = matches[0].patterns[0].elements.iter().find_map(|e| {
        if let PatternElement::Edge(ep) = e {
            ep.connection_type.clone()
        } else {
            None
        }
    });
    assert_eq!(
        first_edge_type.as_deref(),
        Some("RARE"),
        "RARE (lower edge-type cost) should be promoted to first MATCH; \
         got first edge type = {first_edge_type:?}"
    );
}

#[test]
fn test_reorder_match_clauses_promotes_later_id_anchor_without_cache() {
    let mut query = parse_cypher(
        "MATCH (h:Hub)-[:WIDE]->(leaf:Leaf) \
         MATCH (h)-[:ANCHORED]->(anchor:Anchor {id: 7}) \
         RETURN h, leaf",
    )
    .unwrap();
    let graph = DirGraph::new();

    optimize(&mut query, &graph, &HashMap::new());

    let edge_types: Vec<_> = query
        .clauses
        .iter()
        .filter_map(|clause| match clause {
            Clause::Match(m) => m.patterns[0]
                .elements
                .iter()
                .find_map(|element| match element {
                    PatternElement::Edge(edge) => edge.connection_type.as_deref(),
                    _ => None,
                }),
            _ => None,
        })
        .collect();
    assert_eq!(edge_types, ["ANCHORED", "WIDE"]);
    assert!(!graph.has_edge_type_counts_cache());
}

#[test]
fn test_reorder_match_clauses_anchor_partition_is_stable() {
    let mut query = parse_cypher(
        "MATCH (h)-[:WIDE]->(leaf) \
         MATCH (h)-[:FIRST_ANCHOR]->({id: 1}) \
         MATCH (h)-[:SECOND_ANCHOR]->({id: 2}) \
         RETURN h",
    )
    .unwrap();

    optimize(&mut query, &DirGraph::new(), &HashMap::new());

    let edge_types: Vec<_> = query
        .clauses
        .iter()
        .filter_map(|clause| match clause {
            Clause::Match(m) => m.patterns[0]
                .elements
                .iter()
                .find_map(|element| match element {
                    PatternElement::Edge(edge) => edge.connection_type.as_deref(),
                    _ => None,
                }),
            _ => None,
        })
        .collect();
    assert_eq!(edge_types, ["FIRST_ANCHOR", "SECOND_ANCHOR", "WIDE"]);
}

#[test]
fn test_reorder_match_clauses_does_not_move_independent_anchor() {
    let mut query = parse_cypher(
        "MATCH (h)-[:WIDE]->(leaf) \
         MATCH (other)-[:ANCHORED]->({id: 1}) \
         RETURN h, other",
    )
    .unwrap();

    optimize(&mut query, &DirGraph::new(), &HashMap::new());

    let first_edge_type = query.clauses.iter().find_map(|clause| match clause {
        Clause::Match(m) => m.patterns[0]
            .elements
            .iter()
            .find_map(|element| match element {
                PatternElement::Edge(edge) => edge.connection_type.as_deref(),
                _ => None,
            }),
        _ => None,
    });
    assert_eq!(first_edge_type, Some("WIDE"));
}

#[test]
fn test_reorder_match_clauses_skips_when_cache_missing() {
    // Same query shape as above, but no edge_type_counts_cache populated.
    // The reorder pass must bail (avoid triggering an O(E) cache build
    // at plan time) and leave the original clause order intact.
    let mut query = parse_cypher(
        "MATCH (p)-[:VERY_COMMON]->({id: 1}) \
         MATCH (p)-[:RARE]->({id: 2}) \
         RETURN p",
    )
    .unwrap();

    let graph = DirGraph::new();
    // Confirm cache is unset to start.
    assert!(!graph.has_edge_type_counts_cache());

    let params = HashMap::new();
    optimize(&mut query, &graph, &params);

    // Original textual order preserved: VERY_COMMON first.
    let matches: Vec<_> = query
        .clauses
        .iter()
        .filter_map(|c| match c {
            Clause::Match(m) => Some(m),
            _ => None,
        })
        .collect();
    assert_eq!(matches.len(), 2);
    let first_edge_type = matches[0].patterns[0].elements.iter().find_map(|e| {
        if let PatternElement::Edge(ep) = e {
            ep.connection_type.clone()
        } else {
            None
        }
    });
    assert_eq!(first_edge_type.as_deref(), Some("VERY_COMMON"));
    // Cache must still be empty — the planner did not force a build.
    assert!(
        !graph.has_edge_type_counts_cache(),
        "planner must not warm the edge-type-counts cache from the optimization path"
    );
}

#[test]
fn test_reorder_match_clauses_requires_id_anchor() {
    // No id anchor on either MATCH → cannot use the edge-count proxy
    // safely (other-end fan-in dominates and isn't captured). Pass must
    // leave order intact even if the cache is populated.
    let mut query = parse_cypher(
        "MATCH (p)-[:VERY_COMMON]->(q) \
         MATCH (p)-[:RARE]->(r) \
         RETURN p",
    )
    .unwrap();

    let graph = DirGraph::new();
    {
        let mut cache = graph.edge_type_counts_cache.write().unwrap();
        let mut counts = HashMap::new();
        counts.insert("VERY_COMMON".to_string(), 1_000_000);
        counts.insert("RARE".to_string(), 1_000);
        *cache = Some(std::sync::Arc::new(counts));
    }

    let params = HashMap::new();
    optimize(&mut query, &graph, &params);

    let matches: Vec<_> = query
        .clauses
        .iter()
        .filter_map(|c| match c {
            Clause::Match(m) => Some(m),
            _ => None,
        })
        .collect();
    let first_edge_type = matches[0].patterns[0].elements.iter().find_map(|e| {
        if let PatternElement::Edge(ep) = e {
            ep.connection_type.clone()
        } else {
            None
        }
    });
    assert_eq!(
        first_edge_type.as_deref(),
        Some("VERY_COMMON"),
        "without id-anchored endpoints the proxy is unreliable; do not reorder"
    );
}

#[test]
fn test_fuse_match_return_aggregate_count_distinct() {
    // Single-MATCH + RETURN with count(DISTINCT v) on the OTHER node variable
    // — the planner should now fuse this and set distinct_count=true.
    let mut query = parse_cypher(
        "MATCH (a:Person)-[:KNOWS]->(b:Person) \
         RETURN a, count(DISTINCT b) AS friends \
         ORDER BY friends DESC LIMIT 10",
    )
    .unwrap();

    let graph = DirGraph::new();
    let params = HashMap::new();
    optimize(&mut query, &graph, &params);

    let mut found = false;
    for clause in &query.clauses {
        if let Clause::FusedMatchReturnAggregate {
            distinct_count,
            top_k,
            ..
        } = clause
        {
            assert!(*distinct_count, "distinct_count flag must be set");
            assert!(
                top_k.is_some(),
                "ORDER BY count DESC LIMIT 10 must absorb into top_k"
            );
            found = true;
        }
    }
    assert!(
        found,
        "FusedMatchReturnAggregate must fire for count(DISTINCT) shape"
    );
}

#[test]
fn test_fuse_untyped_global_edge_count() {
    let mut query = parse_cypher("MATCH ()-[r]->() RETURN count(r) AS n").unwrap();
    let graph = DirGraph::new();
    let params = HashMap::new();
    optimize(&mut query, &graph, &params);
    assert!(matches!(
        query.clauses.as_slice(),
        [Clause::FusedCountAllEdges { alias }] if alias == "n"
    ));
}

#[test]
fn test_untyped_global_edge_count_rejects_constrained_shapes() {
    let queries = [
        "MATCH ()-[r]-() RETURN count(r) AS n",
        "MATCH (a)-[r]->(a) RETURN count(r) AS n",
        "MATCH (:Person)-[r]->() RETURN count(r) AS n",
        "MATCH ()-[r:R|S]->() RETURN count(r) AS n",
        "MATCH p = ()-[r]->() RETURN count(r) AS n",
    ];
    let graph = DirGraph::new();
    let params = HashMap::new();
    for source in queries {
        let mut query = parse_cypher(source).unwrap();
        optimize(&mut query, &graph, &params);
        assert!(
            !query
                .clauses
                .iter()
                .any(|clause| matches!(clause, Clause::FusedCountAllEdges { .. })),
            "constrained edge count must not use global count: {source}"
        );
    }
}

#[test]
fn test_fuse_match_return_aggregate_property_group_topk() {
    let mut query = parse_cypher(
        "MATCH (a:Person)-[:KNOWS]->(b:Person) \
         RETURN a.city AS city, count(b) AS n \
         ORDER BY n DESC LIMIT 10",
    )
    .unwrap();

    let graph = DirGraph::new();
    let params = HashMap::new();
    optimize(&mut query, &graph, &params);

    assert!(
        matches!(
            query.clauses.as_slice(),
            [Clause::FusedMatchReturnAggregate {
                top_k: Some((_, true, 10)),
                distinct_count: false,
                ..
            }]
        ),
        "direct property grouping must merge values inside fused top-k: {:#?}",
        query.clauses
    );
}

#[test]
fn test_property_grouped_distinct_count_is_not_fused() {
    let mut query = parse_cypher(
        "MATCH (a:Person)-[:KNOWS]->(b:Person) \
         RETURN a.city AS city, count(DISTINCT b) AS n",
    )
    .unwrap();

    let graph = DirGraph::new();
    let params = HashMap::new();
    optimize(&mut query, &graph, &params);

    assert!(
        !query
            .clauses
            .iter()
            .any(|clause| matches!(clause, Clause::FusedMatchReturnAggregate { .. })),
        "distinct peer sets cannot be summed after nodes collapse by property value"
    );
}

#[test]
fn test_fuse_match_return_aggregate_global_two_hop_count() {
    let mut query = parse_cypher(
        "MATCH (a:Person)-[:KNOWS]->(b:Person)-[:KNOWS]->(c:Person) \
         RETURN count(*) AS paths",
    )
    .unwrap();

    let graph = DirGraph::new();
    let params = HashMap::new();
    optimize(&mut query, &graph, &params);

    assert!(
        matches!(
            query.clauses.as_slice(),
            [Clause::FusedMatchReturnAggregate {
                distinct_count: false,
                ..
            }]
        ),
        "pure two-hop count must use row-free aggregate fusion: {:#?}",
        query.clauses
    );
}

#[test]
fn test_global_two_hop_count_with_repeated_variable_is_not_fused() {
    let mut query = parse_cypher(
        "MATCH (a:Person)-[:KNOWS]->(b:Person)-[:KNOWS]->(a) \
         RETURN count(*) AS cycles",
    )
    .unwrap();

    let graph = DirGraph::new();
    let params = HashMap::new();
    optimize(&mut query, &graph, &params);

    assert!(
        !query
            .clauses
            .iter()
            .any(|clause| matches!(clause, Clause::FusedMatchReturnAggregate { .. })),
        "repeated node variables require the binding-aware matcher"
    );
}

#[test]
fn test_fuse_match_with_aggregate_count_distinct() {
    // Single-MATCH + WITH with count(DISTINCT v) — pipeline form.
    let mut query = parse_cypher(
        "MATCH (a:Person)-[:KNOWS]->(b:Person) \
         WITH a, count(DISTINCT b) AS friends \
         RETURN a, friends",
    )
    .unwrap();

    let graph = DirGraph::new();
    let params = HashMap::new();
    optimize(&mut query, &graph, &params);

    let mut found = false;
    for clause in &query.clauses {
        if let Clause::FusedMatchWithAggregate { distinct_count, .. } = clause {
            assert!(*distinct_count, "WITH-form distinct_count flag must be set");
            found = true;
        }
    }
    assert!(
        found,
        "FusedMatchWithAggregate must fire for WITH-count-DISTINCT shape"
    );
}

#[test]
fn test_count_distinct_unconstrained_group_not_fused() {
    // Unconstrained group node — fusing would force a 124M-node enumeration
    // on a Wikidata-scale graph. The materializing path is faster in that
    // regime, so the planner must NOT fuse.
    let mut query = parse_cypher(
        "MATCH (a)-[:R]->(b) \
         RETURN b, count(DISTINCT a) AS n \
         ORDER BY n DESC LIMIT 10",
    )
    .unwrap();

    let graph = DirGraph::new();
    let params = HashMap::new();
    optimize(&mut query, &graph, &params);

    let fused_with_distinct = query.clauses.iter().any(|c| {
        matches!(
            c,
            Clause::FusedMatchReturnAggregate {
                distinct_count: true,
                ..
            }
        )
    });
    assert!(
        !fused_with_distinct,
        "untyped group node must skip distinct-count fusion"
    );
}

#[test]
fn test_count_distinct_5_element_pattern_not_fused() {
    // 5-element patterns aren't supported in the distinct-count path yet.
    // The planner should leave the query unfused so the materializing
    // executor produces semantically-correct results.
    let mut query = parse_cypher(
        "MATCH (a:A)-[:R1]->(b)<-[:R2]-(c) \
         RETURN a, count(DISTINCT c) AS n \
         ORDER BY n DESC LIMIT 10",
    )
    .unwrap();

    let graph = DirGraph::new();
    let params = HashMap::new();
    optimize(&mut query, &graph, &params);

    let fused_with_distinct = query.clauses.iter().any(|c| {
        matches!(
            c,
            Clause::FusedMatchReturnAggregate {
                distinct_count: true,
                ..
            }
        )
    });
    assert!(
        !fused_with_distinct,
        "5-element distinct-count pattern must not be fused"
    );
}

#[test]
fn test_fold_pass_through_with_between_matches() {
    // The cohort-top-K shape: `Match WITH p Match Return ...`. The
    // pass-through WITH should be stripped so downstream Match-Match
    // fusion can fire.
    let mut query = parse_cypher(
        "MATCH (p)-[:T1]->({id: 1}) \
         WITH p \
         MATCH (p)-[r]->() \
         RETURN p.title, count(r) AS d \
         ORDER BY d DESC LIMIT 10",
    )
    .unwrap();

    let graph = DirGraph::new();
    let params = HashMap::new();
    optimize(&mut query, &graph, &params);

    // The pass-through WITH is dropped. The multi-MATCH desugar produces one
    // aggregate WITH, which must remain eager because p.title groups by value,
    // not by the p node's identity.
    let bare_with_count = query
        .clauses
        .iter()
        .filter(|c| matches!(c, Clause::With(_)))
        .count();
    assert_eq!(
        bare_with_count, 1,
        "pass-through WITH must be stripped while the value-grouping \
         aggregate WITH remains eager; got query: {:#?}",
        query.clauses
    );

    // The node-keyed streaming aggregate must not accept p.title grouping.
    let has_fused_aggregate = query
        .clauses
        .iter()
        .any(|c| matches!(c, Clause::FusedMatchWithAggregate { .. }));
    assert!(
        !has_fused_aggregate,
        "property-valued grouping must not land on the node-keyed \
         aggregate path; clauses: {:#?}",
        query.clauses
    );
}

#[test]
fn test_fold_pass_through_with_keeps_useful_with() {
    // A non-pass-through WITH (here aliasing or referencing extra
    // variables that subsequent clauses need) must NOT be folded.
    let mut query = parse_cypher("MATCH (p)-[r]->(q) WITH p, r RETURN p, r LIMIT 10").unwrap();

    let graph = DirGraph::new();
    let params = HashMap::new();
    optimize(&mut query, &graph, &params);

    // The pass-through WITH `WITH p, r` covers exactly what RETURN
    // references, so it CAN be safely folded in this case. To ensure
    // the converse test, use a different shape where the WITH
    // *renames* a variable.
    let mut renaming =
        parse_cypher("MATCH (p)-[r]->(q) WITH p AS person RETURN person LIMIT 10").unwrap();
    optimize(&mut renaming, &graph, &params);
    let has_with = renaming
        .clauses
        .iter()
        .any(|c| matches!(c, Clause::With(_)));
    assert!(
        has_with,
        "renaming WITH (`p AS person`) must not be folded — it changes scope"
    );
}

#[test]
fn test_fold_pass_through_with_skipped_when_orderby_follows() {
    // ORDER BY immediately after a WITH binds to the WITH's row scope.
    // Folding the WITH would move the ORDER BY's binding context.
    let mut query =
        parse_cypher("MATCH (p)-[:T]->({id: 1}) WITH p ORDER BY p.title LIMIT 10 RETURN p")
            .unwrap();

    let graph = DirGraph::new();
    let params = HashMap::new();
    optimize(&mut query, &graph, &params);

    // The WITH is preserved because ORDER BY follows it.
    let has_with = query.clauses.iter().any(|c| matches!(c, Clause::With(_)));
    assert!(
        has_with,
        "WITH followed by ORDER BY/SKIP/LIMIT must not be folded; \
         clauses: {:#?}",
        query.clauses
    );
}

#[test]
fn test_desugar_multi_match_return_aggregate() {
    // The Variant-B shape (Match-Match-Return-aggregate, no WITH).
    // Should be rewritten into Match-Match-With(group, agg)-Return so
    // the existing aggregate fusion fires.
    let mut query = parse_cypher(
        "MATCH (p)-[:T1]->({id: 1}) \
         MATCH (p)-[r]->() \
         RETURN p.title, count(r) AS d \
         ORDER BY d DESC LIMIT 10",
    )
    .unwrap();

    let graph = DirGraph::new();
    let params = HashMap::new();
    optimize(&mut query, &graph, &params);

    // The desugar still fires, but its synthesized value-grouping WITH stays
    // eager rather than entering the NodeIndex-keyed aggregate fusion.
    let landed_on_fused_aggregate = query
        .clauses
        .iter()
        .any(|c| matches!(c, Clause::FusedMatchWithAggregate { .. }));
    assert!(
        !landed_on_fused_aggregate,
        "property-valued Match-Match aggregation must not enter the \
         node-keyed streaming path; clauses: {:#?}",
        query.clauses
    );
    assert!(
        query.clauses.iter().any(|c| matches!(c, Clause::With(_))),
        "the safe eager aggregate WITH must remain after desugaring"
    );
}

#[test]
fn test_topk_bails_for_property_value_grouping() {
    // Property expressions group by VALUE, not NodeIndex. The fused
    // accumulator is node-keyed, so it must bail until it can re-bucket
    // equal property values without changing counts.
    let mut query = parse_cypher(
        "MATCH (p)-[:T1]->({id: 1}) \
         MATCH (p)-[r]-(other) \
         WHERE NOT (type(r) = 'T2' AND startNode(r) = other) \
         RETURN p.title AS name, p.description AS desc, count(r) AS d \
         ORDER BY d DESC LIMIT 10",
    )
    .unwrap();

    let graph = DirGraph::new();
    let params = HashMap::new();
    optimize(&mut query, &graph, &params);

    let topk_absorbed = query
        .clauses
        .iter()
        .any(|c| matches!(c, Clause::FusedMatchWithAggregate { top_k: Some(_), .. }));
    assert!(
        !topk_absorbed,
        "property-value grouping must not use node-keyed fusion"
    );
}

#[test]
fn test_topk_bails_for_property_grouping_after_pass_through_with() {
    // Folding the pass-through WITH must not accidentally admit the same
    // unsafe property-value grouping shape downstream.
    let mut query = parse_cypher(
        "MATCH (p)-[:P27]->({id: 20}) \
         WITH p \
         MATCH (p)-[r]-(other) \
         WHERE NOT (type(r) = 'P50' AND startNode(r) = other) \
         RETURN p.title AS name, p.description AS desc, count(r) AS connections \
         ORDER BY connections DESC LIMIT 10",
    )
    .unwrap();

    let graph = DirGraph::new();
    let params = HashMap::new();
    optimize(&mut query, &graph, &params);

    let topk_absorbed = query
        .clauses
        .iter()
        .any(|c| matches!(c, Clause::FusedMatchWithAggregate { top_k: Some(_), .. }));
    assert!(
        !topk_absorbed,
        "property-value grouping must stay unfused after WITH folding"
    );
}

#[test]
fn test_topk_skipped_for_computed_return_expressions() {
    // Computed RETURN expressions (arithmetic on aggregates here) are
    // not safe to absorb — we'd need *all* rows to know which 10 win.
    // Pin the bail-out so a future relaxation has to opt in.
    let mut query = parse_cypher(
        "MATCH (p)-[:T1]->({id: 1}) \
         MATCH (p)-[r]-() \
         WITH p, count(r) AS total, 1 AS one \
         RETURN p.title, total + one AS adjusted \
         ORDER BY total DESC LIMIT 10",
    )
    .unwrap();

    let graph = DirGraph::new();
    let params = HashMap::new();
    optimize(&mut query, &graph, &params);

    let topk_absorbed = query
        .clauses
        .iter()
        .any(|c| matches!(c, Clause::FusedMatchWithAggregate { top_k: Some(_), .. }));
    assert!(
        !topk_absorbed,
        "computed RETURN expressions must not absorb top_k; \
         clauses: {:#?}",
        query.clauses
    );
}

#[test]
fn test_desugar_skips_when_no_aggregate() {
    // No aggregate in RETURN → desugar must not fire (would just add
    // an unnecessary WITH).
    let mut query = parse_cypher(
        "MATCH (p)-[:T1]->({id: 1}) \
         MATCH (p)-[:T2]->({id: 2}) \
         RETURN p.title \
         LIMIT 10",
    )
    .unwrap();

    let graph = DirGraph::new();
    let params = HashMap::new();
    optimize(&mut query, &graph, &params);

    let bare_with_count = query
        .clauses
        .iter()
        .filter(|c| matches!(c, Clause::With(_)))
        .count();
    assert_eq!(
        bare_with_count, 0,
        "desugar must not introduce a WITH when RETURN has no aggregate"
    );
}

#[test]
fn test_desugar_skips_when_multiple_group_vars() {
    // Two distinct group variables (`p.title`, `q.name`) — the simple
    // single-variable rewrite doesn't apply; bail.
    let mut query = parse_cypher(
        "MATCH (p)-[:T1]->({id: 1}) \
         MATCH (p)-[r]->(q) \
         RETURN p.title, q.title, count(r) AS d \
         ORDER BY d DESC LIMIT 10",
    )
    .unwrap();

    let graph = DirGraph::new();
    let params = HashMap::new();
    optimize(&mut query, &graph, &params);

    // Should NOT have produced a FusedMatchWithAggregate — the desugar
    // was correctly skipped, leaving the query for the slow path.
    let fused_count = query
        .clauses
        .iter()
        .filter(|c| matches!(c, Clause::FusedMatchWithAggregate { .. }))
        .count();
    assert_eq!(
        fused_count, 0,
        "multi-group-variable RETURN must not be auto-rewritten"
    );
}

#[test]
fn test_fuse_optional_match_aggregate_single_pattern_fires() {
    let mut query = parse_cypher(
        "MATCH (n:Person) \
         OPTIONAL MATCH (n)-[r:KNOWS]->(m) \
         WITH n, count(*) AS c \
         RETURN n.title, c",
    )
    .unwrap();

    let graph = DirGraph::new();
    let params = HashMap::new();
    optimize(&mut query, &graph, &params);

    let fused_count = query
        .clauses
        .iter()
        .filter(|c| matches!(c, Clause::FusedOptionalMatchAggregate { .. }))
        .count();
    assert_eq!(
        fused_count, 1,
        "single-pattern OPTIONAL MATCH + WITH count(*) must fuse"
    );
}

#[test]
fn test_fuse_optional_match_aggregate_bails_on_multi_pattern() {
    // The fused executor computes ONE per-row match_count by summing
    // pattern counts — wrong for comma-separated multi-pattern OPTIONAL
    // MATCH (row count is the patterns' join; per-variable counts differ
    // per pattern). The gate must leave this to the materialized path.
    let mut query = parse_cypher(
        "MATCH (n:Person) \
         OPTIONAL MATCH (n)-[:KNOWS]->(a), (n)-[:WORKS_AT]->(b) \
         WITH n, count(a) AS ca, count(b) AS cb \
         RETURN n.title, ca, cb",
    )
    .unwrap();

    let graph = DirGraph::new();
    let params = HashMap::new();
    optimize(&mut query, &graph, &params);

    let fused_count = query
        .clauses
        .iter()
        .filter(|c| matches!(c, Clause::FusedOptionalMatchAggregate { .. }))
        .count();
    assert_eq!(
        fused_count, 0,
        "multi-pattern OPTIONAL MATCH must not fuse into FusedOptionalMatchAggregate"
    );
}

/// The lazy-eligibility contract, pinned as a corpus.
///
/// `mark_lazy_eligibility` decides whether a result is returned deferred, and a
/// deferred result holds an `Arc<DirGraph>` for its whole life — which makes the
/// next write through the owning graph copy-on-write the entire graph. So this
/// gate is not merely a projection optimisation: it decides who pays an O(V+E)
/// fork. Pinning the exact shape keeps that reach visible and honest.
///
/// The surprising member is `WHERE`: `optimize` keeps a standalone
/// `Clause::Where` as a safety net even after pushing every predicate into the
/// MATCH (see `test_predicate_pushdown_simple`), and the eligibility walk has no
/// arm for it. So the same point lookup is eligible written with an inline map
/// and ineligible written with `WHERE`.
#[test]
fn lazy_eligibility_corpus() {
    fn is_lazy(q: &str) -> bool {
        let mut query = parse_cypher(q).unwrap();
        let graph = DirGraph::new();
        let params = HashMap::new();
        optimize(&mut query, &graph, &params);
        mark_lazy_eligibility(&mut query);
        query.clauses.iter().any(|c| match c {
            Clause::Return(r) => r.lazy_eligible,
            _ => false,
        })
    }

    // Eligible: bare property projections over an unfiltered or inline-filtered
    // MATCH, with nothing but SKIP/LIMIT after the RETURN.
    for q in [
        "MATCH (u:User) RETURN u.name",
        "MATCH (u:User {id: 1}) RETURN u.name, u.email",
        "MATCH (u:User {id: 1}) RETURN u.name AS name",
        "MATCH (u:User) RETURN u.name LIMIT 10",
        "MATCH (u:User)-[:OWNS]->(t:Task) RETURN u.name, t.title",
        "OPTIONAL MATCH (u:User) RETURN u.name",
    ] {
        assert!(is_lazy(q), "expected lazy-eligible: {q}");
    }

    // Ineligible. Each of these takes the eager path and therefore never pins
    // the graph, whatever its size.
    for q in [
        // A standalone WHERE survives optimisation and disqualifies.
        "MATCH (u:User) WHERE u.id = 1 RETURN u.name",
        // Whole-node returns resolve via NodeRef, not the lazy resolver.
        "MATCH (u:User) RETURN u",
        // Any non-PropertyAccess return item.
        "MATCH (u:User) RETURN u.age + 1",
        "MATCH (u:User) RETURN count(u)",
        // Ordering, dedup and multi-stage pipelines all disqualify.
        "MATCH (u:User) RETURN u.name ORDER BY u.name",
        "MATCH (u:User) RETURN DISTINCT u.name",
        "MATCH (u:User) WITH u.name AS n RETURN n",
        "UNWIND [1, 2] AS x RETURN x",
    ] {
        assert!(!is_lazy(q), "expected NOT lazy-eligible: {q}");
    }

    // The same lookup, two spellings, opposite classifications. Asserted as an
    // explicit pair because it is the least defensible part of the rule: the
    // two queries are semantically identical and a user has no way to know
    // which one they wrote. Whichever way the rule moves, these two should
    // arrive at the same answer — if a future change makes them agree, delete
    // this pair rather than "fixing" it.
    assert!(is_lazy("MATCH (u:User {id: 1}) RETURN u.name"));
    assert!(!is_lazy("MATCH (u:User) WHERE u.id = 1 RETURN u.name"));

    // The exact shapes the graph-pin benchmark relies on. Pinned here because
    // measuring the pin with an ineligible read reports no change and looks
    // like a working fix doing nothing.
    assert!(is_lazy(
        "MATCH (p:Person {id: 0}) RETURN p.name AS name, p.age AS age"
    ));
    assert!(!is_lazy(
        "MATCH (p:Person) WHERE p.id = 0 RETURN p.name AS name, p.age AS age"
    ));
    assert!(is_lazy(
        "MATCH (p:Person) RETURN p.name AS name, p.age AS age"
    ));
}

// ============================================================================
// text_score() — raw query vectors
// ============================================================================
//
// `text_score(n, col, q)` is `vector_score(n, '{col}_emb', q)` after this
// rewrite. A *vector*-shaped `q` therefore has nothing to embed: it passes
// straight through, `texts_to_embed` stays empty, and `execute` never
// consults an embedder (the embedder call is gated on a non-empty collect
// list). A *string*-shaped `q` stays text even when it looks like
// `"[1.0, 2.0]"` — that ambiguity is resolved in favour of text, and
// CYPHER.md says so.

/// Rewrite helper: parse, rewrite, return the query plus the collected texts.
fn rewrite_ts(
    query: &str,
    params: &HashMap<String, Value>,
) -> Result<(CypherQuery, Vec<(String, String)>), String> {
    let mut parsed = parse_cypher(query).unwrap();
    let rewrite = simplification::rewrite_text_score(&mut parsed, params)?;
    Ok((parsed, rewrite.texts_to_embed))
}

/// The `text_score(...)` call in the first RETURN item.
fn first_return_call(query: &CypherQuery) -> (&String, &Vec<Expression>) {
    for clause in &query.clauses {
        if let Clause::Return(r) = clause {
            if let Expression::FunctionCall { name, args, .. } = &r.items[0].expression {
                return (name, args);
            }
        }
    }
    panic!("expected a function call in the first RETURN item");
}

#[test]
fn test_text_score_list_parameter_passes_through() {
    let mut params = HashMap::new();
    params.insert(
        "q".to_string(),
        Value::List(vec![Value::Float64(1.0), Value::Float64(0.0)]),
    );
    let (query, texts) = rewrite_ts(
        "MATCH (n:Doc) RETURN text_score(n, 'summary', $q) AS s",
        &params,
    )
    .unwrap();

    // Nothing to embed → execute() never asks for an embedder.
    assert!(texts.is_empty(), "a vector query must collect no text");

    let (name, args) = first_return_call(&query);
    assert_eq!(name, "vector_score");
    assert!(matches!(
        &args[1],
        Expression::Literal(Value::String(s)) if s == "summary_emb"
    ));
    // arg 2 is untouched — the caller's own parameter reaches vector_score.
    assert!(matches!(&args[2], Expression::Parameter(p) if p == "q"));
}

#[test]
fn test_text_score_list_literal_passes_through() {
    let params = HashMap::new();
    let (query, texts) = rewrite_ts(
        "MATCH (n:Doc) RETURN text_score(n, 'summary', [1.0, 0.0]) AS s",
        &params,
    )
    .unwrap();

    assert!(texts.is_empty());
    let (name, args) = first_return_call(&query);
    assert_eq!(name, "vector_score");
    assert!(matches!(
        &args[1],
        Expression::Literal(Value::String(s)) if s == "summary_emb"
    ));
    assert!(matches!(&args[2], Expression::ListLiteral(_)));
}

#[test]
fn test_text_score_metric_arg_survives_vector_passthrough() {
    let mut params = HashMap::new();
    params.insert(
        "q".to_string(),
        Value::List(vec![Value::Float64(1.0), Value::Float64(0.0)]),
    );
    let (query, texts) = rewrite_ts(
        "MATCH (n:Doc) RETURN text_score(n, 'summary', $q, 'euclidean') AS s",
        &params,
    )
    .unwrap();

    assert!(texts.is_empty());
    let (name, args) = first_return_call(&query);
    assert_eq!(name, "vector_score");
    assert_eq!(args.len(), 4);
    assert!(matches!(
        &args[3],
        Expression::Literal(Value::String(m)) if m == "euclidean"
    ));
}

#[test]
fn test_text_score_string_parameter_still_collects_text() {
    let mut params = HashMap::new();
    params.insert("q".to_string(), Value::String("hello".to_string()));
    let (query, texts) = rewrite_ts(
        "MATCH (n:Doc) RETURN text_score(n, 'summary', $q) AS s",
        &params,
    )
    .unwrap();

    assert_eq!(texts.len(), 1);
    assert_eq!(texts[0].1, "hello");
    let (name, args) = first_return_call(&query);
    assert_eq!(name, "vector_score");
    assert!(matches!(&args[2], Expression::Parameter(p) if p == &texts[0].0));
}

#[test]
fn test_text_score_json_shaped_string_stays_text() {
    // Locked decision: a string is query *text* in text_score, even when it
    // parses as a JSON vector. vector_score keeps the legacy JSON-string form.
    let mut params = HashMap::new();
    params.insert("q".to_string(), Value::String("[1.0, 0.0]".to_string()));
    let (_, texts) = rewrite_ts(
        "MATCH (n:Doc) RETURN text_score(n, 'summary', $q) AS s",
        &params,
    )
    .unwrap();
    assert_eq!(texts.len(), 1);
    assert_eq!(texts[0].1, "[1.0, 0.0]");
}

#[test]
fn test_text_score_rejects_non_string_non_list_parameter() {
    let mut params = HashMap::new();
    params.insert("q".to_string(), Value::Int64(7));
    let err = rewrite_ts(
        "MATCH (n:Doc) RETURN text_score(n, 'summary', $q) AS s",
        &params,
    )
    .unwrap_err();
    assert!(
        err.contains("must be a string or a list of numbers"),
        "unexpected error: {err}"
    );
}

#[test]
fn test_text_score_unknown_parameter_still_errors() {
    let params = HashMap::new();
    let err = rewrite_ts(
        "MATCH (n:Doc) RETURN text_score(n, 'summary', $q) AS s",
        &params,
    )
    .unwrap_err();
    assert!(err.contains("not found"), "unexpected error: {err}");
}

// ============================================================================
// NDV selectivity on identity fields (Track H2)
// ============================================================================
//
// `property_ndv` feeds `estimate_node_selectivity`'s `type_count / ndv`
// equality estimate. It used to read the property map only, which does not
// hold a type's `node_title_field` (`add_nodes` hoists that column into
// `NodeData.title`), so the scan found nothing, `.max(1)` reported NDV = 1,
// and the filter was scored *completely non-selective* — the planner then
// anchored on the other, larger end of the pattern. These pin the anchor
// choice, which the Cypher differential corpus cannot see (both plans return
// the same rows; only the cost differs).

/// `Doc` (the unfiltered end) outnumbers `Keyword` 3:1, and every `Keyword`
/// has a distinct title. `Keyword` is therefore the right anchor for a
/// `title`-equality filter (one node) and the wrong one only if the filter is
/// scored as matching the whole type.
fn title_anchor_graph() -> DirGraph {
    fn typed(graph: &mut DirGraph, node_type: &str, n: i64) {
        let rows: Vec<Vec<Value>> = (1..=n)
            .map(|i| {
                vec![
                    Value::Int64(i),
                    Value::String(format!("{}-{i}", node_type.to_lowercase())),
                ]
            })
            .collect();
        let df = crate::datatypes::DataFrame::from_cypher_rows(
            vec!["id".to_string(), "title".to_string()],
            rows,
        )
        .unwrap();
        crate::graph::mutation::maintain::add_nodes(
            graph,
            df,
            node_type.to_string(),
            "id".to_string(),
            Some("title".to_string()),
            None,
        )
        .unwrap();
    }
    let mut graph = DirGraph::new();
    typed(&mut graph, "Doc", 3000);
    typed(&mut graph, "Keyword", 1000);
    graph
}

fn optimized_start_variable(query: &str, graph: &DirGraph) -> String {
    let mut query = parse_cypher(query).unwrap();
    optimize(&mut query, graph, &HashMap::new());
    let m = query
        .clauses
        .iter()
        .find_map(|c| match c {
            Clause::Match(m) => Some(m),
            _ => None,
        })
        .expect("expected MATCH clause");
    match &m.patterns[0].elements[0] {
        PatternElement::Node(np) => np
            .variable
            .clone()
            .expect("start node should carry a variable"),
        _ => panic!("expected start node"),
    }
}

#[test]
fn test_ndv_counts_the_title_field() {
    let graph = title_anchor_graph();
    assert_eq!(
        graph.property_ndv("Keyword", "title"),
        Some(1000),
        "`title` is Keyword's node_title_field, so its distinct values live on \
         NodeData.title, not in the property map; reporting 1 (or None) makes \
         the planner score a title equality filter as non-selective"
    );
}

#[test]
fn test_title_equality_anchors_on_the_filtered_type() {
    let graph = title_anchor_graph();
    assert_eq!(
        optimized_start_variable(
            "MATCH (a:Doc)-[:MENTIONS]->(b:Keyword) WHERE b.title = 'keyword-7' RETURN a, b",
            &graph,
        ),
        "b",
        "a unique title equality selects one Keyword; anchoring on the 3000 \
         Docs instead means the filter was scored non-selective (NDV=1)"
    );
}

#[test]
fn test_title_in_list_anchors_on_the_filtered_type() {
    let graph = title_anchor_graph();
    assert_eq!(
        optimized_start_variable(
            "MATCH (a:Doc)-[:MENTIONS]->(b:Keyword) \
             WHERE b.title IN ['keyword-7', 'keyword-9'] RETURN a, b",
            &graph,
        ),
        "b",
        "PropertyMatcher::In reads the same NDV; two of 1000 distinct titles \
         is far more selective than a full Doc scan"
    );
}

/// The reporter's real shape: the filtered type names its identity columns
/// itself (`add_nodes(unique_id_field='term_id', node_title_field='term_name')`),
/// so `term_name` is a *registered alias* for the title field — the matcher
/// resolves it, and the statistic feeding the planner has to resolve it too.
fn aliased_identity_graph() -> DirGraph {
    let mut graph = DirGraph::new();
    let rows: Vec<Vec<Value>> = (1..=3000)
        .map(|i| vec![Value::Int64(i), Value::String(format!("doc-{i}"))])
        .collect();
    let df = crate::datatypes::DataFrame::from_cypher_rows(vec!["id".into(), "title".into()], rows)
        .unwrap();
    crate::graph::mutation::maintain::add_nodes(
        &mut graph,
        df,
        "Doc".to_string(),
        "id".to_string(),
        Some("title".to_string()),
        None,
    )
    .unwrap();

    let rows: Vec<Vec<Value>> = (1..=1000)
        .map(|i| vec![Value::Int64(i), Value::String(format!("term-{i}"))])
        .collect();
    let df = crate::datatypes::DataFrame::from_cypher_rows(
        vec!["term_id".into(), "term_name".into()],
        rows,
    )
    .unwrap();
    crate::graph::mutation::maintain::add_nodes(
        &mut graph,
        df,
        "Term".to_string(),
        "term_id".to_string(),
        Some("term_name".to_string()),
        None,
    )
    .unwrap();
    graph
}

#[test]
fn test_aliased_title_equality_anchors_on_the_filtered_type() {
    let graph = aliased_identity_graph();
    assert_eq!(
        graph.property_ndv("Term", "term_name"),
        Some(1000),
        "the statistic has to resolve the alias, not just the anchor it feeds"
    );
    assert_eq!(
        optimized_start_variable(
            "MATCH (a:Doc)-[:MENTIONS]->(b:Term) WHERE b.term_name = 'term-7' RETURN a, b",
            &graph,
        ),
        "b",
        "`term_name` is Term's registered title alias — the matcher resolves it \
         to the title field, so the NDV statistic must resolve it the same way"
    );
}

#[test]
fn test_aliased_id_equality_anchors_on_the_filtered_type() {
    let graph = aliased_identity_graph();
    assert_eq!(
        graph.property_ndv("Term", "term_id"),
        Some(1000),
        "the statistic has to resolve the alias, not just the anchor it feeds"
    );
    assert_eq!(
        optimized_start_variable(
            "MATCH (a:Doc)-[:MENTIONS]->(b:Term) WHERE b.term_id = 7 RETURN a, b",
            &graph,
        ),
        "b",
        "`term_id` is Term's registered id alias; only a literal `id` gets the \
         dedicated selectivity-1 path, so the alias has to come out of the NDV \
         statistic"
    );
}

#[test]
fn test_absent_property_is_no_information_not_zero_selectivity() {
    // The safety net behind the alias fix: when the scan finds *no* values at
    // all, "distinct = 0" must not collapse into "NDV = 1" — that reads as
    // `type_count / 1`, i.e. a filter that excludes nothing, and anchors the
    // join on the other, larger end. No information means fall back to the
    // flat heuristic.
    let graph = aliased_identity_graph();
    assert_eq!(
        graph.property_ndv("Term", "not_a_property"),
        None,
        "an empty scan is no information, not NDV=1"
    );
    assert_eq!(
        optimized_start_variable(
            "MATCH (a:Doc)-[:MENTIONS]->(b:Term) WHERE b.not_a_property = 'x' RETURN a, b",
            &graph,
        ),
        "b",
        "scanning the 1000 filtered Terms beats driving 3000 Docs through the \
         same filter, however unselective the estimate"
    );
}

// ============================================================================
// Multi-key top-K fusion
// ============================================================================

/// Optimize a query against an empty graph and return its clauses.
fn optimized_clauses(query: &str) -> Vec<Clause> {
    let mut parsed = parse_cypher(query).unwrap();
    let graph = DirGraph::new();
    let params = HashMap::new();
    optimize(&mut parsed, &graph, &params);
    parsed.clauses
}

fn node_scan_top_k_keys(query: &str) -> Option<Vec<FusedSortKey>> {
    optimized_clauses(query).into_iter().find_map(|c| match c {
        Clause::FusedNodeScanTopK { sort_keys, .. } => Some(sort_keys),
        _ => None,
    })
}

fn order_by_top_k_keys(query: &str) -> Option<Vec<FusedSortKey>> {
    optimized_clauses(query).into_iter().find_map(|c| match c {
        Clause::FusedOrderByTopK { sort_keys, .. } => Some(sort_keys),
        _ => None,
    })
}

#[test]
fn test_node_scan_top_k_fuses_multi_key_order_by() {
    // Before 0.15.14 the pass required exactly one ORDER BY item, so this
    // shape fell through to a full sort of every matching node.
    let keys = node_scan_top_k_keys(
        "MATCH (n:Item) RETURN n.title AS t ORDER BY n.p0 DESC, n.p1 ASC, n.p2 DESC LIMIT 10",
    )
    .expect("multi-key ORDER BY + LIMIT must fuse into FusedNodeScanTopK");
    assert_eq!(keys.len(), 3, "every ORDER BY item becomes a sort key");
    let directions: Vec<bool> = keys.iter().map(|k| k.ascending).collect();
    assert_eq!(
        directions,
        vec![false, true, false],
        "each key keeps its own direction"
    );
    let nulls: Vec<NullsPlacement> = keys.iter().map(|k| k.nulls).collect();
    assert_eq!(
        nulls,
        vec![
            NullsPlacement::First,
            NullsPlacement::Last,
            NullsPlacement::First
        ],
        "each key resolves its own default NULLS placement (DESC → First)"
    );
}

#[test]
fn test_top_k_keys_keep_explicit_nulls_placement() {
    let keys =
        node_scan_top_k_keys("MATCH (n:Item) RETURN n.title AS t ORDER BY n.p0 DESC NULLS LAST, n.p1 ASC NULLS FIRST LIMIT 5")
            .expect("explicit NULLS modifiers must still fuse");
    assert_eq!(
        keys.iter().map(|k| k.nulls).collect::<Vec<_>>(),
        vec![NullsPlacement::Last, NullsPlacement::First],
        "an explicit NULLS modifier overrides the direction default"
    );
}

#[test]
fn test_top_k_sort_key_written_as_a_return_alias_resolves_to_its_expression() {
    let keys = node_scan_top_k_keys(
        "MATCH (n:Item) RETURN n.p0 AS a, n.p1 AS b ORDER BY a, b DESC LIMIT 5",
    )
    .expect("ORDER BY over RETURN aliases must fuse");
    assert_eq!(keys.len(), 2);
    for (i, key) in keys.iter().enumerate() {
        assert!(
            matches!(&key.expression, Expression::PropertyAccess { .. }),
            "alias key {i} must be rewritten to the RETURN item's expression, \
             which is what the pre-projection scan can evaluate"
        );
        assert_eq!(
            key.return_item,
            Some(i),
            "the key remembers the RETURN item it projects"
        );
    }
}

#[test]
fn test_top_k_bails_when_a_sort_key_reads_an_alias_it_is_not_equal_to() {
    // `a` is only bound after projection, so `a + 1` evaluates to NULL in the
    // fused scan's row scope — fusing this returned zero rows before 0.15.14.
    assert!(
        node_scan_top_k_keys("MATCH (n:Item) RETURN n.p0 AS a ORDER BY a + 1 LIMIT 5").is_none(),
        "a computed expression over a RETURN alias must not fuse"
    );
    assert!(
        order_by_top_k_keys("MATCH (n:Item) RETURN n.p0 AS a ORDER BY a + 1 LIMIT 5").is_none(),
        "the generic pass must bail on the same shape"
    );
    // A RETURN item that is itself a reference to a sibling alias of the same
    // RETURN is unevaluable too — `x` only exists after this projection runs.
    assert!(
        order_by_top_k_keys("MATCH (n:Item) RETURN n.p0 AS x, x AS y ORDER BY y LIMIT 5").is_none(),
        "a matched RETURN item whose expression reads a sibling alias must bail"
    );
    // But an alias bound *upstream* by WITH is a real binding on the row, so
    // that shape stays fusable.
    assert!(
        order_by_top_k_keys("MATCH (n:Item) WITH n.p0 AS x RETURN x AS y ORDER BY y LIMIT 5")
            .is_some(),
        "an upstream WITH alias is bound before RETURN and must still fuse"
    );
}

#[test]
fn test_generic_top_k_fuses_multi_key_order_by() {
    let keys = order_by_top_k_keys(
        "MATCH (a:Person)-[:KNOWS]->(b:Person) RETURN a.name AS n, b.age AS age \
         ORDER BY b.age DESC, a.name ASC LIMIT 10",
    )
    .expect("multi-key ORDER BY + LIMIT must fuse into FusedOrderByTopK");
    assert_eq!(keys.len(), 2);
    assert_eq!(
        keys.iter().map(|k| k.ascending).collect::<Vec<_>>(),
        vec![false, true],
        "mixed directions survive the rewrite"
    );
    // Written as properties rather than as the RETURN aliases, so they are
    // their own expressions and project nothing.
    assert!(keys.iter().all(|k| k.return_item.is_none()));

    let aliased = order_by_top_k_keys(
        "MATCH (a:Person)-[:KNOWS]->(b:Person) RETURN a.name AS n, b.age AS age \
         ORDER BY age DESC, n ASC LIMIT 10",
    )
    .expect("the same shape written over RETURN aliases must fuse too");
    assert_eq!(
        aliased.iter().map(|k| k.return_item).collect::<Vec<_>>(),
        vec![Some(1), Some(0)],
        "each alias key remembers the RETURN item it projects"
    );
}

#[test]
fn test_top_k_still_bails_on_a_non_literal_limit() {
    assert!(
        node_scan_top_k_keys("MATCH (n:Item) RETURN n.title AS t ORDER BY n.p0, n.p1 LIMIT 1 + 1")
            .is_none(),
        "LIMIT must be a positive integer literal"
    );
}

// ── anchor_element_id ──────────────────────────────────────────────────────

/// The clause's resolved slot anchors after a full optimizer run.
fn anchors_of(query: &str, params: &HashMap<String, Value>) -> Vec<(String, usize)> {
    let mut parsed = parse_cypher(query).unwrap();
    let graph = DirGraph::new();
    optimize(&mut parsed, &graph, params);
    parsed
        .clauses
        .iter()
        .filter_map(|c| match c {
            Clause::Match(m) | Clause::OptionalMatch(m) => Some(&m.node_anchors),
            _ => None,
        })
        .flatten()
        .map(|(v, idx)| (v.clone(), idx.index()))
        .collect()
}

#[test]
fn test_element_id_anchor_literal_and_param_agree() {
    let no_params = HashMap::new();
    let params: HashMap<String, Value> =
        HashMap::from([("eid".to_string(), Value::String("7".into()))]);

    let literal = anchors_of("MATCH (v) WHERE elementId(v) = '7' RETURN v", &no_params);
    assert_eq!(literal, vec![("v".to_string(), 7)]);

    // The spelling a client actually sends — the round-tripped element_id as a
    // bound parameter — must resolve to the same anchor as the literal.
    assert_eq!(
        anchors_of("MATCH (v) WHERE elementId(v) = $eid RETURN v", &params),
        literal
    );
    // Commuted operands, and the integer spelling of the same slot.
    assert_eq!(
        anchors_of("MATCH (v) WHERE $eid = elementId(v) RETURN v", &params),
        literal
    );
    assert_eq!(
        anchors_of("MATCH (v) WHERE elementId(v) = 7 RETURN v", &no_params),
        literal
    );
}

#[test]
fn test_element_id_anchor_bails_on_non_conjunctive_and_unusable_values() {
    let no_params = HashMap::new();
    let params: HashMap<String, Value> =
        HashMap::from([("eid".to_string(), Value::String("7".into()))]);

    // A disjunct constrains nothing: every node is still a candidate.
    assert!(anchors_of(
        "MATCH (v) WHERE elementId(v) = $eid OR v.name = 'x' RETURN v",
        &params
    )
    .is_empty());
    assert!(anchors_of("MATCH (v) WHERE NOT elementId(v) = $eid RETURN v", &params).is_empty());
    // Not a slot: a name, a negative number, an unbound parameter.
    assert!(anchors_of("MATCH (v) WHERE elementId(v) = 'abc' RETURN v", &no_params).is_empty());
    assert!(anchors_of("MATCH (v) WHERE elementId(v) = -3 RETURN v", &no_params).is_empty());
    assert!(anchors_of("MATCH (v) WHERE elementId(v) = $eid RETURN v", &no_params).is_empty());
    // A variable this MATCH does not bind belongs to another clause's search
    // space, so the anchor is not this clause's to record.
    assert!(anchors_of(
        "MATCH (a) MATCH (b) WHERE elementId(a) = $eid RETURN b",
        &params
    )
    .is_empty());
}

#[test]
fn test_element_id_anchor_reads_a_conjunct_and_the_scoped_optional_where() {
    let params: HashMap<String, Value> =
        HashMap::from([("eid".to_string(), Value::String("2".into()))]);

    assert_eq!(
        anchors_of(
            "MATCH (v) WHERE v.name = 'x' AND elementId(v) = $eid RETURN v",
            &params
        ),
        vec![("v".to_string(), 2)],
        "the AND spine is descended"
    );
    assert_eq!(
        anchors_of(
            "MATCH (a:Person) OPTIONAL MATCH (v) WHERE elementId(v) = $eid RETURN v",
            &params
        ),
        vec![("v".to_string(), 2)],
        "OPTIONAL MATCH carries its WHERE inside the clause"
    );
}