feat(gfql): execute connected join fast paths

CHANGELOG.md

@@ -39,14 +39,16 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 
 ### Added
 - **GFQL native Polars bindings-row tables (`rows(binding_ops)`) — traversal Cypher on polars (#1709)**: the Cypher multi-alias lowering's `rows(binding_ops=...)` op (one row per matched path) now runs natively on `engine='polars'` for **fixed-length connected patterns** — unblocking traversal-shaped Cypher that previously NIE'd: multi-alias property projections (`MATCH (a)-[e]->(b) RETURN a.x, e.w, b.y`), top-k in-degree (`RETURN b.id, count(a) ... ORDER BY ... LIMIT`, graph-benchmark q1/q2), and fixed multi-hop counts (`MATCH (a)-->(b)-->(c) RETURN count(*)`, q8/q9), with forward/reverse/undirected edges, node/edge filters, and edge-alias payload columns — **plus bounded directed variable-length segments** (`-[*i..k]->`, typed `-[:TYPE*i..k]->`, exactly-k; iterative pair joins with Cypher path multiplicity and zero-hop rows), covering the q3 shape (`MATCH (a:Person)-[:FOLLOWS*1..2]->(b:Person) ... RETURN avg(b.age)`). This rung covers the q1–q4 and q8–q9 binding-table shapes exercised by its parity tests; connected q5–q7 planning/execution is layered in follow-on PRs. Also adds native `with_(extend=True)` (emitted by the bindings-path aggregate lowering) and an honest decline for `group_by(key_prefixes=...)` (whole-row bindings grouping — was a latent silent-wrong-key trap). **Honestly deferred** (`NotImplementedError`, no pandas bridge): unbounded `[*]`, undirected/aliased variable-length, shortestPath scalar bindings, node/edge `query=`/endpoint-match params, cartesian (`MATCH (a),(b)`) mode, seeded re-entry contexts. Differential parity vs the pandas oracle (+20 tests incl. path multiplicity and undirected self-loops; conformance corpus extended). The Polars-GPU engine uses the same lazy target-collected plan; exact-head DGX validation is required before making a GPU performance claim.
+- **GFQL Cypher connected-pattern (q5–q7) planning + execution on all engines**: completes the `rows(binding_ops)` follow-on promised above — multi-alias *connected* `MATCH` patterns that filter and join across shared endpoints (`MATCH (a)-[]->(b)<-[]-(c) WHERE … RETURN …`, star/two-star shapes) now plan and execute natively on pandas, cuDF, `engine='polars'`, and `engine='polars-gpu'` (previously an honest `NotImplementedError` on polars). The connected-join planner pushes eligible single-alias endpoint predicates into per-frame `filter_dict` prefilters when the column dtype makes the pushdown exact, and keeps them as a post-join `where_rows` residual otherwise (nullable/decimal/object/predicate/`AllOf`/`toLower` shapes), so results stay byte-identical to the full-join semantics on every engine. The two-star executor adds grouped-count and bindings-row fast paths with a per-execution result cache (no cross-call staleness under in-place graph mutation) and openCypher-correct ordering — NULL sorts as the largest value (`ORDER BY … ASC` puts nulls last), and `count(a)`/`count(DISTINCT a)`/`count(*)` follow Cypher multiplicity. Differential parity vs the hand-derived openCypher oracle across all four engines (connected-join conformance corpus).
 
 ### Fixed
 - **GFQL seeded indexes now engage for small Polars/cuDF frontiers**: NaN-free native Polars frames retain object identity during normalization, preserving resident CSR validity, while a tunable absolute floor prevents the fractional cost gate from routing tiny seeded hops over low-cardinality slices to a full scan. Scan/index result parity is unchanged; the default floor is a routing heuristic whose engine/data crossover remains subject to measurement.
 - **GFQL Cypher `GRAPH { }` residual predicates now fail safely or apply as graph masks**: `GRAPH { MATCH ... WHERE ... }` no longer silently drops predicates that the graph-state path cannot apply. Safe one-node/one-edge residual filters, including disjunctions and `searchAny(...)`, are applied before graph-state matching; unsupported pattern-predicate, multi-alias, and Polars graph-residual cases now raise clear validation errors instead of returning an over-broad subgraph.
 - **GFQL Cypher `=~` / scalar-fn cross-engine hardening (review wave, dgx-verified)**: (1) composed `=~` (`WHERE … =~ … OR …`, `RETURN`-expression position) now works on `engine='cudf'` — the series evaluator used raw `.str.fullmatch`, which cuDF lacks, and the resulting `AttributeError` was masked as "unsupported predicate op" (it now routes through the `Fullmatch` predicate's engine workarounds, and honest `NotImplementedError` declines pass through instead of being re-labeled); (2) the cuDF fullmatch emulation anchors alternations as a whole (`^(ab|cd)$` — bare `^ab|cd$` silently matched `'abXXX'`); (3) the cuDF case-insensitive `(?i)` lowercase-folding workaround now declines the fold-unsound shapes — uppercase escape classes (`.lower()` turns `\D` into `\d`, silently inverting the predicate), case-crossing character ranges (`(?i)[A-z]` silently narrowed; `[X-b]` folded to an invalid range), and non-ASCII patterns — while lowercase escapes (`\d`, `\.`) keep folding as before; lookaround, backreferences, and named-group refs decline up front (libcudf rejects them at kernel-compile time); (4) polars `Match`/`Fullmatch` lowering applies the same Rust-regex guard as `Contains` (lookaround/backrefs decline instead of a non-NIE `ComputeError` at collect); (5) `toLower`/`toUpper` on a non-string column decline like neo4j's type error instead of broadcasting the stringified Series repr (pandas/cuDF) or raising a non-NIE `SchemaError` (polars-gpu); (6) polars `floor`/`ceil`/`round` cast to `Float64` so integer columns return Float like neo4j and the pandas engine; (7) invalid regex patterns on the composed path raise a clear "invalid regex pattern" error instead of "unsupported predicate op".
 - **GFQL Cypher `round()` hardening (review wave, dgx-verified)**: (1) the `polars` extra's floor is now `polars>=1.29` — `Expr.round(mode=)` shipped in py-1.29.0 (pola-rs/polars#22248), not 1.5 as previously pinned, so 1.5–1.28 installs crashed with a raw `TypeError` on `round(x, p>0)` under `engine='polars'`; (2) `round(x, p>308)` is the identity on both engines (a float64 has no digits there) instead of pandas raising through an unclear decline while polars returned identity — parity restored, `10.0**p` overflow guarded; (3) polars `round(x, p>0)` normalizes `-0.0` to `+0.0` like the pandas kernel (`round(-0.04, 1)` was `0.0` on pandas vs `-0.0` on polars — invisible to value equality, pinned by a sign-bit test); (4) documented the precision>0 decimal-string deviation vs neo4j (`round(2.675, 2)` = `2.67` binary-double here vs `2.68` BigDecimal there) and added deterministic tie/hazard matrix cases so ties actually reach cuDF/polars-gpu (the fixture's normal-distribution floats never tied).
 - **GFQL `materialize_nodes()` on an edges-only graph under `engine='polars'`**: crashed with `AttributeError: 'Series' object has no attribute 'drop_duplicates'` — the node-id derivation used pandas-only `.drop_duplicates()` / `.reset_index()` / `.rename(columns=)` on a polars Series/DataFrame. Now polars-aware (`.unique(maintain_order=True)`, `.select().rename({...})`), matching the pandas oracle. Surfaced by any `let()` DAG on an edges-only graph under Polars (the DAG pre-materializes nodes). pandas/cuDF paths unchanged.
-- **CI: `compute_networkx` HITS test was scipy-version flaky**: `test_compute_networkx_hits_outputs_hubs_and_authorities` used a pure directed 3-cycle (`a→b→c→a`), whose adjacency is a permutation matrix (all singular values equal). networkx HITS computes scores via scipy `svds(k=1)`, which then returns an arbitrary vector from that fully-degenerate singular space; when its components sum to ~0, HITS's `h /= h.sum()` normalization blows up to `inf`/`nan`. Which vector `svds` returns is scipy/LAPACK-version dependent, so the test passed on the pinned scipy but failed on the `nx-upper-scipy` CI profile's newer scipy — a latent flake, not a code regression. Switched to a graph with a well-defined hub/authority structure (`a→b`, `a→c`, `b→c`), whose dominant singular vector is unique (Perron-Frobenius) and therefore version-stable. Test-only change.- **GFQL `materialize_nodes()` on an edges-only graph under `engine='polars'`**: crashed with `AttributeError: 'Series' object has no attribute 'drop_duplicates'` — the node-id derivation used pandas-only `.drop_duplicates()` / `.reset_index()` / `.rename(columns=)` on a polars Series/DataFrame. Now polars-aware (`.unique(maintain_order=True)`, `.select().rename({...})`), matching the pandas oracle. Surfaced by any `let()` DAG on an edges-only graph under Polars (the DAG pre-materializes nodes). pandas/cuDF paths unchanged.- **GFQL `engine='polars-gpu'` LazyFrame-input coercion now collects on the GPU executor**: `Engine.df_to_engine(lazyframe, POLARS_GPU)` materialized a `polars.LazyFrame` input with a bare `.collect()` on the CPU default executor — ignoring `gpu_executor()` and not distinguishing `POLARS` from `POLARS_GPU`. It now routes through the target-aware lazy collect, so a LazyFrame coerced under `polars-gpu` collects on the cudf-polars GPU executor (and `polars` honors `cpu_streaming()`). No change for already-materialized frame inputs (the common path).
+- **CI: `compute_networkx` HITS test was scipy-version flaky**: `test_compute_networkx_hits_outputs_hubs_and_authorities` used a pure directed 3-cycle (`a→b→c→a`), whose adjacency is a permutation matrix (all singular values equal). networkx HITS computes scores via scipy `svds(k=1)`, which then returns an arbitrary vector from that fully-degenerate singular space; when its components sum to ~0, HITS's `h /= h.sum()` normalization blows up to `inf`/`nan`. Which vector `svds` returns is scipy/LAPACK-version dependent, so the test passed on the pinned scipy but failed on the `nx-upper-scipy` CI profile's newer scipy — a latent flake, not a code regression. Switched to a graph with a well-defined hub/authority structure (`a→b`, `a→c`, `b→c`), whose dominant singular vector is unique (Perron-Frobenius) and therefore version-stable. Test-only change.
+- **GFQL `engine='polars-gpu'` LazyFrame-input coercion now collects on the GPU executor**: `Engine.df_to_engine(lazyframe, POLARS_GPU)` materialized a `polars.LazyFrame` input with a bare `.collect()` on the CPU default executor — ignoring `gpu_executor()` and not distinguishing `POLARS` from `POLARS_GPU`. It now routes through the target-aware lazy collect, so a LazyFrame coerced under `polars-gpu` collects on the cudf-polars GPU executor (and `polars` honors `cpu_streaming()`). No change for already-materialized frame inputs (the common path).
 - **GFQL Polars engine `contains` honors `regex=`/`flags=`**: the native polars `filter_by_dict` lowering of the `Contains` predicate always used `str.contains(..., literal=False)`, so a **literal** request (`contains(pat, regex=False)`) was still regex-interpreted — a pattern with a metacharacter over-matched (e.g. `contains('a.c', regex=False)` matched `'abc'`), diverging from pandas/cuDF. It also dropped `flags=`. Now `regex=False` lowers to a literal match (`literal=True`; case-insensitive literal folds both sides, matching pandas' result), and regex mode maps `case=`/`flags=` (IGNORECASE/MULTILINE/DOTALL/VERBOSE) to a Rust-regex inline flag prefix (`(?ims…)`). +1 engine-parametrized differential-parity test.
 - **GFQL adjacency-index cost gate is engine-aware (never slower than scan)**: the seeded-hop planner falls back to a full scan once the frontier covers too large a fraction of the source keys (past that, scanning all edges once beats many index probes). That crossover fraction is *engine-dependent* — a vectorized-scan engine (polars/cuDF/GPU) has a far faster scan, so its crossover is much smaller than pandas'. The gate previously used a single `0.5·n_keys` threshold, so on polars a **resident index under `index_policy='use'` ran ~2× slower than the plain scan** for mid-size frontiers (measured ~frac 0.02–0.5; correct result, just slow). The gate now uses a per-engine crossover (pandas ~0.5, vectorized engines ~0.02; GPU provisional pending large-graph measurement), so a resident index never loses to the un-indexed path on any engine. Small-frontier wins (the point of the index) are unchanged. +1 engine-parametrized regression test.
 - **GFQL `ne()` / `<>` on NULL now follows openCypher/SQL 3-valued logic (pandas)**: `n({"col": ne(x)})` and cypher `WHERE n.col <> x` over a NULL/NA cell used to KEEP the null row on the pandas engine (`NaN != x` → True), diverging from cuDF and the polars engine (both drop it) — and even from pandas' own `WHERE NOT n.col = x` path. Per openCypher/SQL three-valued logic, `null <> x` is `null` (an unknown value cannot be proven unequal to `x`), so a null cell is **not** a match and the row is excluded — consistent with `eq`/`gt`/`lt`/`IN` (which already dropped nulls). Fixed the `NE` predicate to mask out nulls; this corrects both the `filter_dict` predicate path and the single-entity cypher `<>` WHERE path on pandas. cuDF/polars/polars-gpu were already conformant. Verified across all four engines (`ne`, `<>`, `NOT =`, `NOT IN` all drop the null). Note: this is a behavior change for `ne()` on nullable columns under the default pandas engine. (Broader openCypher null-semantics alignment + docs tracked in #1664.)

bin/ci_cypher_surface_guard_baseline.json

@@ -13,5 +13,5 @@
       "max_properties": 0
     }
   },
-  "lowering_py_max_lines": 9237
+  "lowering_py_max_lines": 9244
 }

graphistry/compute/dataframe/join.py

@@ -3,7 +3,7 @@
 import operator
 from typing import Any, Dict, List, Optional, Sequence, Tuple, cast
 
-from graphistry.Engine import Engine
+from graphistry.Engine import Engine, POLARS_ENGINES
 from graphistry.compute.typing import DataFrameT, DomainT
 
 
@@ -20,9 +20,15 @@ def joined_hidden_scalar_columns(frame: DataFrameT) -> DataFrameT:
         if suffix.startswith("__cypher_reentry_") or suffix.startswith("__gfql_hidden_"):
             hidden_suffixes.setdefault(suffix, []).append(column)
     out = frame
+    is_polars = "polars" in type(frame).__module__
     for suffix, columns in hidden_suffixes.items():
         if suffix in out.columns:
             continue
+        if is_polars:
+            import polars as pl
+            expr = pl.coalesce([pl.col(column) for column in columns]).alias(suffix)
+            out = out.with_columns(expr)
+            continue
         series = out[columns[0]]
         for column in columns[1:]:
             if hasattr(series, "combine_first"):
@@ -44,6 +50,13 @@ def joined_alias_columns(frame: DataFrameT) -> DataFrameT:
         elif suffix == "id" and alias not in alias_candidates:
             alias_candidates[alias] = column
     out = frame
+    is_polars = "polars" in type(frame).__module__
+    if is_polars and alias_candidates:
+        import polars as pl
+        return cast(DataFrameT, out.with_columns([
+            pl.col(source_column).alias(alias)
+            for alias, source_column in alias_candidates.items()
+        ]))
     for alias, source_column in alias_candidates.items():
         out = out.assign(**{alias: out[source_column]})
     return out
@@ -65,6 +78,8 @@ def connected_inner_join_rows(
     join_cols_list = list(join_cols)
     keep_cols_list = list(keep_cols)
     rhs = cast(DataFrameT, pattern_rows[keep_cols_list])
+    if engine in POLARS_ENGINES:
+        return cast(DataFrameT, joined_rows.join(rhs, on=join_cols_list, how="inner"))
     if engine != Engine.CUDF:
         return cast(DataFrameT, joined_rows.merge(rhs, on=join_cols_list, how="inner"))
 

graphistry/compute/gfql/cypher/lowering.py

@@ -8209,6 +8209,13 @@ def _connected_join_required_property_aliases(
             return None
         required.update(alias for alias in non_aggregate_aliases if alias in node_aliases)
         required.update(alias for alias in prop_aliases if alias in node_aliases)
+        # A bare node alias used *inside* an aggregate (e.g. ``count(p)`` /
+        # ``count(DISTINCT p)``) is rewritten by _connected_join_alias_identity_expr to
+        # ``p.__gfql_node_id__``, so its identity column must still be attached even though
+        # it is neither a non-aggregate use nor a property access. Attaching by alias here
+        # is a no-op when the alias is already required via a property access (e.g.
+        # ``count(p.age)``); it only adds the otherwise-missing bare-aggregate case.
+        required.update(alias for alias in _aggregate_aliases if alias in node_aliases)
     return required