feat(gfql): native lazy Polars engine — collect-once traversals + cypher row pipeline + followups (multi-hop, to_fixed_point, undirected, more predicates, NIE→native)

.github/workflows/ci.yml

@@ -1020,7 +1020,7 @@ jobs:
     # PR hygiene only: combine coverage collected by existing CPU test jobs and
     # gate executable package lines touched by this PR. Historical uncovered
     # lines outside the diff are intentionally ignored.
-    needs: [changes, test-core-python, test-gfql-core, generate-lockfiles]
+    needs: [changes, test-core-python, test-gfql-core, test-polars, generate-lockfiles]
     if: ${{ github.event_name == 'pull_request' && (needs.changes.outputs.python == 'true' || needs.changes.outputs.gfql == 'true' || needs.changes.outputs.infra == 'true') && !(needs.changes.outputs.docs_only_latest == 'true') }}
     runs-on: ubuntu-latest
     timeout-minutes: 5
@@ -1063,6 +1063,12 @@ jobs:
         name: gfql-coverage-audit-py3.12
         path: build/changed-line-coverage/artifacts/gfql
 
+    - name: Download Polars coverage data
+      uses: actions/download-artifact@v4
+      with:
+        name: polars-coverage-py3.12
+        path: build/changed-line-coverage/artifacts/polars
+
     - name: Changed-line coverage gate
       env:
         BASE_SHA: ${{ github.event.pull_request.base.sha }}
@@ -1073,7 +1079,8 @@ jobs:
         python -m coverage combine \
           --data-file=build/changed-line-coverage/combined.coverage \
           build/changed-line-coverage/artifacts/core/coverage.core-py3.14 \
-          build/changed-line-coverage/artifacts/gfql/coverage.gfql-py3.12
+          build/changed-line-coverage/artifacts/gfql/coverage.gfql-py3.12 \
+          build/changed-line-coverage/artifacts/polars/coverage.polars-py3.12
         python bin/changed_line_coverage.py \
           --data-file build/changed-line-coverage/combined.coverage \
           --base-ref "$BASE_SHA" \
@@ -1389,10 +1396,39 @@ jobs:
         uv pip install -e . --no-deps
 
     - name: Polars tests
+      if: ${{ matrix.python-version != '3.12' }}
       run: |
         source pygraphistry/bin/activate
         ./bin/test-polars.sh
 
+    - name: Polars tests with coverage
+      if: ${{ matrix.python-version == '3.12' }}
+      run: |
+        source pygraphistry/bin/activate
+        mkdir -p build/polars-coverage
+        COVERAGE_FILE=build/polars-coverage/coverage.polars-py3.12 POLARS_COV=1 ./bin/test-polars.sh
+
+    - name: Polars GFQL coverage audit (per-file floors)
+      if: ${{ matrix.python-version == '3.12' }}
+      run: |
+        source pygraphistry/bin/activate
+        python bin/coverage_audit.py \
+          --profile gfql-polars \
+          --skip-tests \
+          --engine-label ci-polars-py3.12 \
+          --output-dir build/polars-coverage-audit \
+          --data-file build/polars-coverage/coverage.polars-py3.12 \
+          --baseline-file graphistry/tests/compute/gfql/coverage_baselines/ci-polars-py3.12.json
+        cat build/polars-coverage-audit/gfql-polars-coverage-audit.md | tee -a "$GITHUB_STEP_SUMMARY"
+
+    - name: Upload polars coverage
+      if: ${{ matrix.python-version == '3.12' }}
+      uses: actions/upload-artifact@v4
+      with:
+        name: polars-coverage-py3.12
+        path: build/polars-coverage/
+        retention-days: 14
+
   test-core-umap:
     needs: [changes, test-minimal-python, test-gfql-core, generate-lockfiles]
     if: ${{ success() && (needs.changes.outputs.ai == 'true' || needs.changes.outputs.infra == 'true' || github.event_name == 'workflow_dispatch' || github.event_name == 'schedule') }}

benchmarks/gfql/cypher_row_pipeline.py

@@ -0,0 +1,182 @@
+#!/usr/bin/env python3
+"""Benchmark the native polars GFQL row pipeline vs pandas for cypher queries.
+
+Phase 2 of the polars engine enables cypher RETURN / LIMIT / SKIP / DISTINCT /
+single-entity WHERE on ``engine='polars'`` (before this increment these raised
+NotImplementedError on polars). The heavy traversal + frame ops (filter, dedup,
+slice) run natively in polars; only the final row-wise entity-text projection is
+host-bridged to pandas. So polars wins most where a row op reduces the set
+before projection (LIMIT, selective WHERE, DISTINCT), and is closest to neutral
+on a full-table whole-entity RETURN (projection dominates, bridge roundtrip).
+
+Reports median latency and the polars speedup (pandas_ms / polars_ms; > 1 means
+polars wins). On a shared host, interleave is implicit (pandas then polars
+back-to-back per query); for regression-grade claims run several times and
+compare distributions (see plans/gfql-polars-engine memory).
+
+Example::
+
+    python benchmarks/gfql/cypher_row_pipeline.py --runs 7 --warmup 2 \
+        --sizes 10000,100000,1000000 --output /tmp/cypher-row.md
+"""
+
+from __future__ import annotations
+
+import argparse
+import statistics
+import time
+from dataclasses import dataclass
+from typing import Callable, List, Optional, Tuple
+
+import numpy as np
+import pandas as pd
+
+import graphistry
+
+# (name, cypher) — exercised on both engines via g.gfql(cypher, engine=...)
+# Native = frame ops (rows/limit/skip/distinct) run in polars; Bridged = the
+# cypher expression engine (select/order_by/group_by) runs host-bridged to pandas.
+WORKLOADS: List[Tuple[str, str]] = [
+    # native frame-op path
+    ("RETURN n LIMIT 10", "MATCH (n) RETURN n LIMIT 10"),
+    ("RETURN n SKIP/LIMIT", "MATCH (n) RETURN n SKIP 5 LIMIT 100"),
+    ("WHERE > RETURN LIMIT", "MATCH (n) WHERE n.score > 90 RETURN n LIMIT 50"),
+    ("RETURN DISTINCT n", "MATCH (n) RETURN DISTINCT n"),
+    ("WHERE > RETURN n", "MATCH (n) WHERE n.score > 50 RETURN n"),
+    ("RETURN n (full)", "MATCH (n) RETURN n"),
+    ("rel RETURN m LIMIT", "MATCH (n)-[e]->(m) RETURN m LIMIT 100"),
+    # host-bridged expression path
+    ("select n.score", "MATCH (n) RETURN n.score"),
+    ("select 2 cols", "MATCH (n) RETURN n.score, n.kind"),
+    ("order_by", "MATCH (n) RETURN n.score ORDER BY n.score DESC"),
+    ("where+select+limit", "MATCH (n) WHERE n.score > 50 RETURN n.score ORDER BY n.score LIMIT 100"),
+    ("group_by count", "MATCH (n) RETURN n.kind, count(n) AS c"),
+]
+
+
+@dataclass
+class ResultRow:
+    workload: str
+    n_nodes: int
+    n_edges: int
+    pandas_ms: Optional[float]
+    polars_ms: Optional[float]
+    error: Optional[str] = None
+
+    @property
+    def speedup(self) -> Optional[float]:
+        if self.pandas_ms and self.polars_ms:
+            return self.pandas_ms / self.polars_ms
+        return None
+
+
+def make_graph(n_nodes: int, n_edges: int, seed: int = 0):
+    rng = np.random.default_rng(seed)
+    nodes = pd.DataFrame({
+        "id": np.arange(n_nodes),
+        "kind": rng.choice(["x", "y", "z"], size=n_nodes),
+        "score": rng.integers(0, 100, size=n_nodes),
+    })
+    edges = pd.DataFrame({
+        "s": rng.integers(0, n_nodes, size=n_edges),
+        "d": rng.integers(0, n_nodes, size=n_edges),
+        "rel": rng.choice(["r1", "r2", "r3"], size=n_edges),
+    })
+    return graphistry.nodes(nodes, "id").edges(edges, "s", "d")
+
+
+def timeit(fn: Callable[[], object], runs: int, warmup: int) -> float:
+    for _ in range(warmup):
+        fn()
+    samples = []
+    for _ in range(runs):
+        t0 = time.perf_counter()
+        fn()
+        samples.append((time.perf_counter() - t0) * 1000.0)
+    return statistics.median(samples)
+
+
+def _polars_graph(g):
+    """Same graph with node/edge frames already in polars, so the polars runs
+    don't pay a per-call pandas->polars input coercion that the pandas runs avoid
+    (a real deployment keeps the graph in its engine's native frame type)."""
+    from graphistry.Engine import Engine, df_to_engine
+    return g.nodes(df_to_engine(g._nodes, Engine.POLARS), g._node).edges(
+        df_to_engine(g._edges, Engine.POLARS), g._source, g._destination)
+
+
+def run(sizes: List[Tuple[int, int]], runs: int, warmup: int) -> List[ResultRow]:
+    rows: List[ResultRow] = []
+    for n_nodes, n_edges in sizes:
+        g_pd = make_graph(n_nodes, n_edges)
+        g_pl = _polars_graph(g_pd)
+        for name, query in WORKLOADS:
+            try:
+                pandas_ms = timeit(lambda: g_pd.gfql(query, engine="pandas"), runs, warmup)
+                polars_ms = timeit(lambda: g_pl.gfql(query, engine="polars"), runs, warmup)
+                rows.append(ResultRow(name, n_nodes, n_edges, pandas_ms, polars_ms))
+            except Exception as exc:  # noqa: BLE001 - bench harness reports, never crashes the sweep
+                rows.append(ResultRow(name, n_nodes, n_edges, None, None, error=f"{type(exc).__name__}: {exc}"))
+    return rows
+
+
+def to_markdown(rows: List[ResultRow]) -> str:
+    lines = [
+        "| workload | nodes | edges | pandas_ms | polars_ms | speedup |",
+        "|----------|-------|-------|-----------|-----------|---------|",
+    ]
+    for r in rows:
+        if r.error:
+            lines.append(f"| {r.workload} | {r.n_nodes} | {r.n_edges} | ERROR | ERROR | {r.error} |")
+        else:
+            lines.append(
+                f"| {r.workload} | {r.n_nodes} | {r.n_edges} | "
+                f"{r.pandas_ms:.1f} | {r.polars_ms:.1f} | {r.speedup:.2f}x |"
+            )
+    return "\n".join(lines)
+
+
+def _parse_sizes(text: str) -> List[Tuple[int, int]]:
+    # "nodes:edges,nodes:edges" or "nodes" (edges defaults to 5x nodes)
+    out: List[Tuple[int, int]] = []
+    for chunk in text.split(","):
+        chunk = chunk.strip()
+        if not chunk:
+            continue
+        if ":" in chunk:
+            nn, ne = chunk.split(":")
+            out.append((int(nn), int(ne)))
+        else:
+            nn = int(chunk)
+            out.append((nn, nn * 5))
+    return out
+
+
+def main() -> None:
+    try:
+        import polars  # noqa: F401
+    except ImportError:
+        raise SystemExit("polars is not installed; install with `pip install polars`")
+
+    parser = argparse.ArgumentParser(description="Benchmark GFQL cypher row pipeline pandas vs polars.")
+    parser.add_argument("--runs", type=int, default=7)
+    parser.add_argument("--warmup", type=int, default=2)
+    parser.add_argument(
+        "--sizes",
+        default="10000,100000,1000000",
+        help="Comma list of node counts (edges=5x) or nodes:edges pairs.",
+    )
+    parser.add_argument("--output", default="", help="Optional path to write the markdown table.")
+    args = parser.parse_args()
+
+    rows = run(_parse_sizes(args.sizes), args.runs, args.warmup)
+    table = to_markdown(rows)
+    print(table)
+    if args.output:
+        with open(args.output, "w") as fh:
+            fh.write(table + "\n")
+        print(f"\nwrote {args.output}")
+
+
+if __name__ == "__main__":
+    main()

benchmarks/gfql/multihop_polars_bench.py

@@ -0,0 +1,70 @@
+"""Multi-hop polars-vs-pandas chain benchmark (validation for the native multi-hop port).
+
+Confirms native multi-hop polars is on-par-or-better than pandas on non-trivial graphs.
+Run on dgx (CPU lane is the relevant comparison; cudf/polars-gpu optional).
+"""
+import time
+import numpy as np
+import pandas as pd
+import graphistry
+from graphistry.compute.ast import n, e_forward, e_undirected
+
+
+def _graph(nn, ne, seed=0):
+    rng = np.random.default_rng(seed)
+    nd = pd.DataFrame({"id": np.arange(nn), "k": rng.integers(0, 5, nn)})
+    ed = pd.DataFrame({
+        "s": rng.integers(0, nn, ne),
+        "d": rng.integers(0, nn, ne),
+        "w": rng.integers(0, 100, ne),
+    })
+    return graphistry.nodes(nd, "id").edges(ed, "s", "d")
+
+
+def _time(g, ch, engine, reps=3):
+    g.chain(ch, engine=engine)  # warm
+    best = float("inf")
+    for _ in range(reps):
+        t = time.perf_counter()
+        out = g.chain(ch, engine=engine)
+        dt = time.perf_counter() - t
+        best = min(best, dt)
+    ne = 0 if out._edges is None else len(out._edges)
+    nn = 0 if out._nodes is None else len(out._nodes)
+    return best * 1000, nn, ne
+
+
+CHAINS = {
+    "fwd-hops2":  [n({"id": [0]}), e_forward(hops=2), n()],
+    "fwd-hops3":  [n({"id": [0]}), e_forward(hops=3), n()],
+    "maxhops4":   [n({"id": [0]}), e_forward(max_hops=4), n()],
+    "sandwiched": [n({"id": [0]}), e_forward(), n(), e_forward(hops=2), n()],
+    "fwd-tofixed": [n({"id": [0]}), e_forward(to_fixed_point=True), n()],
+    "und-hops2":  [n({"id": [0]}), e_undirected(hops=2), n()],
+    "und-maxhops3": [n({"id": [0]}), e_undirected(max_hops=3), n()],
+}
+
+for (nn, ne) in [(10_000, 50_000), (100_000, 500_000), (500_000, 2_000_000)]:
+    g = _graph(nn, ne)
+    print(f"\n=== graph nn={nn:,} ne={ne:,} ===")
+    for name, ch in CHAINS.items():
+        engines = ["pandas", "polars"]
+        try:
+            import cudf  # noqa: F401
+            engines.append("cudf")
+        except Exception:
+            pass
+        row = []
+        ref = None
+        for eng in engines:
+            try:
+                ms, rnn, rne = _time(g, ch, eng)
+                if eng == "pandas":
+                    ref = ms
+                spd = f"{ref/ms:.2f}x" if (ref and eng != "pandas") else "1.00x"
+                row.append(f"{eng}={ms:.1f}ms({spd},n={rnn},e={rne})")
+            except NotImplementedError:
+                row.append(f"{eng}=NIE")
+            except Exception as ex:
+                row.append(f"{eng}=ERR:{type(ex).__name__}")
+        print(f"  {name:12s} " + "  ".join(row))