kglite 0.17.11

//! Cypher query optimizer.

use super::ast::*;
use crate::datatypes::values::Value;
use crate::graph::schema::DirGraph;
use std::collections::{HashMap, HashSet};

mod annotations;
mod invariants;
#[cfg(debug_assertions)]
use invariants::debug_check_invariants;
pub mod cost_model;
pub mod fusion;
pub mod index_selection;
pub mod join_order;
mod nested;
mod node_anchor;
pub mod rel_predicate_pushdown;
pub mod schema_check;
pub mod simplification;
mod var_length_lowering;
mod with_boundary;

use annotations::{
    pass_mark_disjoint_fixed_trails, pass_mark_fast_var_length_paths,
    pass_mark_skip_target_type_check,
};
use cost_model::reorder_predicates_by_cost;
use fusion::{
    fuse_anchored_edge_count, fuse_count_short_circuits, fuse_match_return_aggregate,
    fuse_match_with_aggregate, fuse_match_with_aggregate_top_k, fuse_node_scan_aggregate,
    fuse_node_scan_top_k, fuse_optional_match_aggregate, fuse_order_by_top_k, fuse_spatial_join,
    fuse_text_bm25_order_limit, fuse_vector_score_order_limit, mark_return_lazy_eligible,
};
use index_selection::push_where_into_match;
use join_order::{
    optimize_pattern_start_node, reorder_cyclic_pattern_edges, reorder_match_clauses,
    reorder_match_patterns,
};
pub(crate) use nested::import_pattern_anchors_in_arm;
use node_anchor::anchor_element_id;
use rel_predicate_pushdown::extract_pushable_rel_predicates_with_params;
use var_length_lowering::lower_fixed_var_length_hops;
use with_boundary::{
    pass_fold_aliasing_with, pass_hoist_terminal_return_over_with_top_k, pass_hoist_with_where,
};

use simplification::{
    desugar_multi_match_return_aggregate, fold_or_to_in, fold_pass_through_with,
    narrow_unwind_source, push_distinct_into_match, push_limit_into_aggregate,
    push_limit_into_match, rewrite_count_bound_var_to_star,
};

/// Carries the per-call inputs every pass might need, so a new dependency
/// extends this struct rather than 25+ wrapper signatures.
pub struct PassCtx<'a> {
    pub graph: &'a DirGraph,
    pub params: &'a HashMap<String, Value>,
    pub disabled: &'a HashSet<String>,
    initial_scope: &'a HashSet<String>,
    global_scope: &'a HashSet<String>,
}

type PassFn = fn(&mut CypherQuery, &PassCtx);

/// The optimizer pipeline as a single source of truth. Order is
/// load-bearing — comments on individual entries call out cross-pass
/// dependencies. Adding a new pass: write the impl, write a `pass_*`
/// wrapper, register here with a unique name, doc-comment the wrapper,
/// add at least one query to `tests/test_cypher_differential.py`. The
/// wrapper lives in this file by default and beside its impl when this
/// file is at its line cap (`hoist_with_where` is the first such case).
///
/// ## `CALL { }` (CallSubquery) barrier audit
///
/// `Clause::CallSubquery` is an OPAQUE barrier to every pass below. A
/// subquery's per-row cardinality is unknown at plan time and a
/// correlated body depends on its seeded input — so NO pass may move a
/// clause across it, fuse a window through it, or push a LIMIT/predicate
/// into or past it. Every pass is safe today; only two are not safe
/// purely by shape:
///
/// - `optimize_nested_queries` — **recurses by design.** Owns body
///   optimization; import-aware (disables seed-ignoring fusion for
///   anchored correlated bodies).
/// - `fold_pass_through_with` — **guarded.** Its downstream-ref check
///   records a CallSubquery's import names + body refs (see
///   `collect_clause_variables`), so a `WITH` a correlated CALL depends
///   on is never folded away.
///
/// Every other pass either matches a fixed clause-adjacency window
/// (which a CallSubquery breaks), rewrites/reorders WITHIN one clause,
/// or fires only on a whole-query shape (`[Match, Return]`, len 2); the
/// clause walks hit `_ => continue` or end their span at `_ => break`,
/// and the downstream-dedup-safety scan stops at the first Return/With,
/// which a CallSubquery is not. Two conservative details: prior-scope
/// helpers under-report CALL outputs, so pushdown under-pushes; and
/// `anchor_element_id` writes only a hint while leaving its predicate
/// standing, so a hint written against a body it could not see changes
/// no answer.
pub const PASSES: &[(&str, PassFn)] = &[
    (
        "optimize_nested_queries",
        nested::pass_optimize_nested_queries,
    ),
    // `*k..k` → k explicit hops. Runs FIRST among the structural rewrites:
    // every pass below bails on `var_length.is_some()`, so a segment lowered
    // any later would inherit none of them.
    (
        "lower_fixed_var_length_hops",
        pass_lower_fixed_var_length_hops,
    ),
    // count(bound node/edge var) → count(*): runs early so the rewritten
    // count(*) reaches the count-fusion + light-row MATCH paths.
    (
        "rewrite_count_bound_var_to_star",
        pass_rewrite_count_bound_var_to_star,
    ),
    // Lift a `WITH … WHERE` predicate into a standalone WHERE ahead of the
    // WITH. Runs BEFORE the pushdown passes and the pass-through fold, which
    // are the two things it exists to feed: with the predicate hoisted, both
    // `push_where_into_match` passes see it and `fold_pass_through_with` sees
    // a WITH whose `where_clause` is now empty.
    ("hoist_with_where", pass_hoist_with_where),
    ("push_where_into_match.1", pass_push_where_into_match),
    ("fold_or_to_in", pass_fold_or_to_in),
    // second push_where pass: catches IN predicates created by fold_or_to_in
    ("push_where_into_match.2", pass_push_where_into_match),
    // Slot anchors from elementId() equality. Must precede the join-order
    // passes: they read `node_anchors` to score an anchored variable as
    // fully selective.
    ("anchor_element_id", pass_anchor_element_id),
    (
        "extract_pushable_rel_predicates",
        pass_extract_pushable_rel_predicates,
    ),
    // strip pass-through WITH BEFORE cross-clause MATCH reorder so the
    // latter sees a contiguous Match-Match span when a `WITH p` sat between.
    ("fold_pass_through_with", pass_fold_pass_through_with),
    // Substitute an aliasing WITH away, so the top-K fusion window can form
    // across the boundary it was holding. Runs immediately after
    // `fold_pass_through_with` (which owns the bare-variable case) and well
    // before the fusion block that consumes the result.
    ("fold_aliasing_with", pass_fold_aliasing_with),
    (
        "hoist_terminal_return_over_with_top_k",
        pass_hoist_terminal_return_over_with_top_k,
    ),
    // Runs AFTER fold_pass_through_with: folding a pass-through WITH changes
    // which clauses sit downstream of an UNWIND, and this pass's whole job is
    // to read that downstream set. Running it first would decide against a
    // stale clause list.
    ("narrow_unwind_source", pass_narrow_unwind_source),
    // rewrites Match-Match-Return(group, agg) so the aggregate-fusion +
    // top-K pipeline can pick it up.
    (
        "desugar_multi_match_return_aggregate",
        pass_desugar_multi_match_return_aggregate,
    ),
    ("fuse_spatial_join", pass_fuse_spatial_join),
    // O(1) cost-proxy reorder. Runs BEFORE pattern_start_node so reversal
    // sees the post-reorder clause sequence and tracks bound_vars correctly.
    ("reorder_match_clauses", pass_reorder_match_clauses),
    // Re-root simple cyclic patterns at their most-selective node BEFORE
    // pattern_start_node (which can't help a cycle — both ends are the same
    // variable, so its reverse is a no-op).
    (
        "reorder_cyclic_pattern_edges",
        pass_reorder_cyclic_pattern_edges,
    ),
    (
        "optimize_pattern_start_node",
        pass_optimize_pattern_start_node,
    ),
    ("reorder_match_patterns", pass_reorder_match_patterns),
    ("push_limit_into_match", pass_push_limit_into_match),
    ("push_limit_into_aggregate", pass_push_limit_into_aggregate),
    ("push_distinct_into_match", pass_push_distinct_into_match),
    ("fuse_anchored_edge_count", pass_fuse_anchored_edge_count),
    ("fuse_count_short_circuits", pass_fuse_count_short_circuits),
    (
        "fuse_optional_match_aggregate",
        pass_fuse_optional_match_aggregate,
    ),
    (
        "fuse_match_return_aggregate",
        pass_fuse_match_return_aggregate,
    ),
    ("fuse_match_with_aggregate", pass_fuse_match_with_aggregate),
    // top-K absorption AFTER fuse_match_with_aggregate (which produces
    // FusedMatchWithAggregate) but BEFORE fuse_order_by_top_k (which would
    // otherwise consume the downstream RETURN+ORDER BY+LIMIT).
    (
        "fuse_match_with_aggregate_top_k",
        pass_fuse_match_with_aggregate_top_k,
    ),
    ("fuse_node_scan_aggregate", pass_fuse_node_scan_aggregate),
    ("fuse_node_scan_top_k", pass_fuse_node_scan_top_k),
    (
        "fuse_vector_score_order_limit",
        pass_fuse_vector_score_order_limit,
    ),
    // Same three-clause shape as fuse_order_by_top_k, so it must run first or
    // the generic pass takes it: a bail here is a fall-through to that pass,
    // never a lost answer.
    (
        "fuse_text_bm25_order_limit",
        pass_fuse_text_bm25_order_limit,
    ),
    ("fuse_order_by_top_k", pass_fuse_order_by_top_k),
    (
        "reorder_predicates_by_cost",
        pass_reorder_predicates_by_cost,
    ),
    (
        "mark_fast_var_length_paths",
        pass_mark_fast_var_length_paths,
    ),
    (
        "mark_disjoint_fixed_trails",
        pass_mark_disjoint_fixed_trails,
    ),
    (
        "mark_skip_target_type_check",
        pass_mark_skip_target_type_check,
    ),
];

/// PyAPI uses this to reject typos in the `disabled_passes` kwarg before
/// they silently suppress nothing.
pub fn is_known_pass(name: &str) -> bool {
    PASSES.iter().any(|(n, _)| *n == name)
}

/// Backs the PyAPI's `disable_optimizer=True` shortcut, which expands to
/// "disable everything".
pub fn all_pass_names() -> Vec<String> {
    PASSES.iter().map(|(n, _)| n.to_string()).collect()
}

/// Annotate the top-level query's terminal RETURN with `lazy_eligible`
/// when no downstream operator forces row materialisation. Called once
/// after `optimize`, never recursively, so nested UNION arms don't get
/// marked (their results pass through the union machinery, which expects
/// fully evaluated rows).
pub fn mark_lazy_eligibility(query: &mut CypherQuery) {
    // Don't mark when the top-level query contains a UNION — the union
    // machinery merges materialised rows.
    if query.clauses.iter().any(|c| matches!(c, Clause::Union(_))) {
        return;
    }
    // Don't mark for mutation queries — CREATE/SET/DELETE/REMOVE/MERGE go
    // through `execute_mutable`, which doesn't read the lazy descriptor
    // and would produce empty rows.
    if query.clauses.iter().any(|c| {
        matches!(
            c,
            Clause::Create(_)
                | Clause::Set(_)
                | Clause::Delete(_)
                | Clause::Remove(_)
                | Clause::Merge(_)
        )
    }) {
        return;
    }
    mark_return_lazy_eligible(query);
}

/// Run the optimizer pipeline with no passes disabled. The entry point
/// production callers (executor, transactions, mutations) should use.
pub fn optimize(query: &mut CypherQuery, graph: &DirGraph, params: &HashMap<String, Value>) {
    optimize_with_disabled(query, graph, params, empty_disabled_set());
}

/// Process-lifetime empty `HashSet<String>` used as the no-knob default.
pub fn empty_disabled_set() -> &'static HashSet<String> {
    static EMPTY: std::sync::OnceLock<HashSet<String>> = std::sync::OnceLock::new();
    EMPTY.get_or_init(HashSet::new)
}

/// Run the optimizer pipeline, skipping any pass whose name is in
/// `disabled`. Diagnostic hook for the differential test harness and
/// `cypher(..., disabled_passes=[...])` kwarg — production callers should
/// use the no-knob `optimize()` wrapper.
pub fn optimize_with_disabled(
    query: &mut CypherQuery,
    graph: &DirGraph,
    params: &HashMap<String, Value>,
    disabled: &HashSet<String>,
) {
    let empty_scope = empty_disabled_set();
    optimize_with_disabled_scoped(query, graph, params, disabled, empty_scope, empty_scope);
}

fn optimize_with_disabled_scoped(
    query: &mut CypherQuery,
    graph: &DirGraph,
    params: &HashMap<String, Value>,
    disabled: &HashSet<String>,
    initial_scope: &HashSet<String>,
    global_scope: &HashSet<String>,
) {
    query.optimizer_tags.clear();
    let ctx = PassCtx {
        graph,
        params,
        disabled,
        initial_scope,
        global_scope,
    };
    for (name, pass_fn) in PASSES {
        if disabled.contains(*name) {
            continue;
        }
        let before = query.explain.then(|| format!("{:?}", query.clauses));
        pass_fn(query, &ctx);
        if before.is_some_and(|snapshot| snapshot != format!("{:?}", query.clauses)) {
            query.optimizer_tags.push((*name).to_string());
        }
        #[cfg(debug_assertions)]
        debug_check_invariants(query, name);
    }
}

/// **Pass:** `lower_fixed_var_length_hops` — **Precondition:** a
/// `MATCH` / `OPTIONAL MATCH` with no path assignment. **Pattern
/// matched:** a relationship element whose `var_length` is `(k, k)`,
/// `k >= 1`. **Rewrite:** `k` copies of that element with
/// `var_length` cleared, separated by anonymous unlabelled nodes; type
/// alternations, inline relationship properties and direction are
/// replicated onto every copy, which is what variable-length semantics
/// already demand of each relationship in the segment. **Why:** every
/// pass below — relationship pushdown, start-node selection, the fusion
/// family, `mark_disjoint_fixed_trails`, `mark_skip_target_type_check` —
/// declines a variable-length element, so the star spelling of a
/// fixed-length question paid for none of them (measured 3.4x to 17x
/// inside the one-or-two-hop window; the table is in
/// `var_length_lowering`).
///
/// **Why-bail:** `min != max`; `k == 0` (the zero-length identity is not
/// a hop); `k` or the pattern's post-lowering hop count above the
/// module's ceiling; a bound relationship variable (`r` binds the
/// *list*); a path assignment anywhere in the clause; a pre-existing
/// `edge_filter` or unresolved `-[:$type]->` slot (neither can be set at
/// this point in `PASSES` — they are "the pipeline moved" guards).
///
/// The trail-semantics argument, and why the disjoint-types opt-out
/// cannot fire on a lowered segment, are in `var_length_lowering`'s
/// module docs.
fn pass_lower_fixed_var_length_hops(query: &mut CypherQuery, _ctx: &PassCtx) {
    lower_fixed_var_length_hops(query)
}

/// **Pass:** `push_where_into_match` — Move comparison predicates from
/// a trailing `WHERE` clause into the preceding `MATCH`'s
/// `PropertyMatcher`. The matcher applies them during pattern expansion
/// instead of evaluating them per row, pruning the search early.
///
/// Two sources of predicate, one rewrite: a trailing `Clause::Where` after
/// a `MATCH`/`OPTIONAL MATCH`, and an `OPTIONAL MATCH`'s own
/// `MatchClause::where_clause`. The second is safe because under clause
/// scoping a candidate the predicate rejects and a candidate the pattern
/// never produced are the same outcome — the row is null-extended either
/// way — so moving the test from the predicate into the pattern cannot
/// change which rows survive. The safety-net rule is unchanged in both
/// homes: a partial push leaves the whole predicate standing, and a full
/// push keeps it as the filter every non-pattern-matcher path still relies
/// on.
fn pass_push_where_into_match(query: &mut CypherQuery, ctx: &PassCtx) {
    push_where_into_match(query, ctx.params)
}

/// **Pass:** `anchor_element_id` — Record the slot named by
/// `WHERE elementId(v) = <literal|$param>` on the MATCH clause, so the
/// executor seeds it as a pre-binding instead of scanning for it.
///
/// **Precondition:** a `Clause::Match`/`Clause::OptionalMatch` with a WHERE —
/// its own (the scoped `OPTIONAL MATCH` form) or the adjacent standalone one.
///
/// **Pattern matched:** an `Equals` comparison, either operand order, between
/// `elementId(v)` — where `v` is a node variable of *this* clause's patterns —
/// and a value that parses to a non-negative slot. Only the predicate's `And`
/// spine is descended.
///
/// **Rewrite:** pushes `(v, NodeIndex)` onto `MatchClause::node_anchors`. The
/// predicate is left standing, so this narrows the candidate set without
/// owning the answer: an out-of-range or stale slot resolves to no node, which
/// is what the retained predicate would have concluded.
///
/// **Why-bail:** `Or`/`Not`/`Xor` are not descended (a disjunct constrains
/// nothing, and a negation inverts the reasoning); a non-numeric, negative or
/// unbound value does not name a slot; a variable this clause does not bind
/// belongs to another clause's search space. Conflicting anchors on one
/// variable keep the first — two of them cannot both hold, and the predicate
/// rejects the loser.
fn pass_anchor_element_id(query: &mut CypherQuery, ctx: &PassCtx) {
    anchor_element_id(query, ctx.params)
}

/// **Pass:** `fold_or_to_in` — Rewrite `(a.x = v1 OR a.x = v2 OR ...)`
/// chains into `a.x IN [v1, v2, ...]`. Lets the second
/// `push_where_into_match` push the synthesized IN as a single
/// equality-set matcher.
fn pass_fold_or_to_in(query: &mut CypherQuery, _ctx: &PassCtx) {
    fold_or_to_in(query)
}

/// **Pass:** `rewrite_count_bound_var_to_star` — rewrite non-distinct
/// `count(v)` to `count(*)` when `v` is a mandatorily-bound node/edge variable
/// (so always non-null). Avoids per-row node materialization and heavy binding
/// retention on deep-path counts. WHY-BAIL: DISTINCT, OPTIONAL-bound `v`, or any
/// `WITH` present. Column name preserved via alias.
fn pass_rewrite_count_bound_var_to_star(query: &mut CypherQuery, _ctx: &PassCtx) {
    rewrite_count_bound_var_to_star(query)
}

/// **Pass:** `extract_pushable_rel_predicates` — Inline edge-side
/// predicates (`type(r) = 'X'`, `r.prop OP literal`, `startNode(r) =
/// peer`) from a trailing WHERE into the edge's `rel_predicate`. The
/// matcher applies them during expansion, before per-edge bindings are
/// allocated. WHY-BAIL: predicates referencing unbound vars stay in WHERE.
fn pass_extract_pushable_rel_predicates(query: &mut CypherQuery, ctx: &PassCtx) {
    extract_pushable_rel_predicates_with_params(query, ctx.params)
}

/// **Pass:** `fold_pass_through_with` — Strip `WITH x AS x` /
/// pass-through `WITH *` clauses that don't reshape the row stream.
/// Removing them lets `reorder_match_clauses` see contiguous Match
/// spans for cross-clause reorder; otherwise the WITH would block.
fn pass_fold_pass_through_with(query: &mut CypherQuery, _ctx: &PassCtx) {
    fold_pass_through_with(query)
}

/// **Pass:** `narrow_unwind_source` — mark `UNWIND <var> AS alias` whose
/// source binding is dead after the clause, letting the executor take the list
/// out of the row by move instead of cloning it into all `n` expanded rows.
///
/// **Precondition:** the UNWIND source is a bare variable (a computed source is
/// never bound in the row). **Pattern:** no clause after the UNWIND mentions
/// the source variable, and every such clause has fully enumerable references.
/// **Rewrite:** sets `UnwindClause::consume_source`; adds/removes/reorders
/// nothing and cannot change results. **Why bail:** a write / fused /
/// procedure clause downstream, whose references are not enumerable, or any
/// downstream mention of the variable — either way the binding may still be
/// read, so the copy is kept.
///
/// Fixes the quadratic memory of `WITH collect(x) AS xs UNWIND xs AS y`:
/// `n` rows each retaining an `n`-element copy of the same list.
fn pass_narrow_unwind_source(query: &mut CypherQuery, _ctx: &PassCtx) {
    narrow_unwind_source(query)
}

/// **Pass:** `desugar_multi_match_return_aggregate` — Rewrite
/// `MATCH ... MATCH ... RETURN <group>, <agg>` into the equivalent
/// `MATCH ... MATCH ... WITH <group>, <agg> RETURN <project>` so the
/// aggregate-fusion + top-K pipeline can pick it up. The WITH groups
/// by the user-specified RETURN expressions (per-property), not by the
/// source variable (which would over-finely group when the property
/// has duplicates across instances).
fn pass_desugar_multi_match_return_aggregate(query: &mut CypherQuery, _ctx: &PassCtx) {
    desugar_multi_match_return_aggregate(query)
}

/// **Pass:** `fuse_spatial_join` — Specialize `MATCH ... WHERE
/// contains(geom_a, geom_b)` into a spatial-join iterator that probes a
/// per-query R-tree over the container type instead of running a
/// cartesian product + per-pair filter. WHY-BAIL: a graph with secondary
/// labels — the R-tree is built from primary `type_indices` only, so it
/// would miss secondary-labelled nodes.
fn pass_fuse_spatial_join(query: &mut CypherQuery, ctx: &PassCtx) {
    fuse_spatial_join(query, ctx.graph)
}

/// **Pass:** `reorder_match_clauses` — Reorder adjacent `MATCH` clauses
/// by connection-type total counts (O(1) cost proxy) so the smaller
/// driver runs first.
fn pass_reorder_match_clauses(query: &mut CypherQuery, ctx: &PassCtx) {
    reorder_match_clauses(query, ctx.graph)
}

/// **Pass:** `reorder_cyclic_pattern_edges` — Re-root a simple cyclic pattern
/// (a ring whose start variable repeats at the end) at its most-selective node,
/// orienting the walk so the cheaper incident edge drives first. Turns the
/// cycle-closing segment into an O(1) bound-target check in the matcher.
/// Shape-gated: only fires on simple rings of clean single-typed edges and only
/// on a clear (≥4×) selectivity win, leaving every acyclic pattern unchanged.
fn pass_reorder_cyclic_pattern_edges(query: &mut CypherQuery, ctx: &PassCtx) {
    reorder_cyclic_pattern_edges(query, ctx.graph)
}

/// **Pass:** `optimize_pattern_start_node` — For 3+-element patterns,
/// reverse the pattern so iteration starts from the most-selective node
/// (typically id-anchored or smallest-cardinality type). Reduces the
/// front of the join from O(N) to O(1) when one end is anchored.
fn pass_optimize_pattern_start_node(query: &mut CypherQuery, ctx: &PassCtx) {
    optimize_pattern_start_node(query, ctx.graph)
}

/// **Pass:** `reorder_match_patterns` — Reorder multiple comma-
/// separated patterns within one `MATCH` clause by size/type
/// selectivity. Sibling of `reorder_match_clauses` but operates within
/// a single MATCH.
fn pass_reorder_match_patterns(query: &mut CypherQuery, ctx: &PassCtx) {
    reorder_match_patterns(query, ctx.graph)
}

/// **Pass:** `push_limit_into_match` — Mark the trailing `LIMIT N` as
/// an early-stop hint on the preceding `MATCH` so the executor can
/// short-circuit pattern expansion. WHY-BAIL: requires single-MATCH
/// queries (multi-MATCH + WHERE on a late-bound var silently drops
/// rows).
fn pass_push_limit_into_match(query: &mut CypherQuery, ctx: &PassCtx) {
    push_limit_into_match(query, ctx.graph)
}

/// **Pass:** `push_limit_into_aggregate` — Stamp `group_limit_hint`
/// on a `RETURN/WITH` that has both group keys and aggregates when the
/// next clause is a literal `LIMIT N`. The aggregator stops creating
/// new groups after `N` distinct keys; rows for already-collected keys
/// continue to feed their aggregates. WHY-BAIL: ORDER BY between
/// projection and LIMIT changes which N rows survive (need every group
/// to find the top N), so the pass leaves those queries to the
/// materialised path. DISTINCT, `RETURN … HAVING` and a `WITH`'s inline
/// `WHERE`/`HAVING` also bail — the filter runs *after* the projection,
/// so a capped group set would drop qualifying groups the LIMIT still
/// had room for. The trailing LIMIT clause stays in the plan as a hard
/// cap.
fn pass_push_limit_into_aggregate(query: &mut CypherQuery, ctx: &PassCtx) {
    push_limit_into_aggregate(query, ctx.graph)
}

/// **Pass:** `push_distinct_into_match` — Mark `RETURN DISTINCT` /
/// `WITH DISTINCT` as a hint on the preceding MATCH so the executor
/// can dedup during expansion instead of materializing all rows first.
fn pass_push_distinct_into_match(query: &mut CypherQuery, _ctx: &PassCtx) {
    push_distinct_into_match(query)
}

/// **Pass:** `fuse_anchored_edge_count` — Specialize
/// `MATCH (id:VAL)-[r:T]->(v) RETURN count(*)` into an O(1) anchored
/// edge lookup using the connection type's edge count metadata.
fn pass_fuse_anchored_edge_count(query: &mut CypherQuery, ctx: &PassCtx) {
    fuse_anchored_edge_count(query, ctx.graph)
}

/// **Pass:** `fuse_count_short_circuits` — Answer a `MATCH` + `RETURN`
/// count query from type-bucket / CSR metadata instead of expanding the
/// pattern: `MATCH (n[:Type])` or `MATCH ()-[r[:T]]->()` with
/// `count(*)`/`count(var)`, optionally paired with a `type(r)`/`labels(n)`
/// group key. Typed hop endpoints (`(a:Person)-[:KNOWS]->(b)`) fuse to
/// the same `FusedCountTypedEdge` when `KNOWS` schema endpoints are the
/// singleton `{Person}`. WHY-BAIL: `DISTINCT`, `HAVING`, a path assignment,
/// more than one pattern, an inline property filter or a repeated variable,
/// a multi-label `(n:A:B)` (needs an intersection the O(1) bucket count
/// cannot express), untyped undirected or `[:A|B]` edges, and a typed endpoint
/// whose connection-type schema does not cover every counted orientation.
fn pass_fuse_count_short_circuits(query: &mut CypherQuery, ctx: &PassCtx) {
    fuse_count_short_circuits(query, ctx.graph.has_secondary_labels, ctx.graph)
}

/// **Pass:** `fuse_optional_match_aggregate` — Fuse
/// `OPTIONAL MATCH ... RETURN <agg>` into a single
/// `FusedOptionalMatchAggregate` clause that counts matches per input
/// row without materializing intermediate per-row expansions. WHY-BAIL:
/// edge vars (`count(r)`) count as local-to-OPT; multi-pattern clauses
/// and a clause-owned `WHERE` (`OPTIONAL MATCH … WHERE …`) bail, the latter
/// because the fused counter counts a pattern's matches with no hook to
/// test a predicate per candidate. **No grouping key** bails too: the fused
/// operator is driven by the incoming rows, so an ungrouped aggregate — which
/// yields exactly one row over the whole expansion, and one row with zero even
/// when the driving set is empty — has no driving row to attach its single
/// result to.
fn pass_fuse_optional_match_aggregate(query: &mut CypherQuery, _ctx: &PassCtx) {
    fuse_optional_match_aggregate(query)
}

/// **Pass:** `fuse_match_return_aggregate` — Fuse
/// `MATCH ... RETURN <group_keys>, <agg>` into
/// `FusedMatchReturnAggregate`, building the GROUP-BY hash map inline
/// during pattern expansion.
fn pass_fuse_match_return_aggregate(query: &mut CypherQuery, ctx: &PassCtx) {
    fuse_match_return_aggregate(query, ctx.graph)
}

/// **Pass:** `fuse_match_with_aggregate` — Like
/// `fuse_match_return_aggregate`, but for `MATCH ... WITH <group>,
/// <agg>` (pipeline continues after WITH). Emits
/// `FusedMatchWithAggregate`.
fn pass_fuse_match_with_aggregate(query: &mut CypherQuery, ctx: &PassCtx) {
    fuse_match_with_aggregate(query, ctx.graph)
}

/// **Pass:** `fuse_match_with_aggregate_top_k` — Absorb a downstream
/// `ORDER BY <agg> LIMIT k` into a preceding
/// `FusedMatchWithAggregate`, replacing full sort with heap-pruned
/// top-K (O(n log k) instead of O(n log n)).
fn pass_fuse_match_with_aggregate_top_k(query: &mut CypherQuery, _ctx: &PassCtx) {
    fuse_match_with_aggregate_top_k(query)
}

/// **Pass:** `fuse_node_scan_aggregate` — Untyped `MATCH (n) RETURN
/// <agg>` → specialized scan-only aggregate that walks the node store
/// once without producing intermediate row tuples.
fn pass_fuse_node_scan_aggregate(query: &mut CypherQuery, ctx: &PassCtx) {
    fuse_node_scan_aggregate(query, ctx.params)
}

/// **Pass:** `fuse_node_scan_top_k` — `MATCH (n:Type) RETURN n LIMIT k`
/// → specialized scan that returns the first k nodes of the type
/// without going through the pattern executor.
fn pass_fuse_node_scan_top_k(query: &mut CypherQuery, ctx: &PassCtx) {
    fuse_node_scan_top_k(query, ctx.params)
}

/// **Pass:** `fuse_vector_score_order_limit` — `MATCH ...
/// vector_score(...) ORDER BY score LIMIT k` → top-K via a vector-
/// score min-heap. Projects RETURN expressions only for the k surviving
/// rows.
fn pass_fuse_vector_score_order_limit(query: &mut CypherQuery, _ctx: &PassCtx) {
    fuse_vector_score_order_limit(query)
}

/// **Pass:** `fuse_text_bm25_order_limit` — `RETURN ... text_bm25(...) AS s
/// ORDER BY s DESC LIMIT k` → one postings-driven top-k operator.
fn pass_fuse_text_bm25_order_limit(query: &mut CypherQuery, _ctx: &PassCtx) {
    fuse_text_bm25_order_limit(query)
}

/// **Pass:** `fuse_order_by_top_k` — Generic ORDER BY + LIMIT fusion
/// for any preceding clause that didn't already absorb top-K. Heap-
/// pruned top-K replaces full sort + truncate.
fn pass_fuse_order_by_top_k(query: &mut CypherQuery, _ctx: &PassCtx) {
    fuse_order_by_top_k(query)
}

/// **Pass:** `reorder_predicates_by_cost` — Within a WHERE clause,
/// reorder predicates by estimated evaluation cost so cheap predicates
/// short-circuit AND/OR chains before expensive ones run.
fn pass_reorder_predicates_by_cost(query: &mut CypherQuery, _ctx: &PassCtx) {
    reorder_predicates_by_cost(query)
}

#[cfg(test)]
#[path = "planner_tests.rs"]
mod tests;

#[cfg(test)]
#[path = "planner_fusion_tests.rs"]
mod fusion_tests;

#[cfg(test)]
mod nested_tests;
#[cfg(test)]
#[path = "with_boundary_tests.rs"]
mod with_boundary_tests;