Merge pull request #1762 from graphistry/refactor/gfql-seeded-fastpath-module

refactor(gfql): extract seeded fast-path specializations to dedicated modules (#1755)

Author: lmeyerov

CHANGELOG.md

@@ -66,6 +66,7 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 - **GFQL `engine='polars'`/`'polars-gpu'` can now run off-engine analytic `call()`s (umap/hypergraph/compute_cugraph/…) — `call_mode='auto'` (default)**: a GFQL `call()` that invokes a Plottable-method analytic with **no native polars implementation** (and never will — `umap`, `hypergraph`, `compute_cugraph`, `compute_igraph`, `layout_*`, `collapse`, `get_topological_levels`, …) previously raised `NotImplementedError` under a polars engine, forcing users off polars for the whole pipeline. It now runs as a **mode-gated, warned modality switch**: `call_mode='auto'` (default) bridges the graph off-engine (`polars`→pandas, `polars-gpu`→cuDF **on-device**), runs the analytic, and coerces the result back to polars **losslessly via Arrow**, warning once per (process, function). `call_mode='strict'` keeps the honest `NotImplementedError` decline (for benchmark integrity / a hard memory ceiling). `polars-gpu` is **GPU-or-error**: it bridges to cuDF and declines rather than silently dropping a GPU analytic to host pandas. This is **deliberately narrower than CHAIN traversal/filter/row ops**, which stay parity-or-NIE (a bridge there would hide a missing impl and cheat a benchmark) — the split is mechanical (`is_row_pipeline_call`), and the chain and DAG surfaces bridge consistently. Configurable via `graphistry.compute.gfql.lazy.set_call_mode('auto'|'strict')` (Python) or `GFQL_POLARS_CALL_MODE` (env), resolving Python override > env > default('auto'), read live. Verified on dgx: `compute_cugraph` PageRank is byte-parity with the pandas oracle under both `polars` and `polars-gpu`.
 
 ### Changed
+- **GFQL seeded fast-path specializations extracted to dedicated modules (#1755)**: pure move, no behavior change. The seeded typed-hop helpers (`_seeded_typed_hop_pandas_cudf`, `_seeded_typed_return_dst_pandas_cudf`, `_seeded_typed_return_dst_polars`) move from `chain.py` to a new `graphistry/compute/chain_fast_paths.py`, and the cypher-side dispatcher (`_execute_seeded_typed_hop_fast_path`) moves from `gfql_unified.py` to `gfql_fast_paths.py`, following the one-directional-import module-extraction convention (#1731): `chain_fast_paths.py` imports only leaves (`.ast`, `.typing`, `Plottable`), and neither fast-path module imports back into its executor. Keeps `chain.py`/`gfql_unified.py` focused on the general executors while specializations accumulate in their own files.
 - **GFQL Cypher parser: internal cleanup — WHERE consumed directly from `MatchClause`**: the grammar bundles a trailing `WHERE` onto its `MATCH` clause; the transformer previously split it back out into a synthetic standalone item and re-attached it, so the legacy clause-assembler ran unchanged (a temporary seam that kept the LALR switch byte-identical). The assembler now consumes `MatchClause.where` directly (primary MATCH keeps its WHERE on the clause; a post-WITH re-entry MATCH's WHERE goes to `reentry_wheres`), deleting the split/re-attach round-trip and the now-unreachable standalone-WHERE handling in both `query_body` and `graph_constructor`. Pure internal refactor, **no behavior change**: verified byte-identical ASTs vs the prior parser across a 1,989-query repo corpus, and the full cypher suite passes.
 - **GFQL Cypher parser: Earley → LALR(1) (~70× faster parse, same AST)**: `parse_cypher` now uses a single LALR(1) parser (contextual lexer) instead of Earley (dynamic lexer) — ~0.25ms vs ~17ms per query on representative queries. The WHERE grammar was unified to one `where_clause: "WHERE" expr` rule (the former `where_predicates | expr` dual rule was a genuine reduce/reduce ambiguity that forced Earley), and the structured flat-AND predicate form is recovered by a post-parse lift that re-parses the WHERE body with a LALR sub-parser (`start="where_predicate_chain"`). The full WHERE language is preserved — OR/XOR/NOT, parentheses, arithmetic, IN, pattern predicates — nothing is restricted to a subset. The grammar was then **purified to (near-)unambiguous**: WHERE binds to its preceding clause *in the grammar* (bundled into `match_clause`; the WITH..WHERE attachment ambiguity is eliminated, not tie-broken), and name-rooted dot chains derive only via `qualified_name` (the `property_access` redundancy is gone) — dropping the redundant `label_predicate_expr` rule (`(n:Admin)` parses via `grouped_expr`), and excluding a top-level `IN` from list-literal elements (`[x IN xs ...` is comprehension syntax; allowing a bare `IN` list element made it overlap). Net: the LALR conflict profile goes from 8 shift/reduce to **ZERO conflicts** — the grammar is now **provably unambiguous LALR(1)** and builds under Lark's `strict=True` (a build-time proof, machine-checked in CI as zero conflicts, plus a strict-mode build test where the optional `interegular` dep is present). Every input has a single derivation. **Machine-checked invariants** (`test_grammar_invariants.py`): (1) **zero LALR conflicts** (dependency-free, always-on in CI) plus a `strict=True` build test (skipped where the optional `interegular` dep is absent) — a grammar edit that introduces any ambiguity fails CI; (2) semantic ambiguity is ZERO — every corpus query has exactly one Earley derivation and it transforms to a single AST, no exceptions; (3) a rule-coverage gate (`test_every_grammar_rule_is_exercised_by_the_corpus`) forces every new grammar rule through the invariants; (4) differential tests run the production pipeline under both parsers (byte-identical ASTs, identical rejections). **Full-repo differentials** (1,850+ scraped queries, LALR-vs-Earley and old-vs-new production): **zero AST divergences**, and a small set of deliberate language fixes — accept-by-accident shapes with ill-defined semantics, now honest syntax errors and pinned as tests (`DELIBERATE_LANGUAGE_FIXES`): `RETURN DISTINCT` with no items (Earley re-lexed DISTINCT as a variable name), double WHERE (the old positional attachment kept *both* predicates in different AST fields), double post-WITH WHERE, WHERE after UNWIND, WHERE before MATCH in a graph constructor, and the invalid "list of IN-booleans" (`[x IN xs, y]`; use `[(x IN xs), y]` for a genuine bool list). The lift stays an internal optimization: only a flat `AND` chain over present columns lifts to `filter_dict`; parens / OR / XOR / NOT and any absent-column case stay on `where_rows` (a correctness boundary, since `where_rows` treats an absent property as null while `filter_dict` would raise). Full cypher suite (1,681 tests) passes. Downstream execution (`filter_dict` vs `where_rows` routing) is unchanged. A pre-existing `<>`-over-null 3VL divergence between the two execution paths (surfaced by #1653's metamorphic test) is tracked separately in #1683.
 

graphistry/compute/chain.py

@@ -11,6 +11,7 @@
 from .ast import ASTObject, ASTNode, ASTEdge, Direction, from_json as ASTObject_from_json, serialize_binding_ops
 from .typing import DataFrameT, SeriesT
 from .util import generate_safe_column_name
+from .chain_fast_paths import _seeded_typed_hop_pandas_cudf
 from graphistry.compute.validate.validate_schema import validate_chain_schema
 from graphistry.compute.gfql.same_path_types import (
     WhereComparison,
@@ -684,190 +685,6 @@ def _chain_otel_attrs(
     return attrs
 
 
-def _seeded_scalar_filters(fd: Optional[Dict[str, Any]], df: DataFrameT) -> Optional[Dict[str, Any]]:
-    """Resolve a filter dict to plain scalar column==value pairs, or None to bail
-    to the general path. Mirrors filter_by_dict.resolve_filter_column exactly for
-    the shapes it accepts: the cypher ``label__X: True`` form maps to ``type``
-    equality ONLY when no list-valued ``labels`` column exists (labels-containment
-    is not scalar equality) and the frame is not edge-shaped — same precedence as
-    the live resolver. Anything else (predicates, non-scalar values, absent
-    columns) bails, so the full path keeps its exact semantics incl. E301."""
-    from graphistry.compute.filter_by_dict import _looks_like_edge_dataframe
-    if not fd:
-        return {}
-    cols = set(df.columns)
-    out: Dict[str, Any] = {}
-    for k, v in fd.items():
-        if not isinstance(v, (int, float, str, bool)):
-            return None  # predicate / non-scalar -> bail to the general path
-        if k in cols:
-            out[k] = v
-        elif (isinstance(k, str) and k.startswith("label__") and v is True
-              and "labels" not in cols and "type" in cols
-              and not _looks_like_edge_dataframe(df)):
-            out["type"] = k[len("label__"):]
-        else:
-            return None  # labels-list / unknown column -> bail
-    return out
-
-
-def _seeded_typed_hop_pandas_cudf(
-    g: Plottable, n0: ASTNode, n2: ASTNode, e1: ASTEdge,
-    src: str, dst: str, node: str, direction: Direction,
-) -> Optional[Plottable]:
-    """#1755 lever-3: engine-generic (pandas + cuDF) fast path for a scalar-filtered
-    seeded typed 1-hop. Value-identical to the general seeded branch for the covered
-    shape (all node/edge filters are plain scalars, directed) — same rows, columns,
-    and dtypes; row order and RangeIndex may differ — collapsing it into a
-    few DataFrame filters so a seeded lookup lands sub-ms. Uses only the shared
-    pandas/cuDF DataFrame API (no numpy array drops) so the same body runs on both
-    engines. Returns None to fall back for anything it does not cover (predicates,
-    undirected, missing columns) — the caller then runs the general branch."""
-    if direction == "undirected":
-        return None
-
-    nodes_df, edges_df = g._nodes, g._edges
-    if nodes_df is None or edges_df is None:
-        return None
-    n0f = _seeded_scalar_filters(n0.filter_dict, nodes_df)
-    n2f = _seeded_scalar_filters(n2.filter_dict, nodes_df)
-    ef = _seeded_scalar_filters(e1.edge_match, edges_df)
-    if n0f is None or n2f is None or ef is None:
-        return None
-    from_col, to_col = (src, dst) if direction == "forward" else (dst, src)
-
-    # from-side seed FIRST: reduce edges to the seed's out-edges before the
-    # edge_match compare, so the type filter runs on the tiny frontier rather than
-    # all edges — this is what makes a seeded lookup sub-ms. The id filter goes
-    # first (int, unique -> ~1 row in one pass) so any remaining object filters
-    # (label__X->type) run on that tiny survivor frame, not the whole node table.
-    if n0f:
-        seed_nodes = nodes_df
-        for k, v in sorted(n0f.items(), key=lambda kv: 0 if kv[0] == node else 1):
-            seed_nodes = seed_nodes[seed_nodes[k] == v]
-        edges = edges_df[edges_df[from_col].isin(seed_nodes[node].dropna())]
-    else:
-        edges = edges_df
-    if ef:  # typed edge (edge_match) — now on the reduced frontier
-        for k, v in ef.items():
-            edges = edges[edges[k] == v]
-
-    # Gather candidate endpoint nodes (both endpoints of surviving edges), then run
-    # the dest filter, dangling-edge drop and final-node selection on the small
-    # candidate/edge frames. Selecting from nodes_df keeps only real nodes, so the
-    # endpoint-in-nodes check subsumes the old NaN-endpoint guard. Membership sets
-    # are dropna()'d: pandas .isin matches NaN<->NaN, but the general branch's BFS
-    # joins never join on null keys, so a null id/endpoint must not link.
-    cand = nodes_df[
-        nodes_df[node].isin(edges[src].dropna()) | nodes_df[node].isin(edges[dst].dropna())
-    ].drop_duplicates(subset=[node])
-    if n2f:  # destination-node filter (to-side)
-        n2_cand = cand
-        for k, v in n2f.items():
-            n2_cand = n2_cand[n2_cand[k] == v]
-        n2_ok = n2_cand[node]
-    else:
-        n2_ok = cand[node]
-    to_vals = edges[to_col]
-    keep = edges[src].isin(cand[node].dropna()) & edges[dst].isin(cand[node].dropna()) & to_vals.isin(n2_ok.dropna())
-    edges = edges[keep]
-    cand = cand[cand[node].isin(edges[src]) | cand[node].isin(edges[dst])]
-    return g.nodes(cand).edges(edges)
-
-
-def _seeded_typed_return_dst_pandas_cudf(
-    g: Plottable, n0: ASTNode, n2: ASTNode, e1: ASTEdge,
-    src: str, dst: str, node: str, direction: Direction,
-) -> Optional[Tuple[DataFrameT, DataFrameT]]:
-    """#1755 cypher RETURN-alias fast path: like _seeded_typed_hop_pandas_cudf but
-    returns ONLY the destination (RETURN-alias) node rows + surviving edges — no
-    seed-node gather, no Plottable round-trip — so the seeded cypher projection
-    lands sub-ms. Engine-generic (pandas + cuDF): only the shared DataFrame API,
-    no numpy array drops. Returns ``(dst_node_rows, edges)`` or None to fall back."""
-    if direction == "undirected":
-        return None
-    nodes_df, edges_df = g._nodes, g._edges
-    if nodes_df is None or edges_df is None:
-        return None
-    n0f = _seeded_scalar_filters(n0.filter_dict, nodes_df)
-    n2f = _seeded_scalar_filters(n2.filter_dict, nodes_df)
-    ef = _seeded_scalar_filters(e1.edge_match, edges_df)
-    if n0f is None or n2f is None or ef is None or not n0f:
-        return None
-    from_col, to_col = (src, dst) if direction == "forward" else (dst, src)
-    # id-first seed reduction: filter by the id column first (int/unique -> ~1 row)
-    # so any remaining object filters (label__X->type) run on the tiny survivor
-    # frame, never materializing an object column over the whole node table.
-    # Membership sets are dropna()'d: pandas .isin matches NaN<->NaN, but the full
-    # pipeline's joins never join on null keys, so a null id/endpoint must not link.
-    seed_nodes = nodes_df
-    for k, v in sorted(n0f.items(), key=lambda kv: 0 if kv[0] == node else 1):
-        seed_nodes = seed_nodes[seed_nodes[k] == v]
-    edges = edges_df[edges_df[from_col].isin(seed_nodes[node].dropna())]
-    if ef:
-        for k, v in ef.items():
-            edges = edges[edges[k] == v]
-    # destination nodes = real nodes that are edge to-endpoints, then the dest
-    # filter, dangling-edge drop and dedup on the small dst/edge frames.
-    dstn = nodes_df[nodes_df[node].isin(edges[to_col].dropna())]
-    if n2f:
-        for k, v in n2f.items():
-            dstn = dstn[dstn[k] == v]
-    edges = edges[edges[to_col].isin(dstn[node].dropna())]
-    dstn = dstn[dstn[node].isin(edges[to_col].dropna())].drop_duplicates(subset=[node])
-    return dstn, edges
-
-
-def _seeded_typed_return_dst_polars(
-    g: Plottable, n0: ASTNode, n2: ASTNode, e1: ASTEdge,
-    src: str, dst: str, node: str, direction: Direction,
-) -> Optional[Tuple[DataFrameT, DataFrameT]]:
-    """#1755 polars analog of _seeded_typed_return_dst_pandas_cudf: same seed-first
-    reduction (seed out-edges -> typed-edge filter -> destination nodes) expressed
-    with polars filters, so a seeded cypher RETURN on polars/polars-gpu also lands
-    sub-ms. Returns ``(dst_node_rows, edges)`` (polars frames) or None to fall back
-    to the full lazy pipeline. Value-identical node set to the full path for the
-    covered shape (scalar filters, directed, single hop); row order may differ."""
-    import polars as pl
-    if direction == "undirected":
-        return None
-    nodes_df, edges_df = g._nodes, g._edges
-    # Eager polars frames only: LazyFrame has no get_column, and mixed-engine
-    # node/edge frames must take the full path — decline rather than crash.
-    if not isinstance(nodes_df, pl.DataFrame) or not isinstance(edges_df, pl.DataFrame):
-        return None
-
-    n0f = _seeded_scalar_filters(n0.filter_dict, nodes_df)
-    n2f = _seeded_scalar_filters(n2.filter_dict, nodes_df)
-    ef = _seeded_scalar_filters(e1.edge_match, edges_df)
-    if n0f is None or n2f is None or ef is None or not n0f:
-        return None
-    from_col, to_col = (src, dst) if direction == "forward" else (dst, src)
-
-    # from-side seed: reduce the node frame to the seed rows, take their ids.
-    # Membership sets are drop_nulls()'d (null ids/endpoints never link, matching
-    # the full pipeline's joins) and passed via .implode() (Series-arg is_in is
-    # deprecated in polars 1.42, see polars#22149).
-    seed_nodes = nodes_df
-    for k, v in n0f.items():
-        seed_nodes = seed_nodes.filter(pl.col(k) == v)
-    from_ids = seed_nodes.get_column(node).drop_nulls()
-    if from_ids.len() == 0:
-        return nodes_df.clear(), edges_df.clear()
-    edges = edges_df.filter(pl.col(from_col).is_in(from_ids.implode()))
-    for k, v in ef.items():  # typed edge on the reduced frontier
-        edges = edges.filter(pl.col(k) == v)
-    dst_ids = edges.get_column(to_col).drop_nulls().unique()
-    dstn = nodes_df.filter(pl.col(node).is_in(dst_ids.implode()))
-    for k, v in n2f.items():  # destination-node filter
-        dstn = dstn.filter(pl.col(k) == v)
-    # drop dangling edges + dedup destination nodes (mirror the pandas tail)
-    keep_ids = dstn.get_column(node).drop_nulls()
-    edges = edges.filter(pl.col(to_col).is_in(keep_ids.implode()))
-    dstn = dstn.filter(pl.col(node).is_in(edges.get_column(to_col).implode())).unique(subset=[node], maintain_order=True)
-    return dstn, edges
-
-
 def _try_chain_fast_path(
     g_in: Plottable,
     ops: List[ASTObject],