kglite 0.18.0

use crate::datatypes::values::Value;
use crate::graph::core::membership::MembershipSet;
use crate::graph::schema::InternedKey;
use petgraph::graph::{EdgeIndex, NodeIndex};
use std::collections::HashMap;

#[derive(Debug, Clone)]
pub struct Pattern {
    pub elements: Vec<PatternElement>,
}

#[derive(Debug, Clone)]
pub enum PatternElement {
    Node(NodePattern),
    Edge(EdgePattern),
}

/// Pattern for matching nodes: (var:Type {prop: value})
///
/// `node_type` holds the primary (first) label — every existing
/// consumer reads through this field. `extra_labels` holds any
/// additional labels from Cypher-standard `(n:A:B:C)` syntax; the
/// executor AND-intersects them against `node_type` candidates.
/// Empty for single-label patterns (the hot path).
#[derive(Debug, Clone)]
pub struct NodePattern {
    pub variable: Option<String>,
    /// The first (or only) label. Under alternation this is `alt_labels[0]`
    /// — a consumer that reads it alone silently narrows the pattern to one
    /// branch, the exact wrong-answer class `EdgePattern::conn_filter`'s doc
    /// records four separate bugs for on the edge side. Route through
    /// [`NodePattern::label_alternatives`] /
    /// [`NodePattern::multi_label_constrained`].
    pub node_type: Option<String>,
    /// `:A:B` AND-chain extras. Empty whenever `alt_labels` is set —
    /// alternation and conjunction are mutually exclusive in v1 (mixing is
    /// a parse error, so no precedence is silently committed to).
    pub extra_labels: Vec<String>,
    /// `:A|B|C` alternation, every branch including the first. `None` for
    /// plain patterns.
    pub alt_labels: Option<Vec<String>>,
    pub properties: Option<HashMap<String, PropertyMatcher>>,
    /// Label slots written as a parameter (`(n:$label)`), pending
    /// [`crate::graph::languages::cypher::dynamic_labels::resolve`]. See
    /// [`ParamLabel`] — empty for every literal pattern, and empty for every
    /// pattern that reaches the planner.
    pub label_params: Vec<ParamLabel>,
}

impl NodePattern {
    /// All labels a node may match through: the alternation branches, or
    /// the single `node_type`. Empty slice for untyped patterns.
    pub fn label_alternatives(&self) -> &[String] {
        match &self.alt_labels {
            Some(alts) => alts.as_slice(),
            None => self.node_type.as_slice(),
        }
    }

    /// True when the pattern constrains labels beyond one primary-or-single
    /// label — an AND-chain (`extra_labels`) or an alternation. The single
    /// guard consumers that only understand the plain shape must use
    /// (replacing the old `!extra_labels.is_empty()` idiom, which alternation
    /// would silently walk past).
    pub fn multi_label_constrained(&self) -> bool {
        !self.extra_labels.is_empty() || self.alt_labels.is_some()
    }
}

/// Pattern for matching edges: -[:TYPE {prop: value}]->
///
/// `*min..max` fills `var_length`: `*` means 1 hop to the default max, `*..5`
/// 1 to 5, `*2` exactly 2, `*1..3` a closed range, `*2..` 2 to the default max.
#[derive(Debug, Clone)]
pub struct EdgePattern {
    pub variable: Option<String>,
    pub connection_type: Option<String>,
    /// Alternation from pipe syntax `[:A|B|C]`; an edge matches ANY of them.
    /// `connection_type` holds only the first — always read both through
    /// [`EdgePattern::conn_filter`].
    pub connection_types: Option<Vec<String>>,
    pub direction: EdgeDirection,
    pub properties: Option<HashMap<String, PropertyMatcher>>,
    /// `(min_hops, max_hops)`; `None` means exactly one hop.
    pub var_length: Option<(usize, usize)>,
    /// Whether `var_length`'s maximum was written (`*..5`, `*1..3`, `*2`)
    /// rather than defaulted to the var-length cap by an open form (`*`,
    /// `*N..`). A var-length MATCH applies the cap either way; `shortestPath`
    /// reads this to treat an open form as unbounded.
    pub var_length_max_written: bool,
    /// Whether the matcher must retain path identity. When false,
    /// variable-length expansion may use global BFS dedup and fixed-length
    /// expansion omits its exact-trail allocation. Set false only when the
    /// planner proves the surrounding query does not consume path identity.
    pub needs_path_info: bool,
    /// When true, the connection type metadata guarantees the target node
    /// matches the pattern's type, so the node_weight() lookup can be skipped.
    /// Set by the query planner when connection_type_metadata confirms a single
    /// target type (outgoing) or source type (incoming).
    pub skip_target_type_check: bool,
    /// Inline filter pushed from a downstream `WHERE` that references only
    /// the edge variable (and the edge's structural peer in this pattern),
    /// applied in the matcher hot loop before a row is materialized.
    /// Populated by the Cypher planner's
    /// [`super::super::languages::cypher::planner::rel_predicate_pushdown`]
    /// pass; `None` for hand-built patterns.
    pub edge_filter: Option<RelEdgeFilter>,
    /// Relationship-type slots written as a parameter (`-[:$type]->`),
    /// pending resolution. See [`ParamLabel`]; slot `i` indexes
    /// `connection_types[i]`, and slot 0 additionally covers
    /// `connection_type`.
    pub type_params: Vec<ParamLabel>,
}

/// A label or relationship-type slot whose text comes from a query parameter
/// (`(n:$label)`, `(n:$(label))`, `-[:$type]->`).
///
/// **This exists only between the parser and the resolver.** Parsed ASTs are
/// cached by query *text* and re-run with different parameters, so the parser
/// records the reference here and parks the source spelling (`"$label"`) in
/// the string slot; `cypher::dynamic_labels::resolve` writes the bound name
/// into that slot and clears this list before validation, planning and
/// execution, leaving every downstream consumer only the literal form.
///
/// The marker is deliberately **out of band** rather than a sentinel spelling
/// inside the string: no literal label, however written (`` `$label` ``
/// included), can then be mistaken for a parameter reference, and no
/// parameter *value* can ever be re-read as a reference.
///
/// A slot that somehow reached the planner unresolved names a type spelled
/// `$label`, which matches nothing — it under-returns, never over-returns.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ParamLabel {
    /// Which slot the parameter fills. For [`NodePattern`], 0 is `node_type`
    /// and `n > 0` is `extra_labels[n - 1]`; for [`EdgePattern`], `n` is
    /// `connection_types[n]`.
    pub slot: usize,
    /// Parameter name, without the `$`.
    pub param: String,
}

impl ParamLabel {
    /// The source spelling parked in the string slot until resolution.
    pub fn placeholder(param: &str) -> String {
        format!("${param}")
    }
}

impl EdgePattern {
    /// The set of connection types this pattern accepts, interned once.
    ///
    /// Always build the filter through this constructor rather than
    /// reading [`EdgePattern::connection_type`] directly: for `[:A|B]`
    /// the singular field holds only `A`, so a consumer that ignores
    /// `connection_types` silently narrows the pattern to its first
    /// branch. That narrowing produced wrong *answers* (not just slow
    /// plans) in the fused counters, the anchored-count fusion, the
    /// EXISTS fast path and the `skip_target_type_check` annotation.
    pub fn conn_filter(&self) -> ConnTypeFilter {
        match (&self.connection_types, &self.connection_type) {
            // `[:A|B]` — the parser does not deduplicate, and a counting
            // consumer that sums per type would double-count `[:A|A]`,
            // so dedup here where every consumer inherits it.
            (Some(types), _) if !types.is_empty() => {
                let mut keys: Vec<InternedKey> = Vec::with_capacity(types.len());
                for ty in types {
                    let key = InternedKey::from_str(ty);
                    if !keys.contains(&key) {
                        keys.push(key);
                    }
                }
                match keys.len() {
                    1 => ConnTypeFilter::One(keys[0]),
                    _ => ConnTypeFilter::AnyOf(keys),
                }
            }
            (_, Some(ty)) => ConnTypeFilter::One(InternedKey::from_str(ty)),
            _ => ConnTypeFilter::Any,
        }
    }
}

/// Which connection types an [`EdgePattern`] accepts. Only the alternation
/// case allocates, so the hot single-type path stays allocation-free.
#[derive(Debug, Clone)]
pub enum ConnTypeFilter {
    /// Untyped edge (`-[]->`) — every connection type matches.
    Any,
    One(InternedKey),
    /// Alternation (`[:A|B|C]`); deduplicated, always ≥ 2 entries.
    AnyOf(Vec<InternedKey>),
}

impl ConnTypeFilter {
    /// Pre-filter hint for `edges_directed_filtered` / `iter_peers_filtered`.
    ///
    /// `Some` only when exactly one type is accepted — a backend that
    /// pre-filters (the disk CSR) can then skip non-matching edges
    /// entirely. For an alternation this is `None` and the caller MUST
    /// post-filter with [`Self::accepts`]; the memory and mapped backends
    /// treat the argument as a hint and return every edge regardless, so
    /// the post-filter is required in all cases anyway.
    #[inline]
    pub fn hint(&self) -> Option<InternedKey> {
        match self {
            ConnTypeFilter::One(key) => Some(*key),
            _ => None,
        }
    }

    #[inline]
    pub fn accepts(&self, key: InternedKey) -> bool {
        match self {
            ConnTypeFilter::Any => true,
            ConnTypeFilter::One(want) => key == *want,
            ConnTypeFilter::AnyOf(keys) => keys.contains(&key),
        }
    }

    #[inline]
    pub fn is_any(&self) -> bool {
        matches!(self, ConnTypeFilter::Any)
    }

    /// Sum a per-type count over every accepted type: `f` is called once
    /// with `None` (count all types) for an untyped edge, otherwise once
    /// per accepted type. Summing is exact because each edge carries
    /// exactly one connection type and the key list is deduplicated.
    pub fn try_fold_counts<E>(
        &self,
        mut f: impl FnMut(Option<InternedKey>) -> Result<usize, E>,
    ) -> Result<usize, E> {
        match self {
            ConnTypeFilter::Any => f(None),
            ConnTypeFilter::One(key) => f(Some(*key)),
            ConnTypeFilter::AnyOf(keys) => {
                let mut total = 0usize;
                for key in keys {
                    total = total.saturating_add(f(Some(*key))?);
                }
                Ok(total)
            }
        }
    }
}

/// Inline edge filter — evaluated during expansion to skip edges the
/// downstream `WHERE` would have discarded.
#[derive(Debug, Clone)]
pub struct RelEdgeFilter {
    pub predicate: RelEdgePredicate,
    /// Which endpoint the matcher expands from; see [`AnchorSide`].
    pub anchor: AnchorSide,
}

/// Which pattern endpoint the matcher is treating as the anchor when
/// expanding edges. The planner records this when compiling
/// startNode/endNode predicates so the matcher can answer those at
/// runtime against `direction`.
///
/// `Target` is currently unused — the rel-pushdown pass only emits
/// `Source`-anchored filters because the planner reverses pattern
/// direction elsewhere (`reorder_match_patterns`) before pushdown
/// runs. Kept around so a future planner pass that emits filters
/// after-reversal has a name for "the right endpoint is the anchor."
#[allow(dead_code)]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum AnchorSide {
    /// Anchor is the left endpoint of the pattern (typical case —
    /// `(anchor)-[r]-(peer)`). Matcher's `direction == Outgoing`
    /// means the edge goes anchor→peer; `Incoming` means peer→anchor.
    Source,
    /// Anchor is the right endpoint of the pattern (used when the
    /// planner reverses pattern direction for selectivity). Matcher's
    /// `direction == Outgoing` then means peer→anchor in the original
    /// pattern's frame of reference.
    Target,
}

/// Compiled per-edge predicate used by [`RelEdgeFilter`], evaluated in the
/// matcher hot loop by [`RelEdgePredicate::eval`]. A predicate the planner
/// cannot compile into this enum stays in the `WHERE` clause and is
/// evaluated by the materialized predicate evaluator instead.
///
/// `StartNodeIs` / `EndNodeIs` (with a bound `NodeIndex`) are
/// reserved for a future pushdown that sees `startNode(r) = $param`
/// or `startNode(r) = priorVar` after parameter / pre-binding
/// resolution. The current pass only emits the peer-relative variants.
#[allow(dead_code)]
#[derive(Debug, Clone)]
pub enum RelEdgePredicate {
    /// Always-true sentinel — used as the identity for And/Or
    /// simplification at compile time.
    True,
    False,
    /// `type(r)` is one of these connection types.
    TypeIn(Vec<InternedKey>),
    /// `r.<prop> OP <value>` for the supplied operator.
    Property {
        prop: String,
        op: PropOp,
        value: Value,
    },
    /// `startNode(r) = <peer endpoint>` / `endNode(r) = <peer
    /// endpoint>`. Encoded as a direction equality in the matcher's
    /// frame of reference: `StartNodeIsPeer` is true iff the edge's
    /// source equals the pattern's peer, which (for a `Source`-anchored
    /// pattern) maps to `direction == Incoming`.
    StartNodeIsPeer,
    EndNodeIsPeer,
    /// `r.<endpoint>(...) = <bound NodeIndex>` — startNode/endNode
    /// compared against a node bound elsewhere (pre-bindings or a
    /// prior-clause variable). The matcher checks the resolved
    /// NodeIndex against the edge's source/target.
    StartNodeIs(NodeIndex),
    EndNodeIs(NodeIndex),
    And(Vec<RelEdgePredicate>),
    Or(Vec<RelEdgePredicate>),
    Not(Box<RelEdgePredicate>),
}

/// Comparison operator for [`RelEdgePredicate::Property`].
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum PropOp {
    Eq,
    Ne,
    Gt,
    Ge,
    Lt,
    Le,
    StartsWith,
    Contains,
    EndsWith,
}

impl RelEdgePredicate {
    /// Evaluate the predicate for a single edge during expansion.
    ///
    /// `connection_type` and `get_prop` expose the edge body without
    /// materializing an `EdgeData`; `peer_is_start` is the matcher's per-edge
    /// direction resolved once against the stored [`AnchorSide`].
    ///
    /// Returns `true` only for Cypher `true` — both `false` and `null` reject
    /// the edge in a WHERE context.
    #[inline]
    pub fn eval(
        &self,
        connection_type: InternedKey,
        peer_is_start: bool,
        edge_source: NodeIndex,
        edge_target: NodeIndex,
        get_prop: &impl Fn(&str) -> Option<Value>,
    ) -> bool {
        self.eval_nullable(
            connection_type,
            peer_is_start,
            edge_source,
            edge_target,
            get_prop,
        ) == Some(true)
    }

    /// Evaluate with Cypher's three-valued boolean semantics.
    ///
    /// `None` represents `null`/unknown. Keeping this state through boolean
    /// composition is essential: `NOT null` is still `null`, while
    /// `false AND null` is `false` and `true OR null` is `true`.
    #[inline]
    fn eval_nullable(
        &self,
        connection_type: InternedKey,
        peer_is_start: bool,
        edge_source: NodeIndex,
        edge_target: NodeIndex,
        get_prop: &impl Fn(&str) -> Option<Value>,
    ) -> Option<bool> {
        match self {
            RelEdgePredicate::True => Some(true),
            RelEdgePredicate::False => Some(false),
            RelEdgePredicate::TypeIn(types) => Some(types.contains(&connection_type)),
            RelEdgePredicate::Property { prop, op, value } => match get_prop(prop) {
                Some(Value::Null) | None => None,
                Some(v) => Some(match op {
                    PropOp::Eq => {
                        return crate::graph::core::filtering::predicate_values_equal(&v, value)
                    }
                    PropOp::Ne => {
                        return crate::graph::core::filtering::predicate_values_equal(&v, value)
                            .map(|v| !v)
                    }
                    // Three-valued, through the same helper the executor's
                    // `evaluate_comparison_tristate` uses: an edge property a
                    // literal has no ordering rule against is `null`, so
                    // `WHERE NOT (r.w < 1)` drops it exactly as
                    // `WHERE NOT (n.w < 1)` drops the node.
                    PropOp::Gt => {
                        return crate::graph::core::filtering::ordering_matches(
                            &v,
                            value,
                            |ordering| ordering == std::cmp::Ordering::Greater,
                        )
                    }
                    PropOp::Ge => {
                        return crate::graph::core::filtering::ordering_matches(
                            &v,
                            value,
                            |ordering| ordering != std::cmp::Ordering::Less,
                        )
                    }
                    PropOp::Lt => {
                        return crate::graph::core::filtering::ordering_matches(
                            &v,
                            value,
                            |ordering| ordering == std::cmp::Ordering::Less,
                        )
                    }
                    PropOp::Le => {
                        return crate::graph::core::filtering::ordering_matches(
                            &v,
                            value,
                            |ordering| ordering != std::cmp::Ordering::Greater,
                        )
                    }
                    PropOp::StartsWith => matches!(
                        (&v, value),
                        (Value::String(text), Value::String(prefix)) if text.starts_with(prefix)
                    ),
                    PropOp::Contains => matches!(
                        (&v, value),
                        (Value::String(text), Value::String(needle)) if text.contains(needle)
                    ),
                    PropOp::EndsWith => matches!(
                        (&v, value),
                        (Value::String(text), Value::String(suffix)) if text.ends_with(suffix)
                    ),
                }),
            },
            RelEdgePredicate::StartNodeIsPeer => Some(edge_source == edge_target || peer_is_start),
            RelEdgePredicate::EndNodeIsPeer => Some(edge_source == edge_target || !peer_is_start),
            RelEdgePredicate::StartNodeIs(idx) => Some(edge_source == *idx),
            RelEdgePredicate::EndNodeIs(idx) => Some(edge_target == *idx),
            RelEdgePredicate::And(items) => {
                let mut saw_unknown = false;
                for predicate in items {
                    match predicate.eval_nullable(
                        connection_type,
                        peer_is_start,
                        edge_source,
                        edge_target,
                        get_prop,
                    ) {
                        Some(false) => return Some(false),
                        None => saw_unknown = true,
                        Some(true) => {}
                    }
                }
                if saw_unknown {
                    None
                } else {
                    Some(true)
                }
            }
            RelEdgePredicate::Or(items) => {
                let mut saw_unknown = false;
                for predicate in items {
                    match predicate.eval_nullable(
                        connection_type,
                        peer_is_start,
                        edge_source,
                        edge_target,
                        get_prop,
                    ) {
                        Some(true) => return Some(true),
                        None => saw_unknown = true,
                        Some(false) => {}
                    }
                }
                if saw_unknown {
                    None
                } else {
                    Some(false)
                }
            }
            RelEdgePredicate::Not(inner) => inner
                .eval_nullable(
                    connection_type,
                    peer_is_start,
                    edge_source,
                    edge_target,
                    get_prop,
                )
                .map(|value| !value),
        }
    }
}

#[derive(Debug, Clone, Copy, PartialEq)]
pub enum EdgeDirection {
    Outgoing, // -[]->
    Incoming, // <-[]-
    Both,     // -[]-
}

#[derive(Debug, Clone)]
pub enum PropertyMatcher {
    Equals(Value),
    /// Deferred parameter resolution: matched at execution time from params map
    EqualsParam(String),
    /// Deferred variable resolution: resolved against projected row values
    /// from WITH/UNWIND before pattern matching. Example:
    /// `WITH "Oslo" AS city MATCH (n:Person {city: city})`
    EqualsVar(String),
    /// Deferred node-property resolution: resolved against an already-bound
    /// node's property at row-execute time. Pushed by the planner from a
    /// correlated `WHERE cur.prop = prior.other_prop` so the pattern executor
    /// can pick an indexed lookup when `(cur_type, prop)` is indexed.
    EqualsNodeProp {
        var: String,
        prop: String,
    },
    /// Deferred expression equality: an inline-map value written as a full
    /// Cypher expression (`MATCH (d {id: row[0]})`), the value grammar a
    /// `CREATE` map accepts. The executor evaluates it against each row, as
    /// it does [`Self::EqualsVar`]; a NULL result matches nothing.
    EqualsExpr(Box<crate::graph::languages::cypher::ast::Expression>),
    /// Pushed from `WHERE n.prop IN [v1, v2, ...]` by the planner.
    ///
    /// A [`MembershipSet`] rather than a bare `Vec<Value>`: the list is
    /// probed once per candidate node, so the coercion-normalized index is
    /// built once here instead of re-scanning the list per row. Derefs to
    /// `&[Value]`, so planner code that inspects the values is unaffected.
    In(MembershipSet),
    /// Comparison matchers: pushed from `WHERE n.prop > val` etc. by the planner.
    /// Enables filter pushdown into MATCH and range index acceleration.
    GreaterThan(Value),
    GreaterOrEqual(Value),
    LessThan(Value),
    LessOrEqual(Value),
    /// Combined range: both a lower and upper bound on the same property.
    /// Used when WHERE has e.g. `n.year >= 2015 AND n.year <= 2022`.
    /// Booleans indicate inclusive (true) vs exclusive (false).
    Range {
        lower: Value,
        lower_inclusive: bool,
        upper: Value,
        upper_inclusive: bool,
    },
    /// String prefix matcher — pushed from `WHERE n.prop STARTS WITH 'X'`.
    /// Enables persistent prefix index acceleration on disk graphs.
    StartsWith(String),
    /// String substring matcher — evaluated while discovering node candidates.
    Contains(String),
    /// String suffix matcher — evaluated while discovering node candidates.
    EndsWith(String),
}

/// A single pattern match with variable bindings.
/// Uses Vec instead of HashMap — patterns add 1-6 unique variables, so
/// linear search beats hashing and the bindings sit in one allocation.
#[derive(Debug, Clone)]
pub struct PatternMatch {
    pub bindings: Vec<(String, MatchBinding)>,
    /// Exact fixed-length trail, stored outside named bindings so anonymous
    /// hops do not allocate synthetic variable names. Boxed so matches whose
    /// planner proves that trail tracking is unnecessary carry only one
    /// pointer of overhead rather than an inline `Vec` payload.
    #[doc(hidden)]
    pub exact_path: Option<Box<(NodeIndex, Vec<PathHop>)>>,
}

/// One exact hop in a matched path.
///
/// Node identity alone is insufficient because a graph may contain parallel
/// relationships between the same endpoints.  Keeping the edge slot here also
/// lets path consumers preserve relationship properties and orientation.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct PathHop {
    pub node: NodeIndex,
    pub edge: EdgeIndex,
    pub connection_type: InternedKey,
}

#[derive(Debug, Clone)]
pub enum MatchBinding {
    Node {
        index: NodeIndex,
        node_type: String,
        title: String,
        id: Value,
        properties: HashMap<String, Value>,
    },
    /// Lightweight node reference — stores only NodeIndex (4 bytes).
    /// Used in Cypher executor path where node data is resolved on demand from graph.
    NodeRef(NodeIndex),
    /// Relationship binding. Deliberately index-only: consumers resolve
    /// edge properties on demand from the graph via `edge_index` (the
    /// Cypher executor's `EdgeBinding` and the fluent API's dict
    /// conversion both do), so the matcher never clones a property map
    /// into the binding on the expansion hot path.
    Edge {
        source: NodeIndex,
        target: NodeIndex,
        edge_index: EdgeIndex,
        connection_type: InternedKey,
    },
    /// Variable-length path binding for patterns like -[:TYPE*1..3]->
    VariableLengthPath {
        source: NodeIndex,
        target: NodeIndex,
        hops: usize,
        /// Exact hops, excluding `source` and including `target`.
        path: Vec<PathHop>,
    },
}

#[cfg(test)]
mod tests {
    use super::*;

    fn property(op: PropOp) -> RelEdgePredicate {
        RelEdgePredicate::Property {
            prop: "tag".to_string(),
            op,
            value: Value::String("foo".to_string()),
        }
    }

    fn eval_with(predicate: &RelEdgePredicate, get_prop: &impl Fn(&str) -> Option<Value>) -> bool {
        predicate.eval(
            InternedKey::from_str("R"),
            true,
            NodeIndex::new(0),
            NodeIndex::new(1),
            get_prop,
        )
    }

    #[test]
    fn relationship_not_preserves_missing_property_unknown() {
        for op in [PropOp::Eq, PropOp::Ne] {
            let negated = RelEdgePredicate::Not(Box::new(property(op)));
            assert!(!eval_with(&negated, &|_| None));
            assert!(!eval_with(&negated, &|_| Some(Value::Null)));
        }
    }

    #[test]
    fn relationship_nested_null_survives_negation() {
        for op in [PropOp::Eq, PropOp::Ne] {
            let predicate = RelEdgePredicate::Property {
                prop: "tag".into(),
                op,
                value: Value::List(vec![Value::Int64(1)]),
            };
            let read = |_: &str| Some(Value::List(vec![Value::Null]));
            assert_eq!(
                predicate.eval_nullable(
                    InternedKey::from_str("R"),
                    true,
                    NodeIndex::new(0),
                    NodeIndex::new(1),
                    &read
                ),
                None
            );
            assert!(!eval_with(&predicate, &read));
            assert!(!eval_with(
                &RelEdgePredicate::Not(Box::new(predicate)),
                &read
            ));
        }
    }

    #[test]
    fn self_loop_peer_is_both_relationship_endpoints() {
        for predicate in [
            RelEdgePredicate::StartNodeIsPeer,
            RelEdgePredicate::EndNodeIsPeer,
        ] {
            for peer_is_start in [false, true] {
                assert_eq!(
                    predicate.eval_nullable(
                        InternedKey::from_str("R"),
                        peer_is_start,
                        NodeIndex::new(0),
                        NodeIndex::new(0),
                        &|_| None
                    ),
                    Some(true)
                );
            }
        }
    }

    #[test]
    fn relationship_boolean_composition_uses_kleene_logic() {
        assert!(!eval_with(
            &RelEdgePredicate::And(vec![RelEdgePredicate::False, property(PropOp::Eq),]),
            &|_| None,
        ));
        assert!(eval_with(
            &RelEdgePredicate::Or(vec![RelEdgePredicate::True, property(PropOp::Eq),]),
            &|_| None,
        ));
        assert!(!eval_with(
            &RelEdgePredicate::Not(Box::new(RelEdgePredicate::Or(vec![
                RelEdgePredicate::False,
                property(PropOp::Eq),
            ]))),
            &|_| None,
        ));
        assert!(!eval_with(
            &RelEdgePredicate::And(vec![RelEdgePredicate::False, property(PropOp::Eq),]),
            &|_| Some(Value::Null)
        ));
    }

    /// A relationship-property ordering comparison must answer `null` for a
    /// pair of types no ordering rule relates, exactly as the executor's
    /// `evaluate_comparison_tristate` does. When this collapsed to `false`,
    /// `WHERE NOT (r.w < 1)` kept string-valued edges that
    /// `WHERE NOT (n.w < 1)` dropped.
    #[test]
    fn relationship_ordering_without_a_rule_is_null() {
        for op in [PropOp::Gt, PropOp::Ge, PropOp::Lt, PropOp::Le] {
            for (stored, literal) in [
                (Value::String("foo".to_string()), Value::Int64(1)),
                (Value::Int64(1), Value::String("foo".to_string())),
                (Value::Boolean(true), Value::Int64(1)),
                (Value::List(vec![Value::Int64(1)]), Value::Int64(2)),
            ] {
                let predicate = RelEdgePredicate::Property {
                    prop: "w".to_string(),
                    op,
                    value: literal,
                };
                let read = |_: &str| Some(stored.clone());
                assert_eq!(
                    predicate.eval_nullable(
                        InternedKey::from_str("R"),
                        true,
                        NodeIndex::new(0),
                        NodeIndex::new(1),
                        &read
                    ),
                    None,
                    "{stored:?} {op:?} must be null"
                );
                // Unknown rejects the edge in both the positive and the
                // negated form — that is what `null` buys over `false`.
                assert!(!eval_with(&predicate, &read));
                assert!(!eval_with(
                    &RelEdgePredicate::Not(Box::new(predicate)),
                    &read
                ));
            }
        }
    }

    /// NaN is a number, not an incomparable type: every ordering comparison
    /// against it is `false`, so `NOT (r.w < 1)` keeps the edge.
    #[test]
    fn relationship_ordering_against_nan_is_false_not_null() {
        for op in [PropOp::Gt, PropOp::Ge, PropOp::Lt, PropOp::Le] {
            let predicate = RelEdgePredicate::Property {
                prop: "w".to_string(),
                op,
                value: Value::Int64(1),
            };
            let read = |_: &str| Some(Value::Float64(f64::NAN));
            assert_eq!(
                predicate.eval_nullable(
                    InternedKey::from_str("R"),
                    true,
                    NodeIndex::new(0),
                    NodeIndex::new(1),
                    &read
                ),
                Some(false),
                "NaN {op:?} 1 must be false"
            );
            assert!(eval_with(
                &RelEdgePredicate::Not(Box::new(predicate)),
                &read
            ));
        }
    }

    #[test]
    fn relationship_text_predicates_match_strings_and_reject_nulls() {
        for (op, text) in [
            (PropOp::StartsWith, "foobar"),
            (PropOp::Contains, "xfooy"),
            (PropOp::EndsWith, "barfoo"),
        ] {
            let predicate = property(op);
            assert!(eval_with(&predicate, &|_| Some(Value::String(
                text.to_string()
            ))));
            assert!(!eval_with(&predicate, &|_| None));
            assert!(!eval_with(&predicate, &|_| Some(Value::Null)));
            assert!(!eval_with(&predicate, &|_| Some(Value::Int64(7))));
        }
    }
}