kglite 0.16.8

//! Schema validation pass — runs after parse, before optimize.
//!
//! Catches unknown property references inside **pattern-literal syntax**
//! (`{prop: value}`) before the executor commits to a scan. These are
//! unambiguously property names — a typo here can never be a virtual
//! column or computed alias, so rejecting produces zero false positives.
//!
//! Covers:
//! - Variable scope across projections, subqueries, expressions, and updates.
//! - `MATCH (n:T {prop: v})` and `OPTIONAL MATCH` (read patterns).
//! - `EXISTS { MATCH (n:T {prop: v}) }` inside WHERE / AND / OR / NOT
//!   subqueries.
//! - **`CREATE (n:T {prop: v})`** and `CREATE`-style multi-element paths.
//! - **`MERGE (n:T {prop: v})`** including the embedded `CREATE` shape.
//!
//! Deliberately *never rejects on an open schema* (these are legal, so they
//! are warned about non-fatally instead — see below). `lock_schema()` promotes
//! some of them to errors; the bullets note where it does:
//! - Unknown node types in MATCH **on an open schema** (`MATCH
//!   (n:Nonexistent)` legitimately returns zero rows and is a common
//!   existence-check idiom). Under `lock_schema()` this *is* rejected —
//!   see [`validate_label`], which carries the rationale.
//! - Unknown connection types (same rationale, both schema states — an
//!   edge type is not yet part of what a schema lock covers).
//! - Property references in WHERE / RETURN expressions (virtual columns,
//!   timeseries sub-nodes, aliases can be legitimate `n.prop` accesses
//!   not present in `node_type_metadata`). Under `lock_schema()` an absent
//!   one *is* rejected — by [`warnings::strict_read_error`] at the session
//!   layer, not by [`validate_schema`] here.
//! - `SET n.prop = X` and `REMOVE n.prop` — SET may legitimately
//!   introduce new properties depending on kglite's mutation policy;
//!   REMOVE of a non-existent property is benign.
//!
//! ## Non-fatal "did you mean?" warnings
//!
//! [`collect_unknown_pattern_warnings`] gathers the five families of
//! silently-empty-or-null query, all with an edit-distance hint where one
//! applies and none of them rejecting (every shape below is legal Cypher):
//!
//! 1. **Unknown node label / relationship type** in a MATCH — the most common
//!    "why is my query empty?" typo. A warning may only claim the pattern
//!    returns no rows when that is true: an unknown branch of a relationship
//!    alternation (`-[:UNKNOWN|KNOWS]->`) is worded per-branch, because the
//!    pattern still matches through its siblings.
//! 2. **Absent property in a `WHERE`** — `null <op> x` is false, so the
//!    predicate filters out every row.
//! 3. **Absent property in a `RETURN` / `WITH` / `ORDER BY`** — a silently
//!    all-null column, worse than an empty result because a sibling
//!    `n.name` title-aliases to a real value and the rows read as
//!    half-correct. See [`warnings::AbsentSite`].
//! 4. **A relationship pattern pointing the wrong way** — every edge of the
//!    type runs the other way, so the pattern matches nothing. See
//!    [`warnings::reversed_direction_warnings`] for the conservatism rules.
//! 5. **A comparison a declared property type makes vacuous** — `WHERE
//!    p.age > 'forty'` where `p.age IS :: INTEGER` is null on every row.
//!    Both operands are classified: a literal, a `$param` the caller bound,
//!    or a second typed property. Two type sources, DDL declaration before
//!    `define_schema()` field types, each quoted in its own vocabulary. See
//!    [`type_mismatch`] for the family resolver, the precedence and the
//!    never-warn classes. It rides [`warnings::QueryWarnings::type_mismatch`],
//!    and `lock_schema()` promotes only the write-enforced half: a
//!    `define_schema()` field type is a declaration nothing checks at write
//!    time, so that half stays a warning in both schema states.
//!
//! Warnings travel structurally on `QueryDiagnostics::warnings` (every
//! programmatic surface, MCP included) and to stderr via
//! [`warnings::emit_query_warnings`] (kglite's `warning:` convention, for interactive
//! users) — one computation, two consumers.
//!
//! Both surfaces — the fatal check and these warnings — reach patterns through
//! the single traversal in [`walk_query_patterns`] (see it for the covered and
//! uncovered nesting forms). The exception is the var → label map
//! ([`warnings::match_var_labels`]), which families 2-5 consult and which is
//! built from top-level MATCH patterns only, for want of a scope model: a var
//! rebound by a projection (`WITH n AS m`) is simply absent from it, so those
//! checks stay silent about `m` rather than guessing.

use super::super::ast::*;
use super::super::executor::helpers::expression_to_string;
use super::simplification::collect_expression_refs;
use crate::graph::core::pattern_matching::{Pattern, PatternElement};
use crate::graph::mutation::validation::did_you_mean;
use crate::graph::schema::{DirGraph, InternedKey};
use std::collections::{HashMap, HashSet};
use std::convert::Infallible;

mod type_mismatch;
mod warnings;

pub(crate) use type_mismatch::strict_type_error;
pub use warnings::collect_unknown_pattern_warnings;
pub(crate) use warnings::{collect_query_warnings, emit_query_warnings, strict_read_error};
pub use warnings::{query_warning_sink, set_query_warning_sink, QueryWarningSink};

/// Built-in fields valid on any node type, tolerated without a metadata
/// entry — mirrors BUILTIN_FIELDS in `mutation/validation.rs`.
const BUILTIN_FIELDS: &[&str] = &["id", "title", "name", "type"];

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SchemaErrorKind {
    UnknownProperty,
    /// Raised **only under a locked schema** — see [`validate_label`].
    UnknownNodeType,
    UndefinedVariable,
}

#[derive(Debug, Clone)]
pub struct SchemaError {
    #[allow(dead_code)] // Test-only.
    pub kind: SchemaErrorKind,
    pub message: String,
}

impl std::fmt::Display for SchemaError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{}", self.message)
    }
}

impl std::error::Error for SchemaError {}

/// Validate a parsed Cypher query against the graph schema.
///
/// Runs in O(clauses) with O(1) lookups against `node_type_metadata`.
/// Returns early on the first violation.
///
/// Skips validation entirely when the graph has neither declared node
/// types (`node_type_metadata`) nor live nodes (`type_indices`) — that
/// state appears in tests and during initial construction.
pub fn validate_schema(query: &CypherQuery, graph: &DirGraph) -> Result<(), SchemaError> {
    validate_scope(query, &HashSet::new())?;

    if graph.node_type_metadata.is_empty() && graph.type_indices.is_empty() {
        return Ok(());
    }

    validate_query(query, graph)
}

fn undefined_variable(name: &str) -> SchemaError {
    SchemaError {
        kind: SchemaErrorKind::UndefinedVariable,
        message: format!("Undefined variable '{}'", name),
    }
}

fn require_variable(name: &str, scope: &HashSet<String>) -> Result<(), SchemaError> {
    if scope.contains(name) {
        Ok(())
    } else {
        Err(undefined_variable(name))
    }
}

fn bind_pattern(pattern: &Pattern, scope: &mut HashSet<String>) {
    for element in &pattern.elements {
        match element {
            PatternElement::Node(node) => scope.extend(node.variable.iter().cloned()),
            PatternElement::Edge(edge) => scope.extend(edge.variable.iter().cloned()),
        }
    }
}

fn bind_create_pattern(pattern: &CreatePattern, scope: &mut HashSet<String>) {
    for element in &pattern.elements {
        match element {
            CreateElement::Node(node) => scope.extend(node.variable.iter().cloned()),
            CreateElement::Edge(edge) => scope.extend(edge.variable.iter().cloned()),
        }
    }
}

/// Validate a row source's own expression against the scope it sees, then
/// declare the variable it binds.
///
/// `UNWIND` and `LOAD CSV` are the only clauses that originate rows and bind
/// exactly one variable, and the rule is identical for both, so it lives here
/// rather than twice in [`validate_scope`]. For `LOAD CSV` the incoming scope is
/// always empty (it is leading-position only), which is what lets `FROM $path`
/// through while rejecting `FROM n.path`.
fn bind_row_source(clause: &Clause, scope: &mut HashSet<String>) -> Result<(), SchemaError> {
    let (source, variable) = match clause {
        Clause::Unwind(unwind) => (&unwind.expression, &unwind.alias),
        Clause::LoadCsv(load) => (&load.source, &load.variable),
        // Unreachable: the caller matches exactly these two variants.
        _ => return Ok(()),
    };
    validate_expression_scope(source, scope)?;
    scope.insert(variable.clone());
    Ok(())
}

/// What a trailing `ORDER BY` may reference after an **aggregating** RETURN.
///
/// An aggregating RETURN emits one row per group. Only two kinds of thing have
/// a defined value on that row: the projected columns, and the variables the
/// grouping keys pin down (the executor carries their bindings forward — see
/// `executor::helpers::carry_group_bindings`). Ordering by anything else is
/// what Neo4j rejects outright: `count(c)` collapses many `c`s into a scalar,
/// so `ORDER BY c.label` has no single value to sort on.
///
/// kglite splits the difference: the well-defined half sorts correctly, the
/// ambiguous half is rejected with a message naming the fix. Before this check
/// existed it evaluated the sort key to NULL on every row and handed back
/// insertion order as though the clause had been honoured.
struct AggregateOrderScope {
    /// Projected column names plus every variable the grouping keys read.
    allowed: HashSet<String>,
    /// Rendered aggregate projections → their alias, if any. `ORDER BY count(p)`
    /// resolves at runtime only when `count(p)` is projected *unaliased*, since
    /// then the result column is literally named `count(p)`; aliasing it moves
    /// the value to the alias and leaves the expression form unresolvable.
    aggregates: HashMap<String, Option<String>>,
    /// Projected aliases, listed in the error message as the way out.
    aliases: Vec<String>,
}

impl AggregateOrderScope {
    /// `None` when no restriction applies — either the projection does not
    /// aggregate (a plain projection keeps every binding on its rows, so
    /// ordering by a non-projected expression stays well defined), or it
    /// contains `*` and we cannot enumerate what it carries.
    fn for_projection(items: &[ReturnItem]) -> Option<Self> {
        if !items
            .iter()
            .any(|item| is_aggregate_expression(&item.expression))
        {
            return None;
        }
        if items
            .iter()
            .any(|item| matches!(item.expression, Expression::Star))
        {
            return None;
        }

        let mut allowed = HashSet::new();
        let mut aliases = Vec::new();
        let mut aggregates = HashMap::new();
        for item in items {
            if let Some(alias) = &item.alias {
                allowed.insert(alias.clone());
                aliases.push(alias.clone());
            } else if let Expression::Variable(name) = &item.expression {
                allowed.insert(name.clone());
                aliases.push(name.clone());
            }
            if is_aggregate_expression(&item.expression) {
                aggregates.insert(expression_to_string(&item.expression), item.alias.clone());
            } else {
                collect_expression_refs(&item.expression, &mut allowed);
            }
        }
        Some(AggregateOrderScope {
            allowed,
            aggregates,
            aliases,
        })
    }

    fn projected_hint(&self) -> String {
        if self.aliases.is_empty() {
            String::new()
        } else {
            format!(
                ", or order by a projected column ({})",
                self.aliases.join(", ")
            )
        }
    }

    fn validate(&self, order: &OrderByClause) -> Result<(), SchemaError> {
        for item in &order.items {
            if is_aggregate_expression(&item.expression) {
                let rendered = expression_to_string(&item.expression);
                match self.aggregates.get(&rendered) {
                    // Projected unaliased — the executor resolves the sort key
                    // from the identically-named result column.
                    Some(None) => continue,
                    Some(Some(alias)) => {
                        return Err(SchemaError {
                            kind: SchemaErrorKind::UndefinedVariable,
                            message: format!(
                                "ORDER BY '{rendered}' cannot be resolved after an aggregating \
                                 RETURN because that aggregate is projected as '{alias}'. \
                                 Order by the alias instead (`ORDER BY {alias}`)."
                            ),
                        });
                    }
                    None => {
                        return Err(SchemaError {
                            kind: SchemaErrorKind::UndefinedVariable,
                            message: format!(
                                "ORDER BY cannot compute the aggregate '{rendered}' after an \
                                 aggregating RETURN. Project it first \
                                 (e.g. `RETURN ..., {rendered} AS sort_key ORDER BY sort_key`){}.",
                                self.projected_hint()
                            ),
                        });
                    }
                }
            }
            let mut refs = HashSet::new();
            collect_expression_refs(&item.expression, &mut refs);
            for name in refs {
                if !self.allowed.contains(&name) {
                    return Err(SchemaError {
                        kind: SchemaErrorKind::UndefinedVariable,
                        message: format!(
                            "ORDER BY cannot use '{name}' after an aggregating RETURN: \
                             '{name}' is not part of the grouping keys, so it has no single \
                             value per group. Add the sort key to the RETURN list \
                             (e.g. `RETURN ..., {name}.<property> AS sort_key ORDER BY sort_key`){}.",
                            self.projected_hint()
                        ),
                    });
                }
            }
        }
        Ok(())
    }
}

/// Validate a RETURN's own expressions, then produce the scope its trailing
/// clauses see: everything already in scope plus the projection's aliases.
///
/// RETURN is terminal for the pipeline, so unlike WITH it does not narrow the
/// scope — HAVING and a trailing ORDER BY may still name pre-projection
/// variables. Whether an *aggregating* RETURN leaves those variables with a
/// usable value is a separate question, answered by [`AggregateOrderScope`].
fn scope_after_return(
    return_clause: &ReturnClause,
    scope: HashSet<String>,
) -> Result<HashSet<String>, SchemaError> {
    for item in &return_clause.items {
        validate_expression_scope(&item.expression, &scope)?;
    }
    let mut having_scope = scope;
    having_scope.extend(
        return_clause
            .items
            .iter()
            .filter_map(|item| item.alias.clone()),
    );
    if let Some(having) = &return_clause.having {
        validate_predicate_scope(having, &having_scope)?;
    }
    Ok(having_scope)
}

/// Validate a WITH's own expressions, then produce the scope it projects
/// forward: only the aliases and bare variables it names, unless it carries
/// `*` (which preserves the incoming scope wholesale).
fn scope_after_with(
    with_clause: &WithClause,
    scope: HashSet<String>,
) -> Result<HashSet<String>, SchemaError> {
    for item in &with_clause.items {
        validate_expression_scope(&item.expression, &scope)?;
    }
    let preserves_all = with_clause
        .items
        .iter()
        .any(|item| matches!(item.expression, Expression::Star));
    let mut projected = if preserves_all { scope } else { HashSet::new() };
    for item in &with_clause.items {
        if let Some(alias) = &item.alias {
            projected.insert(alias.clone());
        } else if let Expression::Variable(name) = &item.expression {
            projected.insert(name.clone());
        }
    }
    if let Some(where_clause) = &with_clause.where_clause {
        validate_predicate_scope(&where_clause.predicate, &projected)?;
    }
    Ok(projected)
}

/// Every property expression written by a CREATE / MERGE pattern must resolve
/// in `scope`. Shared by both clauses, which write the same element shape.
fn validate_write_pattern_properties(
    pattern: &CreatePattern,
    scope: &HashSet<String>,
) -> Result<(), SchemaError> {
    for element in &pattern.elements {
        let properties = match element {
            CreateElement::Node(node) => &node.properties,
            CreateElement::Edge(edge) => &edge.properties,
        };
        for (_, expression) in properties {
            validate_expression_scope(expression, scope)?;
        }
    }
    Ok(())
}

/// SET items — the clause's own, or a MERGE's `ON CREATE` / `ON MATCH` list.
fn validate_set_items(items: &[SetItem], scope: &HashSet<String>) -> Result<(), SchemaError> {
    for item in items {
        match item {
            SetItem::Property {
                variable,
                expression,
                ..
            }
            | SetItem::Map {
                variable,
                expression,
                ..
            } => {
                require_variable(variable, scope)?;
                validate_expression_scope(expression, scope)?;
            }
            SetItem::Label { variable, .. } => require_variable(variable, scope)?,
        }
    }
    Ok(())
}

fn validate_scope(query: &CypherQuery, initial: &HashSet<String>) -> Result<(), SchemaError> {
    let mut scope = initial.clone();
    // Set by an aggregating RETURN, consumed by the ORDER BY that follows it.
    // Rides through a trailing SKIP/LIMIT; any other clause clears it.
    let mut aggregate_order_scope: Option<AggregateOrderScope> = None;
    for clause in &query.clauses {
        if !matches!(
            clause,
            Clause::OrderBy(_) | Clause::Skip(_) | Clause::Limit(_)
        ) {
            aggregate_order_scope = None;
        }
        match clause {
            Clause::Match(m) | Clause::OptionalMatch(m) => {
                for pattern in &m.patterns {
                    bind_pattern(pattern, &mut scope);
                }
                scope.extend(m.path_assignments.iter().map(|path| path.variable.clone()));
                // The clause's own WHERE sees this clause's pattern variables,
                // so it is validated only after binding them.
                if let Some(wc) = &m.where_clause {
                    validate_predicate_scope(&wc.predicate, &scope)?;
                }
            }
            Clause::Where(where_clause) => {
                validate_predicate_scope(&where_clause.predicate, &scope)?
            }
            Clause::Return(return_clause) => {
                scope = scope_after_return(return_clause, scope)?;
                aggregate_order_scope = AggregateOrderScope::for_projection(&return_clause.items);
            }
            Clause::With(with_clause) => {
                scope = scope_after_with(with_clause, scope)?;
            }
            Clause::OrderBy(order) => {
                for item in &order.items {
                    validate_expression_scope(&item.expression, &scope)?;
                }
                if let Some(restriction) = &aggregate_order_scope {
                    restriction.validate(order)?;
                }
            }
            Clause::Skip(skip) => validate_expression_scope(&skip.count, &scope)?,
            Clause::Limit(limit) => validate_expression_scope(&limit.count, &scope)?,
            Clause::Unwind(_) | Clause::LoadCsv(_) => bind_row_source(clause, &mut scope)?,
            Clause::Union(union) => validate_scope(&union.query, initial)?,
            Clause::Create(create) => {
                for pattern in &create.patterns {
                    bind_create_pattern(pattern, &mut scope);
                    validate_write_pattern_properties(pattern, &scope)?;
                }
            }
            Clause::Set(set) => validate_set_items(&set.items, &scope)?,
            Clause::Delete(delete) => {
                for expression in &delete.expressions {
                    validate_expression_scope(expression, &scope)?;
                }
            }
            Clause::Remove(remove) => {
                for item in &remove.items {
                    let variable = match item {
                        RemoveItem::Property { variable, .. }
                        | RemoveItem::Label { variable, .. } => variable,
                    };
                    require_variable(variable, &scope)?;
                }
            }
            Clause::Merge(merge) => {
                bind_create_pattern(&merge.pattern, &mut scope);
                validate_write_pattern_properties(&merge.pattern, &scope)?;
                for items in [&merge.on_create, &merge.on_match].into_iter().flatten() {
                    validate_set_items(items, &scope)?;
                }
            }
            Clause::Foreach {
                variable,
                list,
                body,
            } => {
                validate_expression_scope(list, &scope)?;
                let mut inner = scope.clone();
                inner.insert(variable.clone());
                validate_scope(
                    &CypherQuery {
                        clauses: body.clone(),
                        explain: false,
                        profile: false,
                        output_format: OutputFormat::Default,
                        optimizer_tags: Vec::new(),
                    },
                    &inner,
                )?;
            }
            Clause::Call(call) => {
                for (_, expression) in &call.parameters {
                    validate_expression_scope(expression, &scope)?;
                }
                scope.extend(
                    call.yield_items
                        .iter()
                        .map(|item| item.alias.as_ref().unwrap_or(&item.name).clone()),
                );
            }
            Clause::CallSubquery { import, body } => {
                for name in import {
                    require_variable(name, &scope)?;
                }
                let imported: HashSet<String> = import.iter().cloned().collect();
                validate_scope(body, &imported)?;
                if let Some(Clause::Return(return_clause)) = body
                    .clauses
                    .iter()
                    .rev()
                    .find(|clause| matches!(clause, Clause::Return(_)))
                {
                    for item in &return_clause.items {
                        if let Some(alias) = &item.alias {
                            scope.insert(alias.clone());
                        } else if let Expression::Variable(name) = &item.expression {
                            scope.insert(name.clone());
                        }
                    }
                }
            }
            // Physical clauses exist only after validation.
            _ => {}
        }
    }
    Ok(())
}

fn validate_predicate_scope(
    predicate: &Predicate,
    scope: &HashSet<String>,
) -> Result<(), SchemaError> {
    match predicate {
        Predicate::And(left, right) | Predicate::Or(left, right) | Predicate::Xor(left, right) => {
            validate_predicate_scope(left, scope)?;
            validate_predicate_scope(right, scope)
        }
        Predicate::Not(inner) => validate_predicate_scope(inner, scope),
        Predicate::Comparison { left, right, .. } => {
            validate_expression_scope(left, scope)?;
            validate_expression_scope(right, scope)
        }
        Predicate::IsNull(expression)
        | Predicate::IsNotNull(expression)
        | Predicate::InLiteralSet {
            expr: expression, ..
        } => validate_expression_scope(expression, scope),
        Predicate::In { expr, list } => {
            validate_expression_scope(expr, scope)?;
            for item in list {
                validate_expression_scope(item, scope)?;
            }
            Ok(())
        }
        Predicate::InExpression { expr, list_expr } => {
            validate_expression_scope(expr, scope)?;
            validate_expression_scope(list_expr, scope)
        }
        Predicate::StartsWith { expr, pattern }
        | Predicate::EndsWith { expr, pattern }
        | Predicate::Contains { expr, pattern } => {
            validate_expression_scope(expr, scope)?;
            validate_expression_scope(pattern, scope)
        }
        Predicate::LabelCheck { variable, .. } => require_variable(variable, scope),
        Predicate::Exists {
            patterns,
            where_clause,
            ..
        } => {
            let mut inner = scope.clone();
            for pattern in patterns {
                bind_pattern(pattern, &mut inner);
            }
            if let Some(where_clause) = where_clause {
                validate_predicate_scope(where_clause, &inner)?;
            }
            Ok(())
        }
    }
}

fn validate_expression_scope(
    expression: &Expression,
    scope: &HashSet<String>,
) -> Result<(), SchemaError> {
    match expression {
        Expression::Variable(name) | Expression::PropertyAccess { variable: name, .. } => {
            require_variable(name, scope)
        }
        Expression::FunctionCall { args, .. } | Expression::ListLiteral(args) => {
            for argument in args {
                validate_expression_scope(argument, scope)?;
            }
            Ok(())
        }
        Expression::Add(left, right)
        | Expression::Subtract(left, right)
        | Expression::Multiply(left, right)
        | Expression::Divide(left, right)
        | Expression::Modulo(left, right)
        | Expression::Concat(left, right)
        | Expression::IndexAccess {
            expr: left,
            index: right,
        } => {
            validate_expression_scope(left, scope)?;
            validate_expression_scope(right, scope)
        }
        Expression::Negate(inner)
        | Expression::IsNull(inner)
        | Expression::IsNotNull(inner)
        | Expression::ExprPropertyAccess { expr: inner, .. } => {
            validate_expression_scope(inner, scope)
        }
        Expression::Case {
            operand,
            when_clauses,
            else_expr,
        } => {
            if let Some(operand) = operand {
                validate_expression_scope(operand, scope)?;
            }
            for (condition, result) in when_clauses {
                match condition {
                    CaseCondition::Predicate(predicate) => {
                        validate_predicate_scope(predicate, scope)?
                    }
                    CaseCondition::Expression(expression) => {
                        validate_expression_scope(expression, scope)?
                    }
                }
                validate_expression_scope(result, scope)?;
            }
            if let Some(else_expr) = else_expr {
                validate_expression_scope(else_expr, scope)?;
            }
            Ok(())
        }
        Expression::ListComprehension {
            variable,
            list_expr,
            filter,
            map_expr,
        } => {
            validate_expression_scope(list_expr, scope)?;
            let mut inner = scope.clone();
            inner.insert(variable.clone());
            if let Some(filter) = filter {
                validate_predicate_scope(filter, &inner)?;
            }
            if let Some(map_expr) = map_expr {
                validate_expression_scope(map_expr, &inner)?;
            }
            Ok(())
        }
        Expression::ListSlice { expr, start, end } => {
            validate_expression_scope(expr, scope)?;
            if let Some(start) = start {
                validate_expression_scope(start, scope)?;
            }
            if let Some(end) = end {
                validate_expression_scope(end, scope)?;
            }
            Ok(())
        }
        Expression::MapProjection { variable, items } => {
            require_variable(variable, scope)?;
            for item in items {
                if let MapProjectionItem::Alias { expr, .. } = item {
                    validate_expression_scope(expr, scope)?;
                }
            }
            Ok(())
        }
        Expression::MapLiteral(entries) => {
            for (_, expression) in entries {
                validate_expression_scope(expression, scope)?;
            }
            Ok(())
        }
        Expression::QuantifiedList {
            variable,
            list_expr,
            filter,
            ..
        } => {
            validate_expression_scope(list_expr, scope)?;
            let mut inner = scope.clone();
            inner.insert(variable.clone());
            validate_predicate_scope(filter, &inner)
        }
        Expression::Reduce {
            accumulator,
            init,
            variable,
            list_expr,
            body,
        } => {
            validate_expression_scope(init, scope)?;
            validate_expression_scope(list_expr, scope)?;
            let mut inner = scope.clone();
            inner.insert(accumulator.clone());
            inner.insert(variable.clone());
            validate_expression_scope(body, &inner)
        }
        Expression::PredicateExpr(predicate) => validate_predicate_scope(predicate, scope),
        Expression::WindowFunction {
            partition_by,
            order_by,
            ..
        } => {
            for expression in partition_by {
                validate_expression_scope(expression, scope)?;
            }
            for item in order_by {
                validate_expression_scope(&item.expression, scope)?;
            }
            Ok(())
        }
        Expression::CountSubquery {
            patterns,
            where_clause,
            ..
        } => {
            let mut inner = scope.clone();
            for pattern in patterns {
                bind_pattern(pattern, &mut inner);
            }
            if let Some(where_clause) = where_clause {
                validate_predicate_scope(where_clause, &inner)?;
            }
            Ok(())
        }
        Expression::Literal(_) | Expression::Parameter(_) | Expression::Star => Ok(()),
    }
}

fn validate_query(query: &CypherQuery, graph: &DirGraph) -> Result<(), SchemaError> {
    walk_query_patterns(query, &mut |site| match site {
        PatternSite::Read(pattern) => validate_pattern(pattern, graph),
        PatternSite::Write(pattern) => validate_create_pattern(pattern, graph),
    })
}

/// A pattern the shared traversal hands to its visitor.
///
/// Read and write patterns are kept apart because the two consumers treat them
/// differently: [`validate_schema`] checks pattern-literal property names in
/// both, while [`collect_unknown_pattern_warnings`] must stay silent on writes
/// — `CREATE (n:NewType)` on an open schema is how a type comes into
/// existence, not a typo.
enum PatternSite<'q> {
    Read(&'q Pattern),
    Write(&'q CreatePattern),
}

/// The one traversal that answers "where can a pattern appear in a query?".
///
/// Both schema surfaces route through it — [`validate_schema`] (fatal, locked
/// schemas) and [`collect_unknown_pattern_warnings`] (non-fatal, open
/// schemas). Two walkers drift: the warning collector used to walk top-level
/// `MATCH` / `OPTIONAL MATCH` itself, so a typo'd label inside `CALL {}`,
/// `WHERE EXISTS {}` or a `UNION` branch produced no warning while the
/// identical typo at top level did. With one walker a newly-covered nesting
/// form lands on both surfaces at once.
///
/// Covers, recursively so nesting composes: `MATCH` / `OPTIONAL MATCH`,
/// `EXISTS {}` at any `AND` / `OR` / `XOR` / `NOT` depth inside a `WHERE`
/// (standalone or attached to a `WITH`), `CALL {}` bodies, `UNION` branches,
/// and `CREATE` / `MERGE` write patterns.
///
/// Not covered — equally for both consumers, which is what keeps them in
/// step: `COUNT {}` subqueries and `EXISTS {}` in expression position (both
/// reach patterns through [`Expression`] rather than [`Predicate`]), and
/// update clauses inside `FOREACH`.
fn walk_query_patterns<E>(
    query: &CypherQuery,
    visit: &mut impl FnMut(PatternSite<'_>) -> Result<(), E>,
) -> Result<(), E> {
    for clause in &query.clauses {
        walk_clause_patterns(clause, visit)?;
    }
    Ok(())
}

fn walk_clause_patterns<E>(
    clause: &Clause,
    visit: &mut impl FnMut(PatternSite<'_>) -> Result<(), E>,
) -> Result<(), E> {
    match clause {
        Clause::Match(m) | Clause::OptionalMatch(m) => {
            for pattern in &m.patterns {
                visit(PatternSite::Read(pattern))?;
            }
            if let Some(wc) = &m.where_clause {
                walk_predicate_patterns(&wc.predicate, visit)?;
            }
        }
        Clause::Where(w) => walk_predicate_patterns(&w.predicate, visit)?,
        Clause::With(w) => {
            if let Some(wc) = &w.where_clause {
                walk_predicate_patterns(&wc.predicate, visit)?;
            }
        }
        Clause::Union(u) => walk_query_patterns(&u.query, visit)?,
        // The importing `WITH` was stripped at parse time, so the body is a
        // self-contained sub-pipeline: recurse into it as into a UNION branch.
        Clause::CallSubquery { body, .. } => walk_query_patterns(body, visit)?,
        Clause::Create(c) => {
            for pattern in &c.patterns {
                visit(PatternSite::Write(pattern))?;
            }
        }
        // MERGE's `ON CREATE SET` / `ON MATCH SET` use `SetItem`, which the
        // schema check intentionally skips — see the module doc-comment.
        Clause::Merge(m) => visit(PatternSite::Write(&m.pattern))?,
        // LOAD CSV binds an untyped map/list, so it carries no pattern —
        // same as UNWIND.
        _ => {}
    }
    Ok(())
}

/// Descend a predicate to the `EXISTS { ... }` patterns nested inside it.
fn walk_predicate_patterns<E>(
    predicate: &Predicate,
    visit: &mut impl FnMut(PatternSite<'_>) -> Result<(), E>,
) -> Result<(), E> {
    match predicate {
        Predicate::And(a, b) | Predicate::Or(a, b) | Predicate::Xor(a, b) => {
            walk_predicate_patterns(a, visit)?;
            walk_predicate_patterns(b, visit)?;
        }
        Predicate::Not(p) => walk_predicate_patterns(p, visit)?,
        Predicate::Exists {
            patterns,
            where_clause,
            ..
        } => {
            for pattern in patterns {
                visit(PatternSite::Read(pattern))?;
            }
            if let Some(w) = where_clause {
                walk_predicate_patterns(w, visit)?;
            }
        }
        _ => {}
    }
    Ok(())
}

/// Infallible companion to [`walk_query_patterns`] for visitors that only
/// collect — the warning path never rejects, so it should not have to spell
/// out an error type it cannot produce.
fn for_each_query_pattern(query: &CypherQuery, visit: &mut impl FnMut(PatternSite<'_>)) {
    let outcome = walk_query_patterns(query, &mut |site| -> Result<(), Infallible> {
        visit(site);
        Ok(())
    });
    match outcome {
        Ok(()) => {}
        // `Infallible` is uninhabited — this arm cannot be constructed.
        Err(never) => match never {},
    }
}

/// Validate a CREATE / MERGE pattern's node-pattern property names
/// against the schema. Edge-pattern properties aren't validated —
/// `connection_type_metadata` is keyed differently and edge schemas
/// in kglite are looser; revisit if a real divergence shows up.
fn validate_create_pattern(pattern: &CreatePattern, graph: &DirGraph) -> Result<(), SchemaError> {
    for element in &pattern.elements {
        if let CreateElement::Node(np) = element {
            if let Some(ref node_type) = np.label {
                for (prop_name, _expr) in &np.properties {
                    validate_property(node_type, prop_name, graph)?;
                }
            }
        }
    }
    Ok(())
}

/// Validate a read pattern's labels (locked schemas only) and its
/// pattern-literal property names.
///
/// A node pattern carrying no explicit label is skipped entirely: property
/// validation needs a node type, and kglite does not infer one from an outer
/// binding, so `CALL { WITH p MATCH (p {prp: v}) }` is not checked.
fn validate_pattern(pattern: &Pattern, graph: &DirGraph) -> Result<(), SchemaError> {
    for element in &pattern.elements {
        if let PatternElement::Node(np) = element {
            // Label check first: on a locked schema an unknown label is a
            // typo, not an empty result set. Checked before the property
            // loop so `MATCH (i:Isue {titel: 1})` reports the label — the
            // outer mistake — rather than a property of a type that does
            // not exist.
            if graph.schema_locked {
                for label in np.node_type.iter().chain(np.extra_labels.iter()) {
                    validate_label(label, graph)?;
                }
            }
            if let Some(ref node_type) = np.node_type {
                if let Some(ref props) = np.properties {
                    for prop_name in props.keys() {
                        validate_property(node_type, prop_name, graph)?;
                    }
                }
            }
        }
    }
    Ok(())
}

/// A label is "known" if it is a declared primary type, currently has live
/// nodes, or is a secondary label applied via `add_label` — `MATCH
/// (n:Reviewer)` is valid even though `Reviewer` is no node's primary type.
/// Mirrors the identical three-way test in
/// [`collect_unknown_pattern_warnings`]; keep the two in step.
fn label_known(label: &str, graph: &DirGraph) -> bool {
    graph.node_type_metadata.contains_key(label)
        || graph.type_indices.contains_key(label)
        || graph
            .secondary_label_index
            .contains_key(&InternedKey::from_str(label))
}

/// Reject an unknown node label. **Caller gates on `graph.schema_locked`.**
///
/// On an open schema kglite is schemaless by design and `MATCH
/// (n:Nonexistent)` legitimately returns zero rows (the existence-check
/// idiom), so this is never reached and
/// [`collect_unknown_pattern_warnings`] delivers the typo hint non-fatally
/// instead. `lock_schema()` is the opt-in "catch my typos" mechanism: it
/// already rejects an unknown *property* here, and an unknown *node type*
/// on the CREATE write path
/// ([`validate_node_creation`](crate::graph::mutation::validation::validate_node_creation)).
/// This closes the read side, where a typo'd label previously returned `[]`
/// — indistinguishable from "no matching data".
///
/// The message deliberately reuses the write path's wording, so a locked
/// schema reports the same mistake identically whether it is read or written.
fn validate_label(label: &str, graph: &DirGraph) -> Result<(), SchemaError> {
    if label_known(label, graph) {
        return Ok(());
    }
    // Cold path only — the candidate list is built solely to explain a
    // confirmed typo.
    let mut valid: Vec<&str> = graph
        .node_type_metadata
        .keys()
        .map(|s| s.as_str())
        .chain(graph.type_indices.keys())
        .collect();
    // `try_resolve`, not `resolve`: the latter panics on a key the interner
    // has no entry for, and an error-message builder must never be the thing
    // that takes the process down. A key we cannot name is simply omitted
    // from the hint.
    valid.extend(
        graph
            .secondary_label_index
            .keys()
            .filter_map(|k| graph.interner.try_resolve(*k)),
    );
    valid.sort_unstable();
    valid.dedup();
    let hint = did_you_mean(label, &valid);
    Err(SchemaError {
        kind: SchemaErrorKind::UnknownNodeType,
        message: format!(
            "Unknown node type '{}'.{}\n  Valid types: {}",
            label,
            hint,
            valid.join(", ")
        ),
    })
}

/// Whether `property` is named by the graph's *declared* schema for
/// `node_type` — `define_schema`'s `required` / `optional` / `types`, the
/// declared primary key, and any `unique` tuple.
///
/// `node_type_metadata` is observed, not declared: it is populated from the
/// values that have actually been written, so a property a caller declared up
/// front but has not stored yet is absent from it. Reading only that map made
/// the typo-guard reject `CREATE (n:Item {p1: 1})` on a type whose schema
/// declares `p1`. The guard itself is deliberate for *undeclared* properties
/// and stays; this closes the case where the answer is written down.
fn property_is_declared(node_type: &str, property: &str, graph: &DirGraph) -> bool {
    let Some(schema) = graph.schema_definition.as_ref() else {
        return false;
    };
    let Some(node) = schema.node_schemas.get(node_type) else {
        return false;
    };
    node.required_fields.iter().any(|f| f == property)
        || node.optional_fields.iter().any(|f| f == property)
        || node.field_types.contains_key(property)
        || node.primary_key.as_deref() == Some(property)
        || node
            .unique
            .iter()
            .flatten()
            .any(|tuple| tuple.iter().any(|f| f == property))
}

fn validate_property(node_type: &str, property: &str, graph: &DirGraph) -> Result<(), SchemaError> {
    if BUILTIN_FIELDS.contains(&property) {
        return Ok(());
    }
    let Some(type_props) = graph.node_type_metadata.get(node_type) else {
        // No declared property metadata for this type — skip rather than
        // false-positive on dynamically-typed or under-declared graphs.
        return Ok(());
    };
    if type_props.is_empty() || type_props.contains_key(property) {
        return Ok(());
    }
    if property_is_declared(node_type, property, graph) {
        return Ok(());
    }
    let candidates: Vec<&str> = type_props.keys().map(|s| s.as_str()).collect();
    let hint = did_you_mean(property, &candidates);
    let mut sorted = candidates;
    sorted.sort();
    Err(SchemaError {
        kind: SchemaErrorKind::UnknownProperty,
        message: format!(
            "Unknown property '{}' on {}.{}\n  Valid properties: {}",
            property,
            node_type,
            hint,
            sorted.join(", ")
        ),
    })
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::graph::languages::cypher::parser::parse_cypher;

    pub(super) fn graph_with_schema() -> DirGraph {
        let mut g = DirGraph::new();
        let mut person_props = HashMap::new();
        person_props.insert("age".to_string(), "int".to_string());
        person_props.insert("email".to_string(), "string".to_string());
        g.upsert_node_type_metadata("Person", person_props);
        let mut paper_props = HashMap::new();
        paper_props.insert("year".to_string(), "int".to_string());
        g.upsert_node_type_metadata("Paper", paper_props);
        g.upsert_connection_type_metadata("KNOWS", "Person", "Person", HashMap::new());
        g.upsert_connection_type_metadata("AUTHORED", "Person", "Paper", HashMap::new());
        g
    }

    /// A property the graph's schema *declares* is not a typo, even though the
    /// type has never carried a value for it (see [`property_is_declared`]).
    ///
    /// The rejection was self-perpetuating: a `CREATE` is how the property
    /// would come to be written, so a Cypher statement stream literally could
    /// not grow a type's schema, whatever the caller declared.
    #[test]
    fn a_declared_property_is_not_a_typo() {
        use crate::graph::schema::{NodeSchemaDefinition, SchemaDefinition, SchemaInstall};

        let mut g = graph_with_schema();
        let mut declared = SchemaDefinition::default();
        let mut person = NodeSchemaDefinition {
            required_fields: vec!["age".to_string()],
            optional_fields: vec!["nickname".to_string()],
            ..Default::default()
        };
        person
            .field_types
            .insert("height".to_string(), "float64".to_string());
        person.unique = Some(vec![vec!["passport".to_string()]]);
        declared.node_schemas.insert("Person".to_string(), person);
        g.set_schema(declared, SchemaInstall::Replace)
            .expect("schema installs on an empty graph");

        for query in [
            "CREATE (n:Person {nickname: 'Al'})",
            "CREATE (n:Person {height: 1.8})",
            "CREATE (n:Person {passport: 'X1'})",
            "MATCH (n:Person {nickname: 'Al'}) RETURN n",
        ] {
            let q = parse_cypher(query).unwrap();
            assert!(
                validate_schema(&q, &g).is_ok(),
                "declared property rejected by the typo-guard: {query} -> {:?}",
                validate_schema(&q, &g).unwrap_err().message
            );
        }

        // The guard itself is unchanged for a property nobody declared.
        let q = parse_cypher("CREATE (n:Person {nicknmae: 'Al'})").unwrap();
        let err = validate_schema(&q, &g).expect_err("undeclared property must still be rejected");
        assert!(
            err.message.contains("Unknown property 'nicknmae'"),
            "{}",
            err.message
        );
    }

    #[test]
    fn validates_known_node_type() {
        let g = graph_with_schema();
        let q = parse_cypher("MATCH (n:Person) RETURN n").unwrap();
        assert!(validate_schema(&q, &g).is_ok());
    }

    #[test]
    fn tolerates_unknown_node_type() {
        // `MATCH (n:Nonexistent) RETURN n` is valid Cypher — returns 0
        // rows, so validation must not reject it. The typo is surfaced
        // non-fatally instead, by `collect_unknown_pattern_warnings`.
        let g = graph_with_schema();
        let q = parse_cypher("MATCH (n:person) RETURN n").unwrap();
        assert!(validate_schema(&q, &g).is_ok());
    }

    #[test]
    fn tolerates_unknown_connection_type() {
        let g = graph_with_schema();
        let q = parse_cypher("MATCH (a:Person)-[:nonexistent]->(b:Person) RETURN a").unwrap();
        assert!(validate_schema(&q, &g).is_ok());
    }

    #[test]
    fn warns_unknown_node_label_with_hint() {
        let g = graph_with_schema();
        let q = parse_cypher("MATCH (n:Persn) RETURN n").unwrap();
        let warnings = collect_unknown_pattern_warnings(&q, &g);
        assert_eq!(warnings.len(), 1, "got: {warnings:?}");
        assert!(warnings[0].contains("unknown node label 'Persn'"));
        assert!(
            warnings[0].contains("Did you mean 'Person'?"),
            "got: {}",
            warnings[0]
        );
    }

    #[test]
    fn warns_unknown_relationship_type_with_hint() {
        let g = graph_with_schema();
        let q = parse_cypher("MATCH (a:Person)-[:KNOWZ]->(b:Person) RETURN a").unwrap();
        let warnings = collect_unknown_pattern_warnings(&q, &g);
        assert_eq!(warnings.len(), 1, "got: {warnings:?}");
        assert!(warnings[0].contains("unknown relationship type 'KNOWZ'"));
        assert!(
            warnings[0].contains("Did you mean 'KNOWS'?"),
            "got: {}",
            warnings[0]
        );
    }

    #[test]
    fn mixed_alternation_warns_per_branch_without_the_no_rows_claim() {
        // `-[:MENTORS|KNOWS]->` matches through KNOWS, so the pattern *does*
        // return rows — the unqualified "returns no rows" wording was a false
        // claim about the query's result, not just an imprecise hint.
        let g = graph_with_schema();
        let q = parse_cypher("MATCH (a:Person)-[:MENTORS|KNOWS]->(b:Person) RETURN a").unwrap();
        let w = collect_unknown_pattern_warnings(&q, &g);
        assert_eq!(w.len(), 1, "got: {w:?}");
        assert!(w[0].contains("'MENTORS'"), "got: {}", w[0]);
        assert!(
            !w[0].contains("returns no rows"),
            "false no-rows claim on a mixed alternation: {}",
            w[0]
        );
        assert!(
            w[0].contains("'KNOWS'"),
            "the surviving branch must be named: {}",
            w[0]
        );
    }

    #[test]
    fn fully_unknown_alternation_keeps_the_no_rows_claim() {
        // Every branch unknown → the pattern really does return no rows.
        let g = graph_with_schema();
        let q = parse_cypher("MATCH (a:Person)-[:MENTORS|ADVISES]->(b:Person) RETURN a").unwrap();
        let w = collect_unknown_pattern_warnings(&q, &g);
        assert_eq!(w.len(), 2, "got: {w:?}");
        assert!(
            w.iter().all(|m| m.contains("returns no rows")),
            "got: {w:?}"
        );
    }

    #[test]
    fn no_warning_for_secondary_label() {
        // A label applied via add_label is valid in MATCH even though it is no
        // node's primary type — must NOT be flagged as unknown (regression
        // guard for a false positive: the warning once claimed `:Reviewer`
        // was unknown and "returns no rows" while it returned rows).
        let mut g = graph_with_schema();
        g.secondary_label_index
            .entry(InternedKey::from_str("Reviewer"))
            .or_default();
        let q = parse_cypher("MATCH (n:Reviewer) RETURN n").unwrap();
        assert!(collect_unknown_pattern_warnings(&q, &g).is_empty());
        let q2 = parse_cypher("MATCH (n:Reviewr) RETURN n").unwrap();
        assert_eq!(collect_unknown_pattern_warnings(&q2, &g).len(), 1);
    }

    #[test]
    fn no_warning_for_known_label_and_relationship() {
        let g = graph_with_schema();
        let q = parse_cypher("MATCH (a:Person)-[:KNOWS]->(b:Person) RETURN a").unwrap();
        assert!(collect_unknown_pattern_warnings(&q, &g).is_empty());
    }

    #[test]
    fn no_warning_on_schemaless_graph() {
        let g = DirGraph::new();
        let q = parse_cypher("MATCH (n:Anything) RETURN n").unwrap();
        assert!(collect_unknown_pattern_warnings(&q, &g).is_empty());
    }

    #[test]
    fn rejects_unknown_property_in_pattern_literal() {
        // The `{agee: 30}` form is unambiguous — validate.
        let g = graph_with_schema();
        let q = parse_cypher("MATCH (n:Person {agee: 30}) RETURN n").unwrap();
        let err = validate_schema(&q, &g).unwrap_err();
        assert_eq!(err.kind, SchemaErrorKind::UnknownProperty);
        assert!(err.message.contains("age"), "got: {}", err.message);
    }

    #[test]
    fn tolerates_unknown_property_in_where_expression() {
        // `n.prop` accesses in WHERE may reference virtual columns
        // (timeseries sub-nodes, computed aliases). Do not flag.
        let g = graph_with_schema();
        let q = parse_cypher("MATCH (n:Person) WHERE n.birth_yr = 1900 RETURN n").unwrap();
        assert!(validate_schema(&q, &g).is_ok());
    }

    #[test]
    fn tolerates_unknown_property_in_return_expression() {
        let g = graph_with_schema();
        let q = parse_cypher("MATCH (n:Person) RETURN n.agee").unwrap();
        assert!(validate_schema(&q, &g).is_ok());
    }

    #[test]
    fn allows_builtin_fields_in_pattern_literal() {
        let g = graph_with_schema();
        let q = parse_cypher("MATCH (n:Person {id: 1}) RETURN n.title").unwrap();
        assert!(validate_schema(&q, &g).is_ok());
    }

    #[test]
    fn skips_validation_on_empty_schema() {
        let g = DirGraph::new();
        let q = parse_cypher("MATCH (n:Anything) RETURN n.whatever").unwrap();
        assert!(validate_schema(&q, &g).is_ok());
    }

    #[test]
    fn validates_untyped_patterns_permissively() {
        // Untyped patterns (no :Label) are common and legal — we only
        // validate when the user has declared intent via a label.
        let g = graph_with_schema();
        let q = parse_cypher("MATCH (n) WHERE n.whatever = 1 RETURN n").unwrap();
        assert!(validate_schema(&q, &g).is_ok());
    }

    #[test]
    fn rejects_unknown_property_on_multi_hop_pattern_literal() {
        let g = graph_with_schema();
        let q = parse_cypher("MATCH (a:Person)-[:KNOWS]->(b:Person {agee: 30}) RETURN a").unwrap();
        let err = validate_schema(&q, &g).unwrap_err();
        assert_eq!(err.kind, SchemaErrorKind::UnknownProperty);
    }

    #[test]
    fn allows_order_by_and_return_of_known_properties() {
        let g = graph_with_schema();
        let q = parse_cypher("MATCH (n:Person) RETURN n.age ORDER BY n.email").unwrap();
        assert!(validate_schema(&q, &g).is_ok());
    }

    #[test]
    fn tolerates_unknown_label_in_where_label_check() {
        let g = graph_with_schema();
        let q = parse_cypher("MATCH (n) WHERE n:person RETURN n").unwrap();
        assert!(validate_schema(&q, &g).is_ok());
    }

    #[test]
    fn rejects_unknown_property_in_create_pattern_literal() {
        // Without this pre-flight check `CREATE (:Person {ttle: 'x'})` silently
        // stores a `ttle` property instead of `title`.
        let g = graph_with_schema();
        let q = parse_cypher("CREATE (:Person {agee: 30})").unwrap();
        let err = validate_schema(&q, &g).unwrap_err();
        assert_eq!(err.kind, SchemaErrorKind::UnknownProperty);
        assert!(err.message.contains("age"), "got: {}", err.message);
    }

    #[test]
    fn allows_known_property_in_create() {
        let g = graph_with_schema();
        let q = parse_cypher("CREATE (:Person {age: 30, email: 'a@b'})").unwrap();
        assert!(validate_schema(&q, &g).is_ok());
    }

    #[test]
    fn rejects_unknown_property_in_create_multi_element_path() {
        let g = graph_with_schema();
        let q =
            parse_cypher("CREATE (a:Person {age: 30})-[:KNOWS]->(b:Person {agee: 25}) RETURN a, b")
                .unwrap();
        let err = validate_schema(&q, &g).unwrap_err();
        assert_eq!(err.kind, SchemaErrorKind::UnknownProperty);
    }

    #[test]
    fn rejects_unknown_property_in_merge_pattern_literal() {
        let g = graph_with_schema();
        let q = parse_cypher("MERGE (n:Person {agee: 30}) RETURN n").unwrap();
        let err = validate_schema(&q, &g).unwrap_err();
        assert_eq!(err.kind, SchemaErrorKind::UnknownProperty);
    }

    #[test]
    fn allows_known_property_in_merge() {
        let g = graph_with_schema();
        let q = parse_cypher("MERGE (n:Person {email: 'a@b'}) RETURN n").unwrap();
        assert!(validate_schema(&q, &g).is_ok());
    }

    #[test]
    fn create_with_untyped_node_is_permissive() {
        // No label → no type metadata to check against; permissive, as MATCH is.
        let g = graph_with_schema();
        let q = parse_cypher("CREATE (n {anything_at_all: 1}) RETURN n").unwrap();
        assert!(validate_schema(&q, &g).is_ok());
    }

    #[test]
    fn create_on_unknown_node_type_is_permissive() {
        // Symmetric with `tolerates_unknown_node_type` for MATCH: only
        // validate when metadata is declared.
        let g = graph_with_schema();
        let q = parse_cypher("CREATE (n:NewType {whatever: 1}) RETURN n").unwrap();
        assert!(validate_schema(&q, &g).is_ok());
    }

    #[test]
    fn create_builtin_field_is_allowed() {
        let g = graph_with_schema();
        let q = parse_cypher("CREATE (n:Person {id: 99, title: 'Eve'}) RETURN n").unwrap();
        assert!(validate_schema(&q, &g).is_ok());
    }

    #[test]
    fn validates_property_inside_call_subquery_body() {
        // A pattern-literal typo inside a CALL { } body must be caught with
        // the same "did you mean?" quality as a top-level pattern.
        let g = graph_with_schema();
        let q = parse_cypher("CALL { MATCH (n:Person {agee: 1}) RETURN n.name AS nm } RETURN nm")
            .unwrap();
        let err = validate_schema(&q, &g).unwrap_err();
        assert_eq!(err.kind, SchemaErrorKind::UnknownProperty);
        assert!(err.message.contains("age"), "got: {}", err.message);
    }

    #[test]
    fn validates_labeled_pattern_literal_inside_correlated_call_body() {
        // A correlated body introducing a NEW labeled node with a property
        // typo is caught, same as a top-level labeled pattern literal.
        let g = graph_with_schema();
        let q = parse_cypher(
            "MATCH (p:Person) CALL { WITH p MATCH (p)-[:KNOWS]->(f:Person {agee: 1}) RETURN count(f) AS c } RETURN p.name, c",
        )
        .unwrap();
        let err = validate_schema(&q, &g).unwrap_err();
        assert_eq!(err.kind, SchemaErrorKind::UnknownProperty);
        assert!(err.message.contains("age"), "got: {}", err.message);
    }

    #[test]
    fn call_subquery_body_non_imported_var_is_fresh_scope() {
        // A bare body variable that shadows an outer name is a FRESH
        // variable (§1.2 rule 1) — without the import, the body's `n` has
        // no declared type, so a property reference is permissively allowed
        // (matches top-level untyped behaviour). No false positive.
        let g = graph_with_schema();
        let q = parse_cypher(
            "MATCH (p:Person) CALL { MATCH (n) RETURN n.anything AS a } RETURN p.name, a",
        )
        .unwrap();
        assert!(validate_schema(&q, &g).is_ok());
    }

    #[test]
    fn validates_property_inside_exists_nested_pattern() {
        let g = graph_with_schema();
        let q = parse_cypher(
            "MATCH (a:Person) WHERE EXISTS { MATCH (a)-[:KNOWS]->(b:Person {agee: 1}) } RETURN a",
        )
        .unwrap();
        let err = validate_schema(&q, &g).unwrap_err();
        assert_eq!(err.kind, SchemaErrorKind::UnknownProperty);
    }

    // ── Locked-schema label rejection ────────────────────────────────────
    //
    // Every clause that can carry a label gets a case — a fix covering only
    // MATCH would recreate the same asymmetry one level down.

    /// Adds an `Issue` type so `Isue` has a real near-miss to suggest —
    /// this is the scenario from the field report, verbatim.
    fn locked_graph() -> DirGraph {
        let mut g = graph_with_schema();
        let mut issue_props = HashMap::new();
        issue_props.insert("status".to_string(), "string".to_string());
        g.upsert_node_type_metadata("Issue", issue_props);
        g.schema_locked = true;
        g
    }

    /// The one assertion that matters: the offending label is named AND the
    /// valid set is enumerated, exactly like the unknown-property message.
    fn assert_rejects_isue(query: &str) {
        let g = locked_graph();
        let q = parse_cypher(query).unwrap();
        let err =
            validate_schema(&q, &g).expect_err(&format!("locked schema should reject: {query}"));
        assert_eq!(err.kind, SchemaErrorKind::UnknownNodeType, "for `{query}`");
        assert_eq!(
            err.message,
            "Unknown node type 'Isue'. Did you mean 'Issue'?\n  Valid types: Issue, Paper, Person",
            "for `{query}`"
        );
    }

    #[test]
    fn locked_schema_rejects_unknown_label_in_match() {
        assert_rejects_isue("MATCH (i:Isue) RETURN i");
    }

    #[test]
    fn locked_schema_rejects_unknown_label_in_optional_match() {
        assert_rejects_isue("MATCH (p:Person) OPTIONAL MATCH (i:Isue) RETURN p, i");
    }

    #[test]
    fn locked_schema_rejects_unknown_label_in_where_pattern_predicate() {
        assert_rejects_isue("MATCH (p:Person) WHERE EXISTS { MATCH (i:Isue) } RETURN p");
    }

    #[test]
    fn locked_schema_rejects_unknown_label_in_call_subquery() {
        assert_rejects_isue("CALL { MATCH (i:Isue) RETURN i } RETURN i");
    }

    #[test]
    fn locked_schema_rejects_unknown_label_in_union_branch() {
        assert_rejects_isue("MATCH (p:Person) RETURN p UNION MATCH (i:Isue) RETURN i");
    }

    #[test]
    fn locked_schema_rejects_unknown_extra_label() {
        // `(n:A:B)` puts the first label in `node_type` and the rest in
        // `extra_labels` — a typo in the tail must be caught too.
        let g = locked_graph();
        let q = parse_cypher("MATCH (n:Person:Revewer) RETURN n").unwrap();
        let err = validate_schema(&q, &g).expect_err("extra label typo should be rejected");
        assert_eq!(err.kind, SchemaErrorKind::UnknownNodeType);
        assert!(
            err.message.starts_with("Unknown node type 'Revewer'."),
            "got: {}",
            err.message
        );
    }

    #[test]
    fn locked_schema_offers_did_you_mean_for_near_miss() {
        let g = locked_graph();
        let q = parse_cypher("MATCH (n:Persn) RETURN n").unwrap();
        let err = validate_schema(&q, &g).expect_err("near-miss label should be rejected");
        assert_eq!(
            err.message,
            "Unknown node type 'Persn'. Did you mean 'Person'?\n  Valid types: Issue, Paper, Person"
        );
    }

    #[test]
    fn locked_schema_accepts_known_labels_everywhere() {
        let g = locked_graph();
        for query in [
            "MATCH (p:Person) RETURN p",
            "MATCH (p:Person) OPTIONAL MATCH (q:Paper) RETURN p, q",
            "MATCH (p:Person) WHERE EXISTS { MATCH (q:Paper) } RETURN p",
            "CALL { MATCH (q:Paper) RETURN q } RETURN q",
            "MATCH (p:Person) RETURN p UNION MATCH (q:Paper) RETURN q",
            // Untyped patterns carry no label to check.
            "MATCH (n) RETURN n",
        ] {
            let q = parse_cypher(query).unwrap();
            assert!(
                validate_schema(&q, &g).is_ok(),
                "locked schema wrongly rejected `{query}`: {:?}",
                validate_schema(&q, &g).err()
            );
        }
    }

    #[test]
    fn locked_schema_reports_the_label_before_the_property() {
        // When both are wrong, the label is the outer mistake — reporting a
        // property of a type that does not exist would be nonsense.
        let g = locked_graph();
        let q = parse_cypher("MATCH (i:Isue {titel: 'x'}) RETURN i").unwrap();
        let err = validate_schema(&q, &g).expect_err("should reject");
        assert_eq!(err.kind, SchemaErrorKind::UnknownNodeType);
    }

    #[test]
    fn open_schema_still_tolerates_unknown_label() {
        // The schemaless default is the product — the zero-row
        // existence-check idiom must keep working untouched. This is the
        // regression guard for the gate on `schema_locked`.
        let g = graph_with_schema();
        assert!(!g.schema_locked);
        for query in [
            "MATCH (i:Isue) RETURN i",
            "MATCH (p:Person) OPTIONAL MATCH (i:Isue) RETURN p, i",
            "MATCH (p:Person) WHERE EXISTS { MATCH (i:Isue) } RETURN p",
            "CALL { MATCH (i:Isue) RETURN i } RETURN i",
            "MATCH (p:Person) RETURN p UNION MATCH (i:Isue) RETURN i",
            "MATCH (n:Person:Revewer) RETURN n",
        ] {
            let q = parse_cypher(query).unwrap();
            assert!(
                validate_schema(&q, &g).is_ok(),
                "open schema must not reject `{query}`"
            );
        }
    }

    #[test]
    fn locked_schema_leaves_unknown_relationship_types_alone() {
        // Edge types are not part of what a schema lock covers on the read
        // path; only the node-label gap was asymmetric with properties.
        let g = locked_graph();
        let q = parse_cypher("MATCH (a:Person)-[:NOSUCH]->(b:Person) RETURN a").unwrap();
        assert!(validate_schema(&q, &g).is_ok());
    }

    #[test]
    fn locked_schema_accepts_secondary_labels() {
        // A label applied via `add_label` is legitimately matchable even
        // though it is no node's primary type — rejecting it would be a
        // false positive.
        let mut g = locked_graph();
        let key = g.interner.try_get_or_intern("Reviewer").unwrap();
        g.secondary_label_index.insert(key, Vec::new());
        g.has_secondary_labels = true;
        let q = parse_cypher("MATCH (n:Reviewer) RETURN n").unwrap();
        assert!(validate_schema(&q, &g).is_ok());
        // …and it appears in the enumerated valid set for a real typo.
        let q2 = parse_cypher("MATCH (n:Isue) RETURN n").unwrap();
        let err = validate_schema(&q2, &g).expect_err("should reject");
        assert_eq!(
            err.message,
            "Unknown node type 'Isue'. Did you mean 'Issue'?\n  Valid types: Issue, Paper, Person, Reviewer"
        );
    }

    // ── Open-schema warnings reach every nested clause ───────────────────
    //
    // Counterpart to the locked-schema block above: both surfaces share
    // [`walk_query_patterns`], so each clause form gets a case here too.

    /// Open (unlocked) schema with an `Issue` type, so `Isue` has a real
    /// near-miss to suggest — the locked block's scenario, unlocked.
    fn open_graph_with_issue() -> DirGraph {
        let mut g = graph_with_schema();
        let mut issue_props = HashMap::new();
        issue_props.insert("status".to_string(), "string".to_string());
        g.upsert_node_type_metadata("Issue", issue_props);
        assert!(!g.schema_locked, "this block is about the open-schema path");
        g
    }

    /// Exactly one warning, naming the typo'd label and suggesting the real
    /// one — and still no error, because a zero-row read stays legal.
    fn assert_warns_isue(query: &str) {
        let g = open_graph_with_issue();
        let q = parse_cypher(query).unwrap();
        let warnings = collect_unknown_pattern_warnings(&q, &g);
        assert_eq!(warnings.len(), 1, "for `{query}`, got: {warnings:?}");
        assert_eq!(
            warnings[0],
            "MATCH references unknown node label 'Isue' — the graph has no such type, so this \
             pattern returns no rows. Did you mean 'Issue'?",
            "for `{query}`"
        );
        assert!(
            validate_schema(&q, &g).is_ok(),
            "warning must stay non-fatal on an open schema: `{query}`"
        );
    }

    #[test]
    fn warns_unknown_label_in_top_level_match() {
        // Control: the surface that already worked must keep working, with
        // the identical message the nested cases assert.
        assert_warns_isue("MATCH (i:Isue) RETURN i");
        assert_warns_isue("MATCH (p:Person) OPTIONAL MATCH (i:Isue) RETURN p, i");
    }

    #[test]
    fn warns_unknown_label_in_call_subquery() {
        assert_warns_isue("CALL { MATCH (i:Isue) RETURN i } RETURN i");
    }

    #[test]
    fn warns_unknown_label_in_where_exists() {
        assert_warns_isue("MATCH (p:Person) WHERE EXISTS { MATCH (i:Isue) } RETURN p");
        // …and behind boolean nesting, which the predicate walk descends.
        assert_warns_isue(
            "MATCH (p:Person) WHERE p.age > 1 AND NOT EXISTS { MATCH (i:Isue) } RETURN p",
        );
        // …and on a WITH's WHERE, not just a standalone one.
        assert_warns_isue("MATCH (p:Person) WITH p WHERE EXISTS { MATCH (i:Isue) } RETURN p");
    }

    #[test]
    fn warns_unknown_label_in_union_branch() {
        assert_warns_isue("MATCH (p:Person) RETURN p UNION MATCH (i:Isue) RETURN i");
    }

    #[test]
    fn warns_unknown_label_nested_two_deep() {
        // Nesting composes: an EXISTS inside a CALL body is reached by the
        // same recursion, so no clause form needs its own special case.
        assert_warns_isue(
            "CALL { MATCH (p:Person) WHERE EXISTS { MATCH (i:Isue) } RETURN p } RETURN p",
        );
    }

    #[test]
    fn warns_unknown_relationship_type_in_nested_clauses() {
        // The relationship half of the same walk — one warning per query,
        // wherever the pattern sits.
        let g = open_graph_with_issue();
        for query in [
            "CALL { MATCH (a:Person)-[:KNOWZ]->(b:Person) RETURN a } RETURN a",
            "MATCH (p:Person) WHERE EXISTS { MATCH (a:Person)-[:KNOWZ]->(b:Person) } RETURN p",
            "MATCH (p:Person) RETURN p UNION MATCH (a:Person)-[:KNOWZ]->(b:Person) RETURN a",
        ] {
            let q = parse_cypher(query).unwrap();
            let warnings = collect_unknown_pattern_warnings(&q, &g);
            assert_eq!(warnings.len(), 1, "for `{query}`, got: {warnings:?}");
            assert!(
                warnings[0].contains("unknown relationship type 'KNOWZ'")
                    && warnings[0].contains("Did you mean 'KNOWS'?"),
                "for `{query}`, got: {}",
                warnings[0]
            );
        }
    }

    #[test]
    fn no_warning_for_known_labels_in_nested_clauses() {
        // The other half of the coverage: descending must not invent
        // warnings for valid nested patterns.
        let g = open_graph_with_issue();
        for query in [
            "CALL { MATCH (i:Issue) RETURN i } RETURN i",
            "MATCH (p:Person) WHERE EXISTS { MATCH (i:Issue) } RETURN p",
            "MATCH (p:Person) RETURN p UNION MATCH (i:Issue) RETURN i",
            "CALL { MATCH (a:Person)-[:KNOWS]->(b:Person) RETURN a } RETURN a",
        ] {
            let q = parse_cypher(query).unwrap();
            assert!(
                collect_unknown_pattern_warnings(&q, &g).is_empty(),
                "spurious warning for `{query}`: {:?}",
                collect_unknown_pattern_warnings(&q, &g)
            );
        }
    }

    #[test]
    fn no_warning_for_new_label_in_nested_write() {
        // Write patterns are how a type comes into existence on an open
        // schema — warning "the graph has no such type" for a CREATE would
        // be wrong. The shared walk visits them (the fatal check needs their
        // property names), so the warning collector must skip them by kind.
        // (A write inside `CALL { }` is rejected by the parser, so top-level
        // CREATE / MERGE are the reachable cases.)
        let g = open_graph_with_issue();
        for query in [
            "CREATE (i:Ticket {title: 'x'})",
            "MERGE (i:Ticket {title: 'x'})",
            "MATCH (p:Person) MERGE (p)-[:KNOWZ]->(t:Ticket {title: 'x'})",
        ] {
            let q = parse_cypher(query).unwrap();
            assert!(
                collect_unknown_pattern_warnings(&q, &g).is_empty(),
                "write pattern wrongly warned for `{query}`: {:?}",
                collect_unknown_pattern_warnings(&q, &g)
            );
        }
    }
}