feat(gfql): physical adjacency indexes for O(degree) seeded traversal

benchmarks/gfql/README.md

@@ -382,3 +382,30 @@ So the honest current comparison is:
 - Neo4j is workable for the smaller Twitter analog, but already materially slower than both Graphistry CPU and GPU on the exact same shape.
 - On the selected GPlus benchmark shape, Neo4j is already dramatically slower than Graphistry CPU (`83.61s`) and Graphistry GPU (`3.41s`) before teardown/cleanup is even done.
 - Raw notes: `plans/gfql-gpu-pagerank-benchmark/results/neo4j_summary.md`
+
+## GFQL physical-index harnesses (`index_*.py`)
+
+Catalog for the seeded-traversal adjacency-index work (PR #1658; see
+`docs/source/gfql/index_adjacency.rst` for the user guide + published numbers):
+
+- `index_smoke.py` / `index_ddl_smoke.py` — correctness smokes: differential
+  parity (index path == scan path) and DDL/wire/`index_policy`/`gfql_explain`
+  round-trips. Container-runnable MIRRORS of the canonical pytest suite
+  (`graphistry/tests/compute/gfql/index/test_index.py`).
+- `index_perf.py` — microbenchmark: seeded 1-hop/2-hop index-vs-scan across
+  engines and frontier sizes; the cost-gate calibration numbers come from here.
+- `index_takeover_bench.py` + `index_vs_kuzu_prepared.py` — **canonical
+  competitor pair** behind the published "9–36x vs Kuzu" seeded-traversal
+  claims: guarded timings (path-taken assertions + result==scan oracle),
+  prepared-statement fairness on the Kuzu side.
+- `index_vs_dbs.py` — broader 3-way sweep (GFQL / Kuzu / Neo4j) over the same
+  seeded shapes; superset of the pair above, heavier setup (dockerized Neo4j).
+- `index_bulk_olap_bench.py` — bulk group-by/degree OLAP shapes across the 4
+  engines (answers "does the index help bulk scans?" — it does not; scans win).
+- `index_largegraph_bench.py` — scale probe on large real graphs
+  (flat-in-N behavior of the seeded path up to ~80M edges).
+
+Methodology for all: warm medians, one engine per process for large runs,
+NO-CHEATING guards (a timing is void unless the intended path was taken AND
+the result matches the scan oracle). Run on an idle box; GPU rows need
+`--gpus all` and RAPIDS.

benchmarks/gfql/RESULTS.md

@@ -85,3 +85,6 @@ Summary-only log for notable benchmark runs. Raw per-scenario outputs live in
 | 2026-03-15 | DGX exploratory (feat/gfql-gpu-pagerank-benchmark) | `neo4j_pipeline_probe.py` Neo4j+GDS analog on SNAP GPlus | GPlus exact analog (`degree_q=0.995`, `pagerank_q=0.9995`) exceeded `3m07s` before the main transaction even finished closing, even after batching seed/core expansion to avoid transaction-memory OOM. | Raw output: `plans/gfql-gpu-pagerank-benchmark/results/neo4j_summary.md` |
 | 2026-03-15 | wip (`feat/gfql-gpu-pagerank-benchmark`) | `load_prepare_cpu_gpu.py` on SNAP Twitter (`degree_q=0.99`) | Cached prep median: CPU `0.2756s` vs GPU `0.1013s` (`2.72x`). Read `0.2156s` vs `0.0862s`; node prep `0.0620s` vs `0.0148s`; bind negligible. | Raw output: `plans/gfql-gpu-pagerank-benchmark/results/twitter_load_prepare_infer.json` |
 | 2026-03-15 | wip (`feat/gfql-gpu-pagerank-benchmark`) | `load_prepare_cpu_gpu.py` on SNAP GPlus (`degree_q=0.995`) | Cached prep median: CPU `8.7160s` vs GPU `3.9323s` (`2.22x`). Read `6.9096s` vs `3.0395s`; node prep `1.8097s` vs `0.8613s`; bind negligible. | Raw output: `plans/gfql-gpu-pagerank-benchmark/results/gplus_load_prepare_infer.json`; explicit integer-dtype probe rejected because Twitter regressed slightly and GPlus pandas overflowed. |
+| 2026-07-01 | 1257dac9 (dev/gfql-seeded-traversal-index) | `index_perf.py` ENGINES=pandas,polars NS=1e4,1e5,1e6 DEG=8 REPS=10 SEEDS=1 (local CPU, small-run proof) | Seeded warm FLAT ~0.30–0.33ms across N=1e4→1e6 (edges 80k→8M); scan O(E): pandas 8.6→808ms (28×→2451×), polars 2.2→45.5ms (7×→145×). warm floor engine-independent (~0.3ms=numpy searchsorted). Parity via 41-passed test suite. | Raw: `plans/gfql-1658-seeded-index/receipts/p1-{pandas,polars}.jsonl` |
+| 2026-07-01 | 1257dac9 (dev/gfql-seeded-traversal-index) | `explicit_vs_auto_sweep.py` (frontier sweep, pandas+polars, N=1e5 E=8e5, guarded match=True) | Index crossover engine-dependent: pandas wins to ~frac 0.5, polars scan ~18× faster → crossover ~frac 0.02. **DECISION: default `use`.** Finding F1(IMPORTANT): cost gate 0.5·n_keys miscalibrated for polars → resident index 0.4–0.5× (2× slower) in F≈1k–50k mid-band; fix=engine-aware gate. | Decision: `plans/gfql-1658-seeded-index/receipts/decision-explicit-vs-auto.md`; raw: `…/p3-explicit-vs-auto.jsonl` |
+| 2026-07-01 | 1257dac9 (dev/gfql-seeded-traversal-index) | `lookup_probe.py` [D-index-wiring] point/range, N=500k M=2M id=1..N, pandas+polars CPU | Bug confirmed (node_id index ignored by filter path, 1.01× w/ index) BUT filter is sub-ms (pandas 0.79ms/polars 0.34ms); the point-lookup cost is the **Cypher RETURN row-pipeline**: RETURN a = pandas 94.7ms vs polars 1.2ms. DECISION: **REJECT node_id→filter wiring** (saves <1ms vs ~94ms RETURN); real lever = pandas RETURN path / use polars-cudf. | `plans/gfql-1658-seeded-index/receipts/lookup-finding.md` |

benchmarks/gfql/index_bulk_olap_bench.py

@@ -0,0 +1,196 @@
+#!/usr/bin/env python3
+"""BULK-OLAP head-to-head: GFQL 4 engines vs kuzu on REAL graphs.
+
+Answers "is bulk OLAP better with GFQL (cudf / polars-gpu)?" The seeded CSR index
+is O(degree) and wins tiny work; this bench deliberately AVOIDS that path and
+measures the BULK regime instead — large-frontier multi-hop + full-graph
+aggregation, i.e. the scan/join work where columnar GPU throughput should pay off
+and the index does NOT help. We run g0.hop (NO resident index -> engine traversal,
+the honest bulk path) so every engine does the same materialized join work.
+
+Tasks (all bulk, all materialized on both sides):
+  BULK1   1-hop forward from K seeds         (edge semijoin, frontier=K)
+  BULK2   2-hop forward from K seeds         (edge-edge join, frontier blows up)
+  DEGALL  full-graph out-degree aggregation  (group_by over ALL edges; pure OLAP)
+K frontier sweep: 1k, 10k, 100k seeds. cudf/polars-gpu should overtake pandas as K
+(hence work) grows; kuzu is the WCOJ/optimizer peer for the multi-hop join.
+
+Trust: GFQL rows reported per engine (engine parity is separately guaranteed by the
+conformance suite); kuzu rows reported alongside with a semantic note. Timing is the
+deliverable — rows are the honesty check that each system did real work.
+
+Env: PARQUET=/path/edges.parquet  KS=1000,10000,100000  ENGINES=pandas,polars,cudf,polars-gpu
+     SYSTEMS=gfql,kuzu  REPS=10  WARM=2  OUT=/tmp/bulk.jsonl  SEED=0
+"""
+from __future__ import annotations
+import json, os, statistics, time, tempfile, shutil
+import numpy as np
+import pandas as pd
+import graphistry
+from graphistry.compute.ast import n, e_forward
+
+
+def _sync(engine):
+    if engine in ("cudf", "polars-gpu"):
+        try:
+            import cupy as cp  # type: ignore
+            cp.cuda.runtime.deviceSynchronize()
+        except Exception:
+            pass
+
+
+def timeit(fn, reps, engine="cpu", warmup=2):
+    for _ in range(warmup):
+        fn(); _sync(engine)
+    ts = []
+    for _ in range(reps):
+        t0 = time.perf_counter(); fn(); _sync(engine)
+        ts.append((time.perf_counter() - t0) * 1e3)
+    ts.sort()
+    return statistics.median(ts)
+
+
+def load_graph():
+    edf = pd.read_parquet(os.environ["PARQUET"]).astype({"src": np.int64, "dst": np.int64})
+    nodes = np.unique(np.concatenate([edf["src"].values, edf["dst"].values]))
+    ndf = pd.DataFrame({"id": nodes})
+    return ndf, edf, nodes
+
+
+def gfql_trav(g0, seed_ids, hops, engine):
+    """BULK seeded multi-hop via the CHAIN API — the one GFQL surface that supports
+    ALL FOUR engines (generic hop() is pandas/cudf only; polars/polars-gpu route
+    through engine_polars). n({id:seeds}) = frontier filter, then e_forward()*hops."""
+    ops = [n({"id": seed_ids})] + [e_forward() for _ in range(hops)]
+    return g0.chain(ops, engine=engine)
+
+
+def run_gfql(ndf, edf, nodes, ks, engines, reps, warm, outf, seed):
+    N, E = len(ndf), len(edf)
+    rng = np.random.default_rng(seed)
+    seed_sets = {k: rng.choice(nodes, size=min(k, len(nodes)), replace=False).tolist() for k in ks}
+    for engine in engines:
+        try:
+            g0 = graphistry.nodes(ndf, "id").edges(edf, "src", "dst")
+            # warm/convert frames onto the engine ONCE (exclude H2D/convert from timing)
+            _ = gfql_trav(g0, seed_sets[ks[0]], 1, engine)
+        except Exception as ex:
+            print(f"  gfql {engine}: SETUP FAILED {type(ex).__name__}: {ex}"); continue
+        # frontier sweep: BULK1 (1-hop) + BULK2 (2-hop)
+        for k in ks:
+            sids = seed_sets[k]
+            for task, hops in (("BULK1", 1), ("BULK2", 2)):
+                try:
+                    res = gfql_trav(g0, sids, hops, engine)
+                    rows = int(res._edges.shape[0]); nn = int(res._nodes.shape[0])
+                    ms = timeit(lambda: gfql_trav(g0, sids, hops, engine), reps, engine, warm)
+                except Exception as ex:
+                    print(f"  gfql {engine} {task} k={k} FAILED: {type(ex).__name__}: {ex}"); continue
+                rec = dict(system="gfql", engine=engine, task=task, k=k, hops=hops,
+                           n=N, edges=E, warm_ms=ms, e_rows=rows, n_rows=nn)
+                print(f"  gfql {engine:11} {task} k={k:>7} {ms:10.3f}ms  e_rows={rows:>10}  n_rows={nn:>9}")
+                if outf: outf.write(json.dumps(rec) + "\n"); outf.flush()
+        # DEGALL: full-graph out-degree aggregation (pure columnar OLAP, no traversal)
+        try:
+            ms, rows = degall(edf, engine, reps, warm)
+            rec = dict(system="gfql", engine=engine, task="DEGALL", k=None, hops=0,
+                       n=N, edges=E, warm_ms=ms, e_rows=rows, n_rows=rows)
+            print(f"  gfql {engine:11} DEGALL{'':>13} {ms:10.3f}ms  groups={rows:>10}")
+            if outf: outf.write(json.dumps(rec) + "\n"); outf.flush()
+        except Exception as ex:
+            print(f"  gfql {engine} DEGALL FAILED: {type(ex).__name__}: {ex}")
+
+
+def degall(edf, engine, reps, warm):
+    """Full-graph out-degree = group_by(src).size() on the chosen engine."""
+    if engine == "pandas":
+        df = edf
+        fn = lambda: df.groupby("src").size()
+    elif engine == "cudf":
+        import cudf
+        df = cudf.from_pandas(edf)
+        fn = lambda: df.groupby("src").size()
+    elif engine in ("polars", "polars-gpu"):
+        import polars as pl
+        df = pl.from_pandas(edf)
+        if engine == "polars-gpu":
+            # raise_on_fail=True: GPU-or-error — a silent CPU fallback would report
+            # CPU time as a polars-gpu result (banned; see the polars-gpu CHANGELOG fix)
+            eng = pl.GPUEngine(executor="in-memory", raise_on_fail=True)
+            fn = lambda: df.lazy().group_by("src").len().collect(engine=eng)
+        else:
+            fn = lambda: df.group_by("src").len()
+    else:
+        raise ValueError(engine)
+    r = fn(); rows = int(r.shape[0])
+    ms = timeit(fn, reps, engine, warm)
+    return ms, rows
+
+
+def run_kuzu(ndf, edf, nodes, ks, reps, warm, outf, seed, tmpdir):
+    try:
+        import kuzu
+    except Exception:
+        print("  kuzu: NOT AVAILABLE (pip install kuzu)"); return
+    rng = np.random.default_rng(seed)
+    seed_sets = {k: rng.choice(nodes, size=min(k, len(nodes)), replace=False).tolist() for k in ks}
+    dbp = tempfile.mkdtemp(dir=tmpdir)
+    db = kuzu.Database(os.path.join(dbp, "kz")); conn = kuzu.Connection(db)
+    conn.execute("CREATE NODE TABLE N(id INT64, PRIMARY KEY(id))")
+    conn.execute("CREATE REL TABLE E(FROM N TO N)")
+    np_path = os.path.join(dbp, "n.parquet"); ep_path = os.path.join(dbp, "e.parquet")
+    ndf.to_parquet(np_path)
+    edf.rename(columns={"src": "from", "dst": "to"}).to_parquet(ep_path)
+    t0 = time.perf_counter()
+    conn.execute(f'COPY N FROM "{np_path}"'); conn.execute(f'COPY E FROM "{ep_path}"')
+    load_ms = (time.perf_counter() - t0) * 1e3
+    print(f"  kuzu load: {load_ms:.0f}ms")
+    # BULK1/BULK2: distinct reachable set from K seeds (materialized columnar via get_as_df)
+    q1 = conn.prepare("MATCH (a:N)-[:E]->(b:N) WHERE a.id IN $seeds RETURN b.id")
+    q2 = conn.prepare("MATCH (a:N)-[:E]->()-[:E]->(b:N) WHERE a.id IN $seeds RETURN b.id")
+    for k in ks:
+        s = seed_sets[k]
+        for task, stmt in (("BULK1", q1), ("BULK2", q2)):
+            try:
+                rows = len(conn.execute(stmt, {"seeds": s}).get_as_df())
+                ms = timeit(lambda: conn.execute(stmt, {"seeds": s}).get_as_df(), reps, "kuzu", warm)
+            except Exception as ex:
+                print(f"  kuzu {task} k={k} FAILED: {type(ex).__name__}: {ex}"); continue
+            rec = dict(system="kuzu", engine="kuzu", task=task, k=k, n=len(ndf), edges=len(edf),
+                       warm_ms=ms, e_rows=rows, n_rows=rows, load_ms=load_ms)
+            print(f"  kuzu {'':11} {task} k={k:>7} {ms:10.3f}ms  rows={rows:>10}  (b.id, not-distinct)")
+            if outf: outf.write(json.dumps(rec) + "\n"); outf.flush()
+    # DEGALL: full out-degree aggregation
+    try:
+        qd = "MATCH (a:N)-[:E]->() RETURN a.id, count(*) AS deg"
+        for _ in range(warm): conn.execute(qd).get_as_df()
+        rows = len(conn.execute(qd).get_as_df())
+        ms = timeit(lambda: conn.execute(qd).get_as_df(), reps, "kuzu", warm)
+        rec = dict(system="kuzu", engine="kuzu", task="DEGALL", k=None, n=len(ndf), edges=len(edf),
+                   warm_ms=ms, e_rows=rows, n_rows=rows, load_ms=load_ms)
+        print(f"  kuzu {'':11} DEGALL{'':>13} {ms:10.3f}ms  groups={rows:>10}")
+        if outf: outf.write(json.dumps(rec) + "\n"); outf.flush()
+    except Exception as ex:
+        print(f"  kuzu DEGALL FAILED: {type(ex).__name__}: {ex}")
+    shutil.rmtree(dbp, ignore_errors=True)
+
+
+def main():
+    ndf, edf, nodes = load_graph()
+    print(f"===== graph: {len(ndf):,} nodes  {len(edf):,} edges =====")
+    ks = [int(x) for x in os.environ.get("KS", "1000,10000,100000").split(",")]
+    engines = os.environ.get("ENGINES", "pandas,polars,cudf,polars-gpu").split(",")
+    systems = os.environ.get("SYSTEMS", "gfql,kuzu").split(",")
+    reps = int(os.environ.get("REPS", "10")); warm = int(os.environ.get("WARM", "2"))
+    seed = int(os.environ.get("SEED", "0"))
+    tmpdir = os.environ.get("TMPDIR_BENCH", "/tmp/bulkbench"); os.makedirs(tmpdir, exist_ok=True)
+    outf = open(os.environ["OUT"], "a") if os.environ.get("OUT") else None
+    if "gfql" in systems:
+        run_gfql(ndf, edf, nodes, ks, engines, reps, warm, outf, seed)
+    if "kuzu" in systems:
+        run_kuzu(ndf, edf, nodes, ks, reps, warm, outf, seed, tmpdir)
+    if outf: outf.close()
+
+
+if __name__ == "__main__":
+    main()

benchmarks/gfql/index_ddl_smoke.py

@@ -0,0 +1,89 @@
+#!/usr/bin/env python3
+"""Smoke test: Cypher DDL + JSON wire + index_policy auto/force + gfql_explain.
+
+Container-runnable MIRROR of graphistry/tests/compute/gfql/index/test_index.py
+(the pytest suite is canonical); kept for quick in-container smoke without pytest.
+"""
+from __future__ import annotations
+
+import sys
+import numpy as np
+import pandas as pd
+
+import graphistry
+from graphistry.compute.gfql.index import (
+    CreateIndex, DropIndex, ShowIndexes, index_op_from_json, parse_index_ddl,
+)
+
+
+def make_graph(n=3000, deg=6, seed=1):
+    rng = np.random.default_rng(seed)
+    m = n * deg
+    edf = pd.DataFrame({"src": rng.integers(0, n, m), "dst": rng.integers(0, n, m)})
+    ndf = pd.DataFrame({"id": np.arange(n)})
+    return graphistry.nodes(ndf, "id").edges(edf, "src", "dst")
+
+
+def check(name, cond):
+    print(f"  {'OK  ' if cond else 'FAIL'} {name}")
+    return 0 if cond else 1
+
+
+def main():
+    g = make_graph()
+    seeds = pd.DataFrame({"id": [0, 1, 5, 9]})
+    fails = 0
+
+    # --- Cypher DDL recognizer ---
+    fails += check("parse CREATE", isinstance(parse_index_ddl("CREATE GFQL INDEX FOR edge_out_adj"), CreateIndex))
+    fails += check("parse CREATE named+col", parse_index_ddl("CREATE GFQL INDEX pk FOR node_id ON id").column == "id")
+    fails += check("parse DROP FOR", isinstance(parse_index_ddl("DROP GFQL INDEX FOR edge_in_adj"), DropIndex))
+    fails += check("parse SHOW", isinstance(parse_index_ddl("SHOW GFQL INDEXES"), ShowIndexes))
+    fails += check("non-DDL -> None", parse_index_ddl("MATCH (a) RETURN a") is None)
+
+    # --- Cypher DDL via gfql() drives the registry ---
+    g2 = g.gfql("CREATE GFQL INDEX FOR edge_out_adj")
+    g2 = g2.gfql("CREATE GFQL INDEX FOR edge_in_adj")
+    g2 = g2.gfql("CREATE GFQL INDEX FOR node_id")
+    si = g2.gfql("SHOW GFQL INDEXES")
+    fails += check("SHOW returns 3 rows", hasattr(si, "shape") and si.shape[0] == 3)
+    g2d = g2.gfql("DROP GFQL INDEX FOR edge_in_adj")
+    fails += check("DROP removed one", g2d.show_indexes().shape[0] == 2)
+
+    # --- JSON wire round-trip ---
+    op = CreateIndex(kind="edge_out_adj")
+    j = op.to_json()
+    fails += check("wire round-trip", index_op_from_json(j) == op)
+    g3 = g.gfql({"type": "CreateIndex", "kind": "edge_out_adj"})
+    fails += check("wire CreateIndex via gfql", g3.show_indexes().shape[0] == 1)
+    show_via_wire = g3.gfql({"type": "ShowIndexes"})
+    fails += check("wire ShowIndexes", hasattr(show_via_wire, "shape") and show_via_wire.shape[0] == 1)
+
+    # --- parity: cypher-DDL-built index hop == scan hop ---
+    base = g.hop(nodes=seeds, hops=2, direction="forward")
+    idxed = g2.hop(nodes=seeds, hops=2, direction="forward")
+    bn = sorted(base._nodes["id"].tolist()); xn = sorted(idxed._nodes["id"].tolist())
+    fails += check("DDL-built index parity (nodes)", bn == xn)
+
+    # --- index_policy auto/force build-on-demand + explain ---
+    gp = make_graph()  # fresh, no resident indexes
+    # explain with a seeded 1-hop chain (explain needs a real, seeded query)
+    from graphistry.compute.ast import n, e_forward
+    seeded_chain = [n({"id": 0}), e_forward(hops=1)]
+    rep_off = gp.gfql_explain(seeded_chain, index_policy="off")
+    fails += check("explain off -> scan", rep_off["used_index"] is False)
+    rep_force = gp.gfql_explain(seeded_chain, index_policy="force")
+    fails += check("explain force -> index", rep_force["used_index"] is True)
+
+    # force on a fresh graph actually returns correct result
+    gp_use = make_graph()
+    r_scan = gp_use.gfql(seeded_chain)
+    r_force = gp_use.gfql(seeded_chain, index_policy="force")
+    fails += check("force parity", sorted(r_scan._nodes["id"].tolist()) == sorted(r_force._nodes["id"].tolist()))
+
+    print(f"\n=== DDL/wire smoke: {'PASS' if fails == 0 else f'{fails} FAILED'} ===")
+    return 1 if fails else 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())