feat(gfql): polars-gpu = GPU target of the lazy Polars engine (collect-once) [redo of #1654]

CHANGELOG.md

@@ -11,8 +11,13 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 ### Added
 - **GFQL native Polars engine — traversals (`engine='polars'`)**: Added a native, vectorized Polars execution engine for the core GFQL traversals `hop()` and `chain()`, dispatched at the engine boundary so the production pandas/cuDF paths are untouched. `Engine.POLARS` is opt-in (explicit `engine='polars'`); `engine='auto'` with Polars input still coerces to pandas as before. Covers forward/reverse/undirected single-hop traversal, directed multi-hop chains, node/edge filter dicts and predicates (lowered to Polars expressions), `edge_match`/`source_node_match`/`destination_node_match`, `target_wave_front`, and alias names; the BFS advances via semi/anti joins (no per-row Python work). Validated by differential parity against the pandas engine (hop + chain test suites plus a randomized fuzzer) and benchmarked vs pandas (`benchmarks/gfql/pandas_vs_polars.py`) — Polars wins at scale (up to ~2.5x on multi-edge chains at millions of edges; crossover ~50–100k rows). Variable-length/multi-hop edges, undirected edges in multi-edge chains, hop labels, and node `query=` raise `NotImplementedError` for now (use `engine='pandas'`).
 - **GFQL native Polars engine — cypher row pipeline (`engine='polars'`)**: Extended the Polars engine to the Cypher `MATCH … RETURN` row surface, natively vectorized. **NO CHEATING:** the polars engine never silently falls back to the pandas engine — every query runs natively on polars or raises an honest `NotImplementedError` pointing at `engine='pandas'` (falling back to pandas would misrepresent pandas performance as polars; only a human may consent to a bridge). `chain_polars` splits boundary `call()` ops (mirroring the pandas `_handle_boundary_calls`) and runs each trailing row op per-op native or raises. **Native polars** (no pandas round-trip): frame ops (`rows`/`limit`/`skip`/`distinct`/`drop_cols`), `select`/`with_`/`return_` projection (a conservative cypher-expr-AST → `pl.Expr` lowering covering property access, arithmetic, comparison, boolean, literals), `order_by` (`.sort`), `group_by` (`count`/`sum`/`avg`/`min`/`max`), `unwind` (literal-list cross-join), the result projection for property/expr columns, and entity-text `RETURN n` rendering for int/string/bool nodes (`pl.concat_str`). **Honestly deferred** (raise `NotImplementedError`, no pandas fallback): multi-entity `rows(binding_ops=…)`, cross-entity same-path `WHERE` (`DFSamePathExecutor`), float/temporal/nested entity-text, and exotic expressions (CASE/list/map/temporal, `collect` aggregates) — these are the forward native-engineering targets. Validated by differential parity vs pandas including a TCK-style conformance lane (`test_engine_polars_cypher_conformance.py`: native-only curated corpus + seeded fuzzer + NULL/3-valued-logic graph + entity-text escaping, plus a `DEFERRED` list asserting deferred queries raise rather than silently bridge) and benchmarked (`benchmarks/gfql/cypher_row_pipeline.py`). **Perf (interleaved, 1M nodes, each engine on its native-frame graph, all fully native):** polars wins **5.6–38×** across the surface — `RETURN n` ~38×, `ORDER BY` ~17×, `WHERE`+`ORDER BY`+`LIMIT` ~14×, traversals 6–7.5×, projections/aggregations/`DISTINCT` 5.6–6.9×. cuDF/pandas paths untouched.
+- **GFQL lazy Polars engine + GPU target (`engine='polars-gpu'`, cudf_polars)**: The Polars traversal engine now builds a single deferred `pl.LazyFrame` plan per single-hop and materializes `out_edges`+`out_nodes` in ONE `collect_all` on a chosen **execution target** (CPU or GPU). `engine='polars-gpu'` (`Engine.POLARS_GPU`, explicit opt-in only — AUTO never selects it) runs that same lazy plan on the RAPIDS cudf_polars backend (`pl.GPUEngine(raise_on_fail=True)` — NO-CHEATING: a GPU-incapable plan node **raises** rather than silently running on CPU and being reported as a GPU result; see Fixed). The collect-once design is what makes GPU pay off: a benchmark showed per-op eager GPU collect was a *regression* (repeated H2D), while collect-once is a **2.84× single-hop GPU win @1M** with CPU parity. Frames stay `pl.DataFrame` (handled like `POLARS` everywhere); the target is carried by a context var set at the chain/hop dispatch boundary, so `engine='polars'` (CPU) is byte-for-byte unchanged. Validated by differential parity `engine='polars-gpu' == engine='polars'` across the cypher conformance corpus + traversals (`test_engine_polars_gpu.py`, skips when no cudf_polars/GPU). Multi-hop and the chain forward/backward fusion (where the GPU win currently dilutes) are follow-up optimizations.
+
+### Added
+- **GFQL Polars-CPU streaming collect (opt-in, large traversals)**: `GFQL_POLARS_CPU_STREAMING=1` runs the polars-CPU lazy collects (`hop`/`chain`) on the polars **streaming** executor instead of the default in-memory collect. Benchmarked ~1.04–1.11× faster on big multi-hop traversals (10M nodes / 80M edges: 20.0→18.0 s) and parity-identical, but ~0.86× (slower) on small/interactive sizes (streaming overhead) — so it is **opt-in, default off** (no change to default behavior). Use for large batch traversals where CPU is the target.
 
 ### Fixed
+- **GFQL `engine='polars-gpu'` silent CPU fallback removed (NO-CHEATING)**: the GPU collect used `pl.GPUEngine(raise_on_fail=False)`, so any plan node the cudf_polars backend can't execute would silently run **on CPU** and still be reported as a `polars-gpu` result — making `engine='polars-gpu'` indistinguishable from `engine='polars'` whenever the plan isn't fully GPU-capable (a benchmark showing near-identical `polars`/`polars-gpu` timings is exactly this tell). Flipped to `raise_on_fail=True` and translate the cudf_polars failure into a clear `NotImplementedError` pointing at `engine='polars'` for native CPU. `polars-gpu` is now **GPU-or-error**: any timing it produces is real on-device work, never CPU mislabeled as GPU. Verified on dgx-spark (LiveJournal 35M): the seeded-frontier `hop`/2-hop chain plan executes fully on GPU without raising (nvidia-smi 92% util during the loop), so existing GPU timings are unchanged — only the honesty guarantee is added. +1 regression test (the GPU-collect error path is translated, not swallowed).
 - **GFQL Polars chain dtype-mismatched join keys**: a chain (`MATCH (a)-[]->(b)…`) crashed under `engine='polars'` with `SchemaError: datatypes of join keys don't match` when an edge endpoint column's dtype differed from the node-id dtype across the int↔float boundary — e.g. a null in a `source`/`destination` column promotes it to `float64` while integer node ids stay `int64`. The hop already aligned join keys; the chain fast paths + combine did not. Now aligns endpoint join keys to the node-id dtype for the traversal and restores the original endpoint dtype on the output (matching pandas). No-op when dtypes already match.
 - **GFQL `chain()` OTel span placement**: the `gfql.chain` OpenTelemetry span (`@otel_traced`) had landed on the internal `_try_chain_fast_path` probe instead of the public `chain()` — so `chain()` lost its span and the span recorded the wrong function/attributes. Moved the decorator back onto `chain()`.
 - **GFQL Polars engine adversarial-review correctness fixes (#1648)**: a multi-dimension adversarial review of the native polars engine found three reachable divergences from the pandas oracle, now fixed (NO-CHEATING — match pandas or decline honestly): (1) **duplicate alias** — a chain reusing an alias (`[n(name='a'), e(), n(name='a')]`) returned a malformed colliding-join schema (`a`/`a_right`) instead of raising; now raises the same `GFQLValidationError` E201 as pandas. (2) **integer-literal division** — `5/2` lowered to polars true division (`2.5`) but Cypher folds it to integer division (`2`), a silent wrong order when embedded non-monotonically (`ORDER BY n.val % (10/4)`); now declines (`NotImplementedError`). Column `/` int (Float on both engines) is unaffected. (3) **chain seed dtype** — an internal `start_nodes` seed whose id-column dtype diverged from the node-id dtype (e.g. an empty crossfilter selection defaulting to float64 vs int64 node ids) crashed the combine join with `SchemaError`; now aligns the seed join key (mirroring the hop). Also removed stale "pandas bridge" docstrings (the bridge was removed earlier), DRY-consolidated the cross-type/NaN dtype classifiers into `engine_polars/dtypes.py`, and documented a narrow `filter_by_dict` genuine-NaN residual (unreachable on the `from_pandas` ingestion path). +3 regression tests.

graphistry/Engine.py

@@ -14,24 +14,29 @@ class Engine(Enum):
     DASK = 'dask'
     DASK_CUDF = 'dask_cudf'
     POLARS = 'polars'
+    # GPU execution TARGET of the lazy Polars engine (cudf_polars): frames stay
+    # ``pl.DataFrame`` (handled exactly like POLARS in all frame ops); only the
+    # lazy ``.collect()`` runs on GPU. Explicit opt-in only — AUTO never selects it.
+    POLARS_GPU = 'polars-gpu'
 
 # Engines whose frames use the polars API (unique/with_columns/...) rather than the
-# pandas API (drop_duplicates/assign/...). The polars-GPU target (Engine.POLARS_GPU)
-# extends this tuple where it is introduced (the lazy GPU engine).
-POLARS_ENGINES = (Engine.POLARS,)
+# pandas API (drop_duplicates/assign/...). POLARS_GPU is the GPU execution target of
+# the same lazy Polars engine — frames stay ``pl.DataFrame``, so it shares the path.
+POLARS_ENGINES = (Engine.POLARS, Engine.POLARS_GPU)
 
 class EngineAbstract(Enum):
     PANDAS = Engine.PANDAS.value
     CUDF = Engine.CUDF.value
     DASK = Engine.DASK.value
     DASK_CUDF = Engine.DASK_CUDF.value
     POLARS = Engine.POLARS.value
+    POLARS_GPU = Engine.POLARS_GPU.value
     AUTO = 'auto'
 
 
 # Type alias for engine parameter - accepts both enum values and string literals
 # Includes 'auto' for automatic detection
-EngineAbstractType = Union[EngineAbstract, Literal['pandas', 'cudf', 'dask', 'dask_cudf', 'polars', 'auto']]
+EngineAbstractType = Union[EngineAbstract, Literal['pandas', 'cudf', 'dask', 'dask_cudf', 'polars', 'polars-gpu', 'auto']]
 
 DataframeLike = Any  # pdf, cudf, ddf, dgdf
 DataframeLocalLike = Any  # pdf, cudf
@@ -216,7 +221,7 @@ def df_to_engine(df, engine: Engine):
         if not isinstance(df, pd.DataFrame):
             df = df_to_engine(df, Engine.PANDAS)
         return dd.from_pandas(df, npartitions=1)
-    elif engine == Engine.POLARS:
+    elif engine in POLARS_ENGINES:
         import polars as pl
         if isinstance(df, pl.DataFrame):
             return df
@@ -297,7 +302,7 @@ def df_concat(engine: Engine):
     elif engine == Engine.CUDF:
         import cudf
         return cudf.concat
-    elif engine == Engine.POLARS:
+    elif engine in POLARS_ENGINES:
         return _pl_concat
     elif engine == Engine.DASK:
         raise NotImplementedError("DASK is an input format, not a compute engine — use engine='auto' or engine='pandas'")
@@ -372,7 +377,7 @@ def df_cons(engine: Engine):
     elif engine == Engine.CUDF:
         import cudf
         return cudf.DataFrame
-    elif engine == Engine.POLARS:
+    elif engine in POLARS_ENGINES:
         import polars as pl
         return pl.DataFrame
     raise ValueError(f'Only engines pandas/cudf supported, got: {engine}')
@@ -395,7 +400,7 @@ def s_cons(engine: Engine):
     elif engine == Engine.CUDF:
         import cudf
         return cudf.Series
-    elif engine == Engine.POLARS:
+    elif engine in POLARS_ENGINES:
         import polars as pl
         return pl.Series
     raise ValueError(f'Only engines pandas/cudf supported, got: {engine}')

graphistry/compute/ComputeMixin.py

@@ -86,7 +86,7 @@ def _is_already_correct(df: Any) -> bool:
             if not has_cudf:
                 return True  # cudf unavailable — skip coercion; downstream handles gracefully
             return 'cudf' in type_mod
-        elif engine == Engine.POLARS:
+        elif engine in (Engine.POLARS, Engine.POLARS_GPU):
             return 'polars' in type_mod
         return True
 

graphistry/compute/chain.py

@@ -799,17 +799,21 @@ def chain(
     engine_concrete_early = resolve_engine(engine, self)
     self = _coerce_input_formats(self, engine_concrete_early)
 
-    if engine_concrete_early == Engine.POLARS:
+    if engine_concrete_early in (Engine.POLARS, Engine.POLARS_GPU):
         # Native polars chain lives in a dedicated dispatched module so the
         # production pandas/cuDF orchestration below stays untouched (see
         # plans/gfql-polars-engine). Correctness gated by differential parity.
+        # POLARS_GPU = the same lazy engine with the GPU execution target.
         if validate_schema:
             Chain(ops if not isinstance(ops, Chain) else ops.chain).validate(collect_all=False)
         from graphistry.compute.gfql.lazy.engine.polars.chain import chain_polars
+        from graphistry.compute.gfql.lazy import target_mode, ExecutionTarget
         # NO pandas fallback here (see plan.md NO-CHEATING): chain_polars raises
         # NotImplementedError for deferred features (var-length/multi-hop edges,
         # undirected multi-edge); that honest signal propagates to the caller.
-        return chain_polars(self, ops, start_nodes=start_nodes)
+        _tgt = ExecutionTarget.GPU if engine_concrete_early == Engine.POLARS_GPU else ExecutionTarget.CPU
+        with target_mode(_tgt):
+            return chain_polars(self, ops, start_nodes=start_nodes)
 
     if policy:
         from graphistry.compute.gfql.call.executor import _thread_local as call_thread_local

graphistry/compute/gfql/lazy/__init__.py

@@ -53,6 +53,16 @@ def __exit__(self, *exc: Any) -> None:
             _TARGET.reset(self._token)
 
 
+import os as _os
+
+# CPU collect engine. The polars STREAMING executor benchmarks faster + more stable
+# than the default collect on big traversal joins (isolated 80M-edge 2-hop semijoin
+# 1669→1040 ms, ~1.6×; end-to-end chain dilutes to ~1.04–1.11× as the forward/backward/
+# combine overhead is unaffected), parity-identical. Opt-in (default off) because
+# small/interactive sizes REGRESS (~0.86× at 100K) from streaming overhead.
+_CPU_STREAMING = _os.environ.get("GFQL_POLARS_CPU_STREAMING", "0") == "1"
+
+
 def _engine_for(target: ExecutionTarget) -> Any:
     """Polars collect engine for a target. ``None`` = default (CPU streaming/in-mem).
 
@@ -61,20 +71,46 @@ def _engine_for(target: ExecutionTarget) -> Any:
     in device memory — the regime the in-memory engine is built for — and it is
     both FASTER (semijoin 1.33×, antijoin 2.58×, unique 1.49× @10M) and STABLE
     (the streaming executor spiked bimodally to ~1 s on the same semijoin; in-memory
-    holds ~30 ms). `raise_on_fail=False` keeps any GPU-incapable node on CPU **in
-    Polars** — NOT a pandas bridge (still honest/native; see NO-CHEATING). For
-    larger-than-device-memory inputs the in-memory engine would OOM rather than
+    holds ~30 ms). ``raise_on_fail=True`` is the NO-CHEATING contract for the GPU
+    target: if any node of the plan is not GPU-executable we RAISE — we never
+    silently run it on CPU and report it as a GPU result. (``raise_on_fail=False``
+    looked honest — fallback stays *in Polars*, not a pandas bridge — but it makes
+    ``engine='polars-gpu'`` indistinguishable from ``engine='polars'`` whenever the
+    plan isn't fully GPU-capable, which silently mislabels CPU work as GPU. A hard
+    raise forces the truth: ``polars-gpu`` is GPU-or-error; use ``polars`` for CPU.)
+    For larger-than-device-memory inputs the in-memory engine would OOM rather than
     stream — acceptable here (gfql graphs in scope fit), revisit if that changes."""
     if target == ExecutionTarget.GPU:
         import polars as pl
-        return pl.GPUEngine(executor="in-memory", raise_on_fail=False)
+        return pl.GPUEngine(executor="in-memory", raise_on_fail=True)
     return None
 
 
+def _gpu_raise(exc: Exception) -> "NotImplementedError":
+    """Translate a cudf-polars GPU-execution failure into the NO-CHEATING error.
+
+    With ``raise_on_fail=True`` the GPU engine raises when a plan node is not
+    GPU-executable. We surface that as a clear NotImplementedError (chained from the
+    polars error) instead of a cryptic ComputeError, pointing at the CPU engine."""
+    return NotImplementedError(
+        "GFQL engine='polars-gpu': this query plan is not fully GPU-executable on the "
+        "installed cudf-polars (NO-CHEATING: we raise rather than silently run it on CPU "
+        "and call it a GPU result). Rerun with engine='polars' for native CPU execution. "
+        f"Underlying GPU error: {type(exc).__name__}: {exc}"
+    )
+
+
 def collect(lf: Any) -> Any:
     """Collect one polars LazyFrame on the active target (CPU/GPU)."""
     eng = _engine_for(active_target())
-    return lf.collect(engine=eng) if eng is not None else lf.collect()
+    if eng is not None:
+        try:
+            return lf.collect(engine=eng)  # pragma: no cover  # GPU-target collect (no GPU in CI)
+        except NotImplementedError:
+            raise
+        except Exception as ex:  # pragma: no cover  # GPU-target collect (no GPU in CI)
+            raise _gpu_raise(ex) from ex
+    return lf.collect(engine="streaming") if _CPU_STREAMING else lf.collect()
 
 
 def collect_all(lfs: List[Any]) -> List[Any]:
@@ -87,8 +123,16 @@ def collect_all(lfs: List[Any]) -> List[Any]:
     eng = _engine_for(active_target())
     if hasattr(pl, "collect_all"):
         try:
-            return pl.collect_all(lfs, engine=eng) if eng is not None else pl.collect_all(lfs)
+            if eng is not None:
+                return pl.collect_all(lfs, engine=eng)  # pragma: no cover  # GPU-target collect (no GPU in CI)
+            return pl.collect_all(lfs, engine="streaming") if _CPU_STREAMING else pl.collect_all(lfs)
         except TypeError:
             # older signature without engine= — collect individually on target
             pass
+        except NotImplementedError:  # pragma: no cover  # GPU-target collect (no GPU in CI)
+            raise
+        except Exception as ex:  # pragma: no cover  # GPU-target collect (no GPU in CI)
+            if eng is not None:
+                raise _gpu_raise(ex) from ex
+            raise
     return [collect(lf) for lf in lfs]

graphistry/compute/gfql_unified.py

@@ -1679,7 +1679,10 @@ def _chain_dispatch(
     context: ExecutionContext,
     start_nodes: Optional[DataFrameT] = None,
 ) -> Plottable:
-    if chain_obj.where and engine in (EngineAbstract.POLARS, "polars", Engine.POLARS):
+    if chain_obj.where and engine in (
+        EngineAbstract.POLARS, "polars", Engine.POLARS,
+        EngineAbstract.POLARS_GPU, "polars-gpu", Engine.POLARS_GPU,
+    ):
         # Cross-entity / same-path WHERE routes through DFSamePathExecutor
         # (df_executor.py), which has no native polars implementation. NO pandas
         # fallback (see plan.md NO-CHEATING) — raise honestly.