kglite 0.13.2

// Pattern AST types for graph pattern matching.
//
// Supports patterns like: (p:Play)-[:HAS_PROSPECT]->(pr:Prospect)-[:BECAME_DISCOVERY]->(d:Discovery)

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

// ============================================================================
// AST Types
// ============================================================================

/// A complete pattern to match against the graph
#[derive(Debug, Clone)]
pub struct Pattern {
    pub elements: Vec<PatternElement>,
}

/// Either a node or edge pattern
#[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>,
    pub node_type: Option<String>,
    /// Additional labels from multi-label patterns like `(n:A:B)`. The
    /// first label lives in `node_type`; these are the rest.
    pub extra_labels: Vec<String>,
    pub properties: Option<HashMap<String, PropertyMatcher>>,
}

/// Pattern for matching edges: -[:TYPE {prop: value}]->
/// Supports variable-length paths with *min..max syntax:
/// - `*` or `*..` means 1 or more hops (default)
/// - `*2` means exactly 2 hops
/// - `*1..3` means 1 to 3 hops
/// - `*..5` means 1 to 5 hops
/// - `*2..` means 2 or more hops (up to default max)
#[derive(Debug, Clone)]
pub struct EdgePattern {
    pub variable: Option<String>,
    pub connection_type: Option<String>,
    /// Multiple allowed connection types from pipe syntax: `[:A|B|C]`.
    /// When set, an edge matches if its type equals ANY of these types.
    /// `connection_type` holds the first type for backward compatibility.
    pub connection_types: Option<Vec<String>>,
    pub direction: EdgeDirection,
    pub properties: Option<HashMap<String, PropertyMatcher>>,
    /// Variable-length path configuration: (min_hops, max_hops)
    /// None means exactly 1 hop (normal edge)
    pub var_length: Option<(usize, usize)>,
    /// 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` predicate that
    /// references only the edge variable (and the structural peer of
    /// the edge in this pattern). Applied during expansion in the
    /// matcher hot loop, *before* a row is materialized — eliminates
    /// edges whose post-materialization WHERE would have rejected
    /// them. The Cypher planner's
    /// [`super::super::languages::cypher::planner::rel_predicate_pushdown`]
    /// pass populates this field; it stays `None` for callers that
    /// build patterns by hand.
    pub edge_filter: Option<RelEdgeFilter>,
}

/// Inline edge filter — evaluated during expansion to skip edges the
/// downstream `WHERE` would have discarded. The single
/// [`RelEdgePredicate`] is wrapped to carry the original anchor side
/// of the edge (which determines how `startNode(r)` / `endNode(r)`
/// map onto the matcher's `direction` parameter).
#[derive(Debug, Clone)]
pub struct RelEdgeFilter {
    pub predicate: RelEdgePredicate,
    /// Which side of the edge the matcher anchors on, so the matcher
    /// can map `direction` back to startNode/endNode semantics.
    /// `Source` means the matcher is expanding outward from the
    /// pattern's left node; `Target` means from the right node.
    /// Set by the planner when it compiles `startNode(r) = …` /
    /// `endNode(r) = …` references.
    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 with `(edge_data, direction)` and an optional
/// pre-resolved peer node. Anything outside this enum the planner is
/// able to compile leaves the WHERE clause untouched and falls
/// through to the materialized predicate evaluator.
///
/// `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,
    /// Always-false sentinel.
    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),
    /// All sub-predicates must hold.
    And(Vec<RelEdgePredicate>),
    /// At least one sub-predicate must hold.
    Or(Vec<RelEdgePredicate>),
    /// Negated sub-predicate.
    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` give access to the edge body
    /// without requiring the matcher to materialize a full
    /// `EdgeData`. `peer_dir` is the matcher's per-edge direction
    /// translated into a consistent "is the peer on the start side?"
    /// boolean (the matcher computes this once per edge using the
    /// stored [`AnchorSide`]).
    ///
    /// Returns `true` only when the predicate evaluates to 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 => crate::graph::core::filtering::values_equal(&v, value),
                    PropOp::Ne => !crate::graph::core::filtering::values_equal(&v, value),
                    PropOp::Gt => matches!(
                        crate::graph::core::filtering::compare_values(&v, value),
                        Some(std::cmp::Ordering::Greater)
                    ),
                    PropOp::Ge => matches!(
                        crate::graph::core::filtering::compare_values(&v, value),
                        Some(std::cmp::Ordering::Greater | std::cmp::Ordering::Equal)
                    ),
                    PropOp::Lt => matches!(
                        crate::graph::core::filtering::compare_values(&v, value),
                        Some(std::cmp::Ordering::Less)
                    ),
                    PropOp::Le => matches!(
                        crate::graph::core::filtering::compare_values(&v, value),
                        Some(std::cmp::Ordering::Less | std::cmp::Ordering::Equal)
                    ),
                    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(peer_is_start),
            RelEdgePredicate::EndNodeIsPeer => Some(!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),
        }
    }
}

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

/// Property value matcher
#[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,
    },
    /// IN-list matching: value must be one of these values.
    /// Pushed from `WHERE n.prop IN [v1, v2, ...]` by the planner.
    In(Vec<Value>),
    /// 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),
}

// ============================================================================
// Match Results
// ============================================================================

/// A single pattern match with variable bindings.
/// Uses Vec instead of HashMap — patterns add 1-6 unique variables,
/// so linear search is faster than hashing and clone is a single memcpy.
#[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,
}

/// A bound value (either node, edge, or variable-length path)
#[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_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)
        ));
    }

    #[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))));
        }
    }
}