feat(gfql): enrich gfql_explain with planner cost diagnostics (LP1)

gfql_explain / index_trace now surface *why* the seeded-hop planner chose
the index or the scan, not just used_index. Per-hop steps carry frontier_n,
n_keys, a free Σ-seed-degree fanout estimate (seed_deg_sum/est_result_rows,
read from the CSR group_offsets already on the adjacency index), the engine
cost-gate threshold, and a human-readable decision_reason recorded at every
bail point. The report lifts est_seed_cardinality/est_result_rows/
chosen_direction/decision_reason to the top level.

Purely additive: the fanout estimate is computed only inside an
index_trace()/gfql_explain context, so the hot traversal path pays nothing
and behavior is unchanged. Stacked on the #1658 CSR index.

+1 engine-parametrized test; ruff + mypy clean; 45 CPU index tests pass.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: lmeyerov

CHANGELOG.md

@@ -14,6 +14,7 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 - **GFQL native Polars engine — variable-length `min_hops>1` traversals (`engine='polars'`/`'polars-gpu'`)**: Forward/reverse lower-bound traversals (`e(min_hops=N, max_hops=M)`) now run natively on the Polars engine — no pandas bridge. The eager Polars hop runs pandas' min_hops algorithm vectorized: a NON-anti-joined BFS (the wavefront carries revisits so a cycle keeps bumping the reach depth until `max_hops`), a 3-case termination gate (`max_reached<min`→empty; goal-edges-labeled-≥min→a descending layered backward-tree walk; cyclic-revisit-only→unpruned ball), and — the subtle part proven out against the pandas oracle — the exact min_hops NODE-output rule: the wavefront is the **endpoints of the retained edges**, MINUS seed nodes never genuinely re-reached at ≥min_hops, with FULL node attributes only for nodes carrying a retained-path hop-label and NULL attributes for source-side endpoint-only nodes (so a subsequent node-attribute filter rejects them, matching pandas' `track_node_hops` labeled path exactly). This is wired up for `chain()`/`gfql()`. **NO CHEATING:** UNDIRECTED `min_hops>1` (needs connected-components + 2-core seed retention) and a *direct* `hop(min_hops>1)` (which would need pandas' separate un-labeled direct-hop node-output plus the `target_wave_front` threading the chain supplies — without them it silently diverges) both raise `NotImplementedError` for `engine='polars'` (use `chain()`/`gfql()`, or `engine='pandas'`). Validated by differential parity vs the pandas oracle: the 500-seed randomized chain fuzzer (`test_polars_chain_fuzz_parity`, hardened to compare null-aware node **attributes** and edge **multiplicity**, not just id/endpoint sets) is **500/500**, a min_hops+attribute-filter amplified fuzz and metamorphic invariants pass, and `engine='polars-gpu'` (cudf_polars) runs the full 500-seed fuzz **500/500** on-device.
 - **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.
+- **GFQL `gfql_explain` planner diagnostics**: `gfql_explain(...)` now surfaces *why* the seeded-hop planner chose the index or the scan, not just a `used_index` yes/no. Each per-hop step carries the frontier seed count (`frontier_n`), the number of distinct source keys (`n_keys`), a **free Σ-seed-degree fanout estimate** (`seed_deg_sum`/`est_result_rows`, read directly from the CSR `group_offsets` already on the adjacency index — no extra scan), the engine's cost-gate crossover fraction (`threshold_frac`), and a human-readable `decision_reason` (e.g. `"frontier below cost gate -> index"`, `"frontier N >= 0.02*n_keys (…) -> scan cheaper"`, `"query not index-coverable"`, `"policy=off"`). The report also lifts `est_seed_cardinality`, `est_result_rows`, `chosen_direction`, and the top-level `decision_reason` for quick inspection. Purely additive and diagnostic — the fanout estimate is computed **only** inside an `index_trace()`/`gfql_explain` context, so the hot traversal path pays nothing and traversal behavior is unchanged.
 
 ### Added
 - **GFQL native Polars engine — more cypher row coverage (`toFloat`, `collect`/`collect(DISTINCT)`, `WHERE … IN`)**: three surfaces that previously raised `NotImplementedError` on `engine='polars'` now run natively, parity-validated vs the pandas oracle across all four engines (and honest-NIE where pandas can't be matched). **`toFloat(x)`** lowers int/uint/bool/float → `Float64` (NaN preserved — float64 has no separate null sentinel, unlike `toInteger`); a non-numeric String declines (NIE) because pandas `astype(float)` *raises* rather than null-on-failure. **`collect(x)` / `collect(DISTINCT x)`** aggregations complete the native `group_by` surface (every other agg was already native): drop nulls, preserve within-group first-occurrence order (`collect` keeps dups; `DISTINCT` dedups keep-first), all-null group → `[]`. **`where_rows`/`WHERE … IN [list]`** membership lowers to `is_in` (a null cell is excluded per openCypher 3VL). No change to any already-native path.

graphistry/compute/gfql/index/api.py

@@ -54,6 +54,46 @@ def _record(decision: dict) -> None:
         steps.append(decision)
 
 
+def _trace_active() -> bool:
+    """True only inside an ``index_trace()`` / ``gfql_explain`` context. Diagnostic
+    enrichment (LP1) is computed only when this is True → zero hot-path cost."""
+    return getattr(_TRACE, "steps", None) is not None
+
+
+def _seed_id_array(nodes, node_col):
+    """Seed ids of the frontier as a host numpy array (device→host copy is fine —
+    only called under an explain trace). None on any failure."""
+    try:
+        col = nodes[node_col]
+        if hasattr(col, "to_numpy"):
+            return col.to_numpy()
+        arr = getattr(col, "values", col)
+        return arr.get() if hasattr(arr, "get") else arr
+    except Exception:
+        return None
+
+
+def _seed_deg_sum(idx, seed_ids) -> Optional[int]:
+    """Σ degree of the resident seeds — the true index expansion cost, FREE from the
+    CSR ``group_offsets`` already on the AdjacencyIndex. Diagnostic only (LP1); the
+    report's planner needs this fanout estimate exposed via EXPLAIN."""
+    try:
+        import numpy as np
+        ks = idx.keys_sorted
+        ks = ks.get() if hasattr(ks, "get") else np.asarray(ks)
+        off = idx.group_offsets
+        off = off.get() if hasattr(off, "get") else np.asarray(off)
+        seed_ids = np.asarray(seed_ids)
+        pos = np.searchsorted(ks, seed_ids)
+        valid = pos < ks.size
+        pos_v = pos[valid]
+        seed_v = seed_ids[valid]
+        hit = pos_v[ks[pos_v] == seed_v]  # keep only real key hits (robust to seed order)
+        return int((off[hit + 1] - off[hit]).sum())
+    except Exception:
+        return None
+
+
 def get_registry(g: Plottable) -> GfqlIndexRegistry:
     return getattr(g, REGISTRY_ATTR, EMPTY_REGISTRY)
 
@@ -279,11 +319,34 @@ def maybe_index_hop(
     (or buildable under auto/force), (b) the query is covered, (c) the frontier is
     not so large that a full scan is cheaper. Correctness is identical either way.
     """
-    if policy == "off":
+    # Diagnostic trace (LP1) is populated only inside an explain context — build the
+    # base record + a `_bail` helper that logs *why* we fell back to scan. All of this
+    # is skipped entirely when not tracing, so the hot path pays nothing.
+    trace = _trace_active()
+    diag: dict = {}
+    if trace:
+        diag = {
+            "op": "hop", "direction": direction, "hops": hops,
+            "policy": policy, "engine": engine.value,
+        }
+        try:
+            diag["frontier_n"] = int(nodes.shape[0])
+        except Exception:
+            pass
+
+    def _bail(reason: str, extra: Optional[dict] = None) -> Optional[Plottable]:
+        if trace:
+            rec = {**diag, "path": "scan", "decision_reason": reason}
+            if extra:
+                rec.update(extra)
+            _record(rec)
         return None
+
+    if policy == "off":
+        return _bail("policy=off")
     registry = get_registry(g)
     if registry.is_empty() and policy not in ("auto", "force"):
-        return None
+        return _bail("no resident index (policy=use)")
     if not _hop_is_index_coverable(
         nodes=nodes, to_fixed_point=to_fixed_point, hops=hops,
         min_hops=rest.get("min_hops"), max_hops=rest.get("max_hops"),
@@ -301,32 +364,44 @@ def maybe_index_hop(
         include_zero_hop_seed=rest.get("include_zero_hop_seed", False),
         target_wave_front=rest.get("target_wave_front"),
     ):
-        return None
+        return _bail("query not index-coverable")
 
     node_col = g._node
     src, dst = g._source, g._destination
     if node_col is None or src is None or dst is None or g._edges is None or g._nodes is None:
-        return None
+        return _bail("graph missing node/edge columns")
 
     if policy in ("auto", "force"):
         registry = _ensure_indexes(g, registry, direction, engine, policy, nodes, src, dst, node_col)
     if registry.is_empty():
-        return None
+        return _bail("no index available (build declined)")
 
     # Cost gate: if the frontier covers a large fraction of distinct sources, the
     # scan path is competitive — fall back (avoids index overhead on bulk-ish hops).
     idx0 = registry.get_valid(
         EDGE_OUT_ADJ if direction != "reverse" else EDGE_IN_ADJ, g._edges, (src, dst), engine
     )
+    frac = _COST_GATE_FRAC.get(engine, _COST_GATE_FRAC_DEFAULT)
+    if trace and idx0 is not None:
+        # Free fanout estimate (Σ seed degree) from the CSR offsets — the planner
+        # signal the report wants EXPLAIN to surface (not just used-index yes/no).
+        seed_ids = _seed_id_array(nodes, node_col)
+        deg_sum = _seed_deg_sum(idx0, seed_ids) if seed_ids is not None else None
+        diag["n_keys"] = int(idx0.n_keys)
+        diag["seed_deg_sum"] = deg_sum
+        diag["est_result_rows"] = deg_sum
+        diag["threshold_frac"] = frac
     if idx0 is None:
         # required direction not resident (undirected needs both); let driver decide
         pass
     elif policy != "force":
         try:
             frontier_n = int(nodes.shape[0])
-            frac = _COST_GATE_FRAC.get(engine, _COST_GATE_FRAC_DEFAULT)
             if idx0.n_keys > 0 and frontier_n >= frac * idx0.n_keys:
-                return None
+                return _bail(
+                    f"frontier {frontier_n} >= {frac}*n_keys "
+                    f"({frac * idx0.n_keys:.0f}) -> scan cheaper"
+                )
         except Exception:
             pass
 
@@ -341,9 +416,13 @@ def maybe_index_hop(
         hops=eff_hops, to_fixed_point=to_fixed_point, direction=direction,
         return_as_wave_front=return_as_wave_front,
     )
-    _record({
-        "op": "hop", "direction": direction, "hops": eff_hops,
-        "path": "index" if result is not None else "scan",
-        "policy": policy, "engine": engine.value,
-    })
+    if trace:
+        _record({
+            **diag, "hops": eff_hops,
+            "path": "index" if result is not None else "scan",
+            "decision_reason": (
+                "frontier below cost gate -> index" if result is not None
+                else "index path not applicable -> scan"
+            ),
+        })
     return result

graphistry/compute/gfql/index/explain.py

@@ -23,11 +23,20 @@ def gfql_explain(g: Any, query: Any, *, index_policy: str = "use", engine: str =
         except Exception as ex:  # report, don't raise — explain is diagnostic
             error = f"{type(ex).__name__}: {ex}"
     used_index = any(s.get("path") == "index" for s in steps)
+    # Surface the planner's cost signal at the top level (LP1): prefer the step that
+    # actually took the index, else the last decision. `est_seed_cardinality` = number
+    # of seeds; `est_result_rows` = estimated fanout (Σ seed degree, free from CSR).
+    ref = [s for s in steps if s.get("path") == "index"] or list(steps)
+    last = ref[-1] if ref else {}
     return {
         "engine": eng.value,
         "index_policy": index_policy,
         "resident_indexes": resident["name"].tolist() if len(resident) else [],
         "steps": list(steps),
         "used_index": used_index,
+        "est_seed_cardinality": last.get("frontier_n"),
+        "est_result_rows": last.get("est_result_rows"),
+        "chosen_direction": last.get("direction"),
+        "decision_reason": last.get("decision_reason"),
         "error": error,
     }

graphistry/tests/compute/gfql/index/test_index.py

@@ -135,6 +135,32 @@ def test_index_policy_force_and_explain(graph, engine):
     assert _sig(r_scan) == _sig(r_force)
 
 
+@pytest.mark.parametrize("engine", ENGINES)
+def test_explain_exposes_planner_diagnostics(graph, engine):
+    """LP1: gfql_explain surfaces the planner's cost signal — seed cardinality, the
+    free Σ-degree fanout estimate (from CSR group_offsets), chosen direction, the
+    cost-gate threshold, and a human-readable decision_reason — not just a used-index
+    yes/no. This is the EXPLAIN enrichment the indexing roadmap calls for."""
+    chain = [n({"id": 0}), e_forward(hops=1)]
+    # force → index path taken → per-step + top-level diagnostics populated
+    rep = graph.gfql_explain(chain, index_policy="force", engine=engine)
+    assert rep["used_index"] is True, (engine, rep)
+    assert rep["chosen_direction"] == "forward"
+    assert isinstance(rep["est_result_rows"], int) and rep["est_result_rows"] >= 0
+    assert "index" in (rep["decision_reason"] or ""), rep["decision_reason"]
+    step = next(s for s in rep["steps"] if s.get("path") == "index")
+    for k in ("frontier_n", "n_keys", "seed_deg_sum", "est_result_rows", "threshold_frac"):
+        assert k in step, (k, step)
+    assert step["est_result_rows"] == step["seed_deg_sum"]  # est == free Σ-degree
+    assert step["seed_deg_sum"] >= 0
+
+    # off (index resident) → the planner records *why* it scanned, not just that it did
+    gi = graph.gfql_index_all(engine=engine)
+    rep_off = gi.gfql_explain(chain, index_policy="off", engine=engine)
+    assert rep_off["used_index"] is False
+    assert rep_off["decision_reason"] == "policy=off", rep_off
+
+
 @pytest.mark.parametrize("engine", ENGINES)
 def test_cost_gate_engine_aware_never_loses_to_scan(engine):
     """F1: the index-vs-scan crossover depends on scan speed, so the cost gate