fix(gfql): expose structured index decisions

Author: lmeyerov

graphistry/compute/gfql/index/api.py

@@ -25,7 +25,7 @@
 from .policy import IndexPolicy, validate_index_policy
 from .types import (
     AdjacencyIndexKind, EdgeIndexDirection, HopDirection, IndexKind,
-    IndexTrace, IndexTraceStep,
+    IndexDecisionCode, IndexTrace, IndexTraceStep,
 )
 
 # Private Plottable attachment keys. Keep access behind helpers.
@@ -140,6 +140,7 @@ def _record_indexed_traversal(
         "hop_details": [] if hop_details is None else hop_details,
         "path": path,
         "decision_reason": reason,
+        "decision_code": "index_selected" if served else "index_path_unavailable",
     }))
 
 
@@ -514,6 +515,8 @@ def maybe_index_hop(
     Cost gate: only route to the index when (a) a valid matching index is resident
     (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.
+    "force" bypasses only the cost gate for a covered query. It still falls back to
+    scan when the index cannot serve the query.
     """
     resolved_policy: IndexPolicy = validate_index_policy(policy) or "use"
 
@@ -533,16 +536,18 @@ def maybe_index_hop(
         except (AttributeError, TypeError, ValueError):
             pass
 
-    def _bail(reason: str) -> Optional[Plottable]:
+    def _bail(reason: str, decision_code: IndexDecisionCode) -> Optional[Plottable]:
         if trace:
-            _record(cast(IndexTraceStep, {**diag, "path": "scan", "decision_reason": reason}))
+            _record(cast(IndexTraceStep, {
+                **diag, "path": "scan", "decision_reason": reason, "decision_code": decision_code,
+            }))
         return None
 
     if resolved_policy == "off":
-        return _bail("policy=off")
+        return _bail("policy=off", "policy_off")
     registry = get_registry(g)
     if registry.is_empty() and resolved_policy not in ("auto", "force"):
-        return _bail("no resident index (policy=use)")
+        return _bail("no resident index (policy=use)", "no_resident_index")
 
     min_hops = cast(Optional[int], rest.get("min_hops"))
     max_hops = cast(Optional[int], rest.get("max_hops"))
@@ -575,18 +580,18 @@ def _bail(reason: str) -> Optional[Plottable]:
         target_wave_front=target_wave_front,
         return_as_wave_front=return_as_wave_front,
     ):
-        return _bail("query not index-coverable")
+        return _bail("query not index-coverable", "not_index_coverable")
     assert nodes is not None
 
     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 _bail("graph missing node/edge columns")
+        return _bail("graph missing node/edge columns", "missing_graph_columns")
 
     if resolved_policy in ("auto", "force"):
         registry = _ensure_indexes(g, registry, direction, engine, resolved_policy, nodes, src, dst, node_col)
     if registry.is_empty():
-        return _bail("no index available (build declined)")
+        return _bail("no index available (build declined)", "index_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).
@@ -612,7 +617,7 @@ def _bail(reason: str) -> Optional[Plottable]:
             if idx0.n_keys > 0 and frontier_n >= frac * idx0.n_keys:
                 return _bail(
                     f"frontier {frontier_n} >= {frac}*n_keys "
-                    f"({frac * idx0.n_keys:.0f}) -> scan cheaper"
+                    f"({frac * idx0.n_keys:.0f}) -> scan cheaper", "scan_cost"
                 )
         except (AttributeError, TypeError, ValueError):
             pass
@@ -629,12 +634,20 @@ def _bail(reason: str) -> Optional[Plottable]:
         edge_match=cast(Optional[dict], rest.get("edge_match")),
     )
     if trace:
+        engine_mismatch_reason = (
+            _engine_mismatch_reason(registry, direction, engine) if result is None else None
+        )
         _record(cast(IndexTraceStep, {
             **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 _engine_mismatch_reason(registry, direction, engine) or "index path not applicable -> scan"
+                else engine_mismatch_reason or "index path not applicable -> scan"
+            ),
+            "decision_code": (
+                "index_selected" if result is not None
+                else "engine_mismatch" if engine_mismatch_reason is not None
+                else "index_path_unavailable"
             ),
         }))
     return result

graphistry/compute/gfql/index/explain.py

@@ -13,7 +13,7 @@
 from graphistry.compute.gfql.query_types import GFQLQuery
 from .api import index_trace, show_indexes
 from .policy import IndexPolicy, validate_index_policy
-from .types import IndexTraceStep
+from .types import IndexDecisionCode, IndexTraceStep
 
 if TYPE_CHECKING:
     from graphistry.compute.ComputeMixin import ComputeMixin
@@ -29,6 +29,7 @@ class GfqlExplainReport(TypedDict):
     est_result_rows: Optional[int]
     chosen_direction: Optional[str]
     decision_reason: Optional[str]
+    decision_code: Optional[IndexDecisionCode]
     error: Optional[str]
 
 
@@ -54,6 +55,8 @@ def gfql_explain(
     # 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 {}
+    if not last and resolved_policy == "off":
+        last = {"decision_reason": "policy=off", "decision_code": "policy_off"}
     resident_names = cast(List[str], resident["name"].tolist() if len(resident) else [])
     return {
         "engine": eng.value,
@@ -65,5 +68,6 @@ def gfql_explain(
         "est_result_rows": cast(Optional[int], last.get("est_result_rows")),
         "chosen_direction": cast(Optional[str], last.get("direction")),
         "decision_reason": cast(Optional[str], last.get("decision_reason")),
+        "decision_code": last.get("decision_code"),
         "error": error,
     }

graphistry/compute/gfql/index/policy.py

@@ -10,7 +10,10 @@
 def validate_index_policy(policy: Optional[str]) -> Optional[IndexPolicy]:
     """Validate a public ``index_policy`` value.
 
-    ``None`` means the caller did not override the default planner behavior.
+    ``None`` means the caller did not override the default planner behavior. "force"
+    is a performance and diagnostic opt-in: it builds missing indexes and bypasses the
+    cost gate for an index-coverable query. It does not require index coverage and never
+    changes query results. A query that the index cannot serve still uses the scan path.
     """
     if policy is None:
         return None

graphistry/compute/gfql/index/types.py

@@ -21,6 +21,17 @@
 
 
 IndexPath = Literal["scan", "index"]
+IndexDecisionCode = Literal[
+    "policy_off",
+    "no_resident_index",
+    "not_index_coverable",
+    "missing_graph_columns",
+    "index_build_declined",
+    "scan_cost",
+    "index_selected",
+    "engine_mismatch",
+    "index_path_unavailable",
+]
 
 
 # One column's constraint in an ``edge_match``/filter dict — exactly the runtime shapes
@@ -54,6 +65,7 @@ class IndexTraceStep(TypedDict, total=False):
     frontier_n: int
     path: IndexPath
     decision_reason: str
+    decision_code: IndexDecisionCode
     n_keys: int
     seed_deg_sum: Optional[int]
     est_result_rows: Optional[int]

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

@@ -230,12 +230,16 @@ def test_index_policy_force_and_explain(graph, engine):
     chain = [n({"id": 0}), e_forward(hops=1)]
     rep_off = graph.gfql_explain(chain, index_policy="off", engine=engine)
     assert rep_off["used_index"] is False
+    assert rep_off["decision_code"] == "policy_off"
+    assert rep_off["decision_reason"] == "policy=off"
     rep_force = graph.gfql_explain(chain, index_policy="force", engine=engine)
     assert rep_force["used_index"] is True
+    assert rep_force["decision_code"] == "index_selected"
     # results identical regardless of policy
     r_scan = graph.gfql(chain, engine=engine)
     r_force = graph.gfql(chain, index_policy="force", engine=engine)
     assert _sig(r_scan) == _sig(r_force)
+    assert rep_force["error"] is None
 
 
 @pytest.mark.parametrize("engine", ENGINES)
@@ -261,6 +265,7 @@ def test_explain_exposes_planner_diagnostics(graph, engine):
     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_code"] == "policy_off"
     assert rep_off["decision_reason"] == "policy=off", rep_off
 
 
@@ -506,7 +511,7 @@ def test_index_min_two_bounded_range_scans_pandas(graph):
         indexed = _force(graph, "pandas").hop(nodes=seeds, engine="pandas", **kwargs)
     assert _sig(base) == _sig(indexed)
     assert not any(step["path"] == "index" for step in steps), steps
-    assert any(step["decision_reason"] == "query not index-coverable" for step in steps), steps
+    assert any(step["decision_code"] == "not_index_coverable" for step in steps), steps
 
 @pytest.mark.parametrize("engine", ENGINES)
 def test_index_duplicate_node_ids(engine):
@@ -848,7 +853,8 @@ def test_chain_range_with_auto_labels_stays_on_scan(typed_graph):
     assert _sig_typed(base) == _sig_typed(indexed)
     assert rep["used_index"] is False, (engine, rep)
     assert not any(step["path"] == "index" for step in steps), steps
-    assert any(step["decision_reason"] == "query not index-coverable" for step in steps), steps
+    assert rep["decision_code"] == "not_index_coverable", rep
+    assert any(step["decision_code"] == "not_index_coverable" for step in steps), steps
 
 
 def test_rebind_edges_revalidates_after_shallow_augmentation():