Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -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.

Expand Down
185 changes: 1 addition & 184 deletions graphistry/compute/chain.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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],
Expand Down
Loading
Loading