graphdblite 0.1.2

use std::collections::HashMap;

use crate::cypher::ast::{
    Assignment, Expr, LiteralValue, Pattern, RemoveItem, ReturnItem, SetItem, SortItem,
};
use crate::types::Direction;

#[cfg(test)]
mod tests {
    use super::*;
    use crate::cypher::ast::{Expr, ExprKind, LiteralValue};

    #[test]
    fn fulltext_lookup_name_for_display() {
        let op = LogicalOp::FullTextLookup {
            label: "Doc".into(),
            alias: "n".into(),
            property: "body".into(),
            op: FullTextOp::Contains,
            term: Expr::synthetic(ExprKind::Literal(LiteralValue::String("foo".into()))),
            remaining_filters: None,
        };
        assert_eq!(op.op_name(), "FullTextLookup");
    }
}

/// Substring-match operator backed by a fulltext index. Distinguishes
/// the three predicate forms because anchored ops (`StartsWith`,
/// `EndsWith`) require a position post-filter the executor applies on
/// FTS candidates.
#[derive(Debug, Clone, Copy, PartialEq)]
pub enum FullTextOp {
    Contains,
    StartsWith,
    EndsWith,
}

/// Lookup-key payload for `IndexLookup`.
///
/// Either a baked-in `Literal` from the AST, or a `Param(name)` placeholder
/// resolved against the executor's `$param` map at exec time. Param-keyed
/// lookups are what let prepared-style queries (`WHERE n.prop = $x`) reuse a
/// cached plan across different parameter values.
#[derive(Debug, Clone, PartialEq)]
pub enum LookupKey {
    Literal(LiteralValue),
    Param(String),
}

/// Logical query plan operator. Language-agnostic IR that the executor consumes.
#[derive(Debug, Clone, PartialEq)]
pub enum LogicalOp {
    /// Emit a single empty row (input for standalone `RETURN` expressions).
    SingleRow,

    /// Scan all nodes with a label.
    Scan { label: String, alias: String },

    /// O(1) lookup of a single node by its internal id. Produced by the
    /// planner from `MATCH (n) WHERE id(n) = <expr>`. Bypasses both `Scan`
    /// and `IndexLookup` — directly fetches by primary key via
    /// `node::get_node`.
    ///
    /// `value_expr` is evaluated against the current record (in correlated
    /// contexts the outer row provides the bindings; in non-correlated
    /// contexts it must be a literal/param/etc. that resolves without
    /// any record). A non-integer or negative result yields zero rows
    /// rather than an error — matches the semantics of an unmatched
    /// `WHERE id(n) = ...` predicate.
    IdLookup { alias: String, value_expr: Expr },

    /// Index-based lookup: use a secondary index instead of a full label scan.
    ///
    /// Carries N >= 1 (property, key) pairs in column order. For `N == 1`
    /// this is a single-property lookup; for `N >= 2` the planner has matched
    /// a prefix of a composite index.
    IndexLookup {
        label: String,
        alias: String,
        /// Full property list of the chosen index (column order). For
        /// single-prop indexes `len == 1`; for composite indexes `len >= 2`.
        /// The executor passes this to `composite_index_prefix_lookup`
        /// so no per-query catalog lookup is required at execution time.
        index_properties: Vec<String>,
        /// The matched prefix: 1..=index_properties.len() (property, key)
        /// pairs, in the same order as the leading columns of
        /// `index_properties`. `lookups.len() <= index_properties.len()`.
        lookups: Vec<(String, LookupKey)>,
        /// Remaining inline property filters not covered by the index.
        remaining_filters: Option<Expr>,
    },

    /// Fulltext index lookup. Generated by the planner from
    /// `Filter(Scan{label}, BinOp(prop CONTAINS|STARTS WITH|ENDS WITH term))`
    /// when an FTS index exists on `(label, property)`. The term is held
    /// as an `Expr` (not a baked string) so `$param` queries stay
    /// plan-cacheable across param values.
    FullTextLookup {
        label: String,
        alias: String,
        property: String,
        op: FullTextOp,
        term: Expr,
        /// Remaining conjunct predicates not covered by the FTS lookup.
        remaining_filters: Option<Expr>,
    },

    /// Expand along edges from a source node.
    Expand {
        input: Box<LogicalOp>,
        src_alias: String,
        dst_alias: String,
        /// Relationship variable name (e.g. `r` in `[r:TYPE]`).
        rel_alias: Option<String>,
        /// Edge type(s) to match. Empty = any type. Multiple = match any of them.
        edge_types: Vec<String>,
        direction: Direction,
        min_hops: u32,
        max_hops: u32,
        /// True when the pattern uses `[*]` or `[*min..max]` syntax.
        /// Even `[*1..1]` is var-length (rel variable binds to a list).
        var_length: bool,
        /// For var-length patterns: inline property filters applied at each hop.
        /// Empty for fixed-length patterns (they use a separate Filter node).
        var_length_prop_filters: HashMap<String, Expr>,
        /// Optional cap on total output rows. Set by the planner's limit-pushdown
        /// pass when this Expand directly drives a Limit (only `Limit -> [Project]* ->
        /// Expand{var_length}` chains qualify). The executor stops enumeration once
        /// this many rows have been produced across all input records.
        result_cap: Option<u64>,
    },

    /// Cross-product of two pipelines (for multi-pattern MATCH).
    CrossProduct {
        left: Box<LogicalOp>,
        right: Box<LogicalOp>,
        /// True when both sides come from comma-separated patterns within a
        /// single MATCH clause — relationship uniqueness applies across them.
        same_match: bool,
    },

    /// Filter records by a predicate.
    Filter {
        input: Box<LogicalOp>,
        predicate: Expr,
    },

    /// Project (RETURN / WITH) columns from the record stream.
    ///
    /// `emit_compound` is `true` for the terminal RETURN so bare-variable
    /// projections yield `Value::Node` / `Value::Edge` values. Intermediate
    /// projections (WITH clauses) leave `emit_compound = false` so downstream
    /// operators continue to see the flat `var.prop` / `var.__id` shape they
    /// rely on for joins, ORDER BY, further pattern matching, etc.
    Project {
        input: Box<LogicalOp>,
        items: Vec<ReturnItem>,
        emit_compound: bool,
    },

    /// Aggregate records.
    Aggregate {
        input: Box<LogicalOp>,
        group_keys: Vec<Expr>,
        aggregates: Vec<AggregateExpr>,
    },

    /// Sort records.
    Sort {
        input: Box<LogicalOp>,
        items: Vec<SortItem>,
    },

    /// Remove duplicate records.
    Distinct { input: Box<LogicalOp> },

    /// Skip the first N records.
    Skip { input: Box<LogicalOp>, count: u64 },

    /// Limit the number of output records.
    Limit { input: Box<LogicalOp>, count: u64 },

    /// Create a node.
    CreateNode {
        labels: Vec<String>,
        alias: Option<String>,
        properties: HashMap<String, Expr>,
    },

    /// Create an edge between two already-bound variables.
    CreateEdge {
        src_alias: String,
        dst_alias: String,
        edge_type: String,
        rel_alias: Option<String>,
        properties: HashMap<String, Expr>,
    },

    /// Sequence of create operations (for multi-pattern CREATE).
    CreateSequence { ops: Vec<LogicalOp> },

    /// MATCH ... CREATE: run input pipeline, then create edges/nodes using bound variables.
    MatchCreate {
        input: Box<LogicalOp>,
        create_ops: Vec<LogicalOp>,
    },

    /// Delete nodes/edges/paths bound to variables or expressions.
    Delete {
        input: Box<LogicalOp>,
        exprs: Vec<crate::cypher::ast::Expr>,
        detach: bool,
    },

    /// Set properties on nodes/edges.
    SetProperty {
        input: Box<LogicalOp>,
        assignments: Vec<Assignment>,
    },

    /// Add labels to a node.
    SetLabel {
        input: Box<LogicalOp>,
        variable: String,
        labels: Vec<String>,
    },

    /// Set all properties on a node (overwrite or merge).
    SetProperties {
        input: Box<LogicalOp>,
        variable: String,
        value: Expr,
        merge: bool,
    },

    /// Remove properties/labels from nodes/edges.
    Remove {
        input: Box<LogicalOp>,
        items: Vec<RemoveItem>,
    },

    /// Merge: match-or-create pattern.
    Merge {
        pattern: Pattern,
        on_create: Vec<SetItem>,
        on_match: Vec<SetItem>,
    },

    /// MATCH ... MERGE: merge a pattern using bound variables from MATCH pipeline.
    MatchMerge {
        input: Box<LogicalOp>,
        merge_pattern: Pattern,
        on_create: Vec<SetItem>,
        on_match: Vec<SetItem>,
    },

    /// Build a Path value from the matched pattern elements and store it
    /// in the record under the given alias. `node_aliases` and `rel_aliases`
    /// list the variables in traversal order.
    MaterializePath {
        input: Box<LogicalOp>,
        path_alias: String,
        node_aliases: Vec<String>,
        rel_aliases: Vec<String>,
    },

    /// Correlated inner join: for each input record, execute right side with
    /// bindings from the left. Only emit combined rows. If right produces
    /// nothing for a left row, that row is dropped.
    CorrelatedJoin {
        input: Box<LogicalOp>,
        right: Box<LogicalOp>,
        /// True when both sides come from comma-separated patterns within a
        /// single MATCH clause — relationship uniqueness applies across them.
        same_match: bool,
    },

    /// Left outer join: for each input record, attempt right side; emit NULLs if no match.
    LeftOuterJoin {
        input: Box<LogicalOp>,
        right: Box<LogicalOp>,
        /// Aliases from the optional pattern that should be NULL-filled on no match.
        optional_aliases: Vec<String>,
        /// Optional WHERE clause applied after joining but before null-filling.
        /// If the predicate fails on a joined row, that row is null-filled instead.
        opt_filter: Option<Expr>,
    },

    /// Unwind a list expression into one record per element.
    Unwind {
        input: Box<LogicalOp>,
        expr: Expr,
        alias: String,
    },

    /// Execute a registered procedure.
    Call {
        input: Box<LogicalOp>,
        procedure_name: String,
        /// Evaluated argument expressions.
        args: Vec<Expr>,
        /// Output columns to yield: (procedure_output_col, optional_alias).
        yield_items: Vec<(String, Option<String>)>,
        /// True when YIELD * — emit all output columns.
        yield_star: bool,
    },

    /// Find shortest path(s) between two bound nodes.
    ShortestPath {
        input: Box<LogicalOp>,
        src_alias: String,
        dst_alias: String,
        path_alias: String,
        edge_type: Option<String>,
        direction: Direction,
        max_hops: u32,
        all_paths: bool,
    },

    /// Union of multiple pipelines. `all` = keep duplicates.
    Union { inputs: Vec<LogicalOp>, all: bool },

    /// Produce a single empty record (used as starting input for scans).
    EmptyRow,

    /// CREATE INDEX ON :Label(prop1, prop2, ...)
    CreateIndex {
        label: String,
        properties: Vec<String>,
    },

    /// DROP INDEX ON :Label(prop1, prop2, ...)
    DropIndex {
        label: String,
        properties: Vec<String>,
    },
}

impl LogicalOp {
    /// Short human-readable name for tracing/debugging.
    pub fn op_name(&self) -> &'static str {
        match self {
            Self::SingleRow => "SingleRow",
            Self::Scan { .. } => "Scan",
            Self::IdLookup { .. } => "IdLookup",
            Self::IndexLookup { .. } => "IndexLookup",
            Self::FullTextLookup { .. } => "FullTextLookup",
            Self::Expand { .. } => "Expand",
            Self::CrossProduct { .. } => "CrossProduct",
            Self::Filter { .. } => "Filter",
            Self::Project { .. } => "Project",
            Self::Aggregate { .. } => "Aggregate",
            Self::Sort { .. } => "Sort",
            Self::Distinct { .. } => "Distinct",
            Self::Skip { .. } => "Skip",
            Self::Limit { .. } => "Limit",
            Self::CreateNode { .. } => "CreateNode",
            Self::CreateEdge { .. } => "CreateEdge",
            Self::CreateSequence { .. } => "CreateSequence",
            Self::MatchCreate { .. } => "MatchCreate",
            Self::Delete { .. } => "Delete",
            Self::SetProperty { .. } => "SetProperty",
            Self::SetLabel { .. } => "SetLabel",
            Self::SetProperties { .. } => "SetProperties",
            Self::Remove { .. } => "Remove",
            Self::Merge { .. } => "Merge",
            Self::MatchMerge { .. } => "MatchMerge",
            Self::MaterializePath { .. } => "MaterializePath",
            Self::CorrelatedJoin { .. } => "CorrelatedJoin",
            Self::LeftOuterJoin { .. } => "LeftOuterJoin",
            Self::Unwind { .. } => "Unwind",
            Self::Call { .. } => "Call",
            Self::ShortestPath { .. } => "ShortestPath",
            Self::Union { .. } => "Union",
            Self::EmptyRow => "EmptyRow",
            Self::CreateIndex { .. } => "CreateIndex",
            Self::DropIndex { .. } => "DropIndex",
        }
    }
}

/// An aggregate expression within an Aggregate operator.
#[derive(Debug, Clone, PartialEq)]
pub struct AggregateExpr {
    pub function: AggregateFunction,
    pub input: Expr,
    pub alias: Option<String>,
    pub distinct: bool,
    /// Second argument for percentile functions (the percentile value).
    pub extra_arg: Option<Expr>,
    /// Original function name as written in the query (preserves case).
    pub original_name: String,
    /// Original full function call text for column naming (preserves whitespace).
    pub original_call_text: Option<String>,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum AggregateFunction {
    Count,
    Sum,
    Avg,
    Min,
    Max,
    Collect,
    PercentileDisc,
    PercentileCont,
    StDev,
    StDevP,
}