feat(gfql): add typed node segment indexes for indexed hops

CHANGELOG.md

@@ -12,6 +12,8 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 
 - **GFQL relationship-type adjacency sidecars** (#2184): `gfql_index_edges(..., type_column=..., type_values=...)` explicitly builds forward and reverse adjacency indexes for selected string or integer relationship values. Matching queries can narrow candidate edge rows while retaining the full edge predicate and canonical results; queries without a live sidecar use ordinary adjacency. Sidecars are built at graph setup, share neighbor arrays with ordinary indexes, and appear as `typed:` rows with incremental CSR bytes in `show_indexes()`. Typed options are keyword-only. cuDF builds may add overhead on dense-community shapes, and sidecars must be rebuilt after rebinding or changing edges.
 
+- **GFQL node-type wavefront segments**: `gfql_index_node_types(..., type_column=..., type_values=...)` builds node row-position segments for selected scalar types or Boolean labels. Matching seeded wavefront hops can narrow source and destination candidates while retaining canonical node filters and results; queries without a live segment use the ordinary paths. Node segments are built at graph setup and appear as `typed_node:` `node_type_segment` rows in `show_indexes()`; selective drop supports all segments, one column, or one segment.
+
 - **GFQL Cypher aggregates over relationship rows**: in a single required `MATCH`, aggregates such as `MATCH (a)-[r]->(b) RETURN a, count(*), sum(r.w), collect(b.id)` run on per-path binding rows, so group keys and arguments can reference any node or relationship alias, including `DISTINCT`, post-aggregate arithmetic (`count(b) + 1`) and multi-alias keys. These shapes previously raised unsupported-query errors.
 
 ### Fixed

bin/test-polars.sh

@@ -193,6 +193,7 @@ POLARS_TEST_FILES=(
     graphistry/tests/compute/gfql/test_unseeded_binding_limit.py
     graphistry/tests/compute/gfql/index/test_indexed_bindings.py
     graphistry/tests/compute/gfql/index/test_typed_adjacency.py
+    graphistry/tests/compute/gfql/index/test_node_type_segments.py
     graphistry/tests/compute/gfql/index/test_indexed_bindings_membership_seed.py
     graphistry/tests/compute/gfql/index/test_index_ddl_in_one_call.py
     graphistry/tests/compute/gfql/cypher/test_in_list_seeds_the_pattern.py

docs/source/gfql/indexing.rst

@@ -200,6 +200,30 @@ The ``typed:`` rows in ``show_indexes()`` report incremental CSR bytes; they
 share the ordinary adjacency's neighbor arrays. Build for the engine used by
 the query, and rebuild after binding a new edge frame or changing edges in place.
 
+Select node types or labels
+~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+For repeated hops that filter nodes by type or label, build row-position sidecars
+the same way as for relationship types (see the previous section):
+``gfql_index_node_types(type_column=..., type_values=...)`` is the node-frame dual
+of ``gfql_index_edges``. A scalar column partitions rows; a Boolean
+``label__<Label>`` column records membership, so one node can appear in several
+segments. Like edge sidecars, these are an explicit load-time cost
+(``gfql_index_all()`` does not build them), queries fall back to the ordinary
+paths and retain the same results, and GFQL still applies the complete node
+predicate after lookup. Rebuild after binding a new node frame or changing nodes
+in place; ``show_indexes()`` reports them as ``node_type_segment`` rows. Drop all
+with ``drop_index("node_type_segment")``, one column with ``column=...``, or one
+segment with both ``column=...`` and ``value=...``.
+
+.. code-block:: python
+
+   labeled_g = graphistry.nodes(pd.DataFrame({
+       "id": [0, 1, 2], "type": ["PERSON", "ROBOT", "PERSON"]}), "id")
+   labeled_g = labeled_g.gfql_index_node_types(
+       type_column="type", type_values=["PERSON"]
+   )
+
 Start from a property column
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 

graphistry/compute/ComputeMixin.py

@@ -32,6 +32,7 @@
 
 if TYPE_CHECKING:
     from graphistry.compute.gfql.index.types import IndexKind
+    from graphistry.compute.gfql.index.registry import PartitionValue
     from graphistry.compute.gfql.index.explain import GfqlExplainReport
     from graphistry.compute.gfql.index.policy import IndexPolicy
     from graphistry.compute.gfql.query_types import GFQLQuery
@@ -663,10 +664,13 @@ def create_index(self, kind, *, column=None, name=None, engine='auto'):
         from graphistry.compute.gfql.index import create_index as _ci
         return _ci(self, kind, column=column, name=name, engine=engine)
 
-    def drop_index(self, kind: Optional['IndexKind'] = None, *, column: Optional[str] = None) -> 'Plottable':
+    def drop_index(
+        self, kind: Optional['IndexKind'] = None, *,
+        column: Optional[str] = None, value: Optional['PartitionValue'] = None,
+    ) -> 'Plottable':
         """Drop one resident GFQL index (by kind, or one property index by column) or all (kind=None). Idempotent; returns a new Plottable."""
         from graphistry.compute.gfql.index import drop_index as _di
-        return _di(self, kind, column=column)
+        return _di(self, kind, column=column, value=value)
 
     def show_indexes(self, engine: EngineAbstractType = 'auto'):
         """Return a pandas DataFrame describing resident GFQL indexes (name, kind, column, valid, usable). Empty if none.
@@ -690,6 +694,17 @@ def gfql_index_edges(self, direction='both', engine='auto', *,
         from graphistry.compute.gfql.index import gfql_index_edges as _gie
         return _gie(self, direction, type_column=type_column, type_values=type_values, engine=engine)
 
+    def gfql_index_node_types(self, engine: EngineAbstractType = 'auto', *,
+                              type_column: Optional[str] = None,
+                              type_values: Optional[Sequence[Union[str, int, bool]]] = None) -> 'Plottable':
+        """Build node type/label segments — the dual of ``gfql_index_edges``'s
+        typed sidecars. With ``type_column`` and ``type_values``, registers
+        per-value node row-position segments (scalar columns partition; Boolean
+        ``label__<Label>`` columns give multi-label membership). Returns a new Plottable.
+        """
+        from graphistry.compute.gfql.index import gfql_index_node_types as _gint
+        return _gint(self, type_column=type_column, type_values=type_values, engine=engine)
+
     def gfql_index_all(self, col_stats_by_type: bool = False, engine='auto'):
         """Convenience: build all GFQL physical indexes (both edge adjacencies + node_id). Returns a new Plottable.
 

graphistry/compute/gfql/index/__init__.py

@@ -2,8 +2,8 @@
 seeded traversal.
 
 Public surface (see api.py): ``create_index``, ``drop_index``, ``show_indexes``,
-``gfql_index_edges``, ``gfql_index_all``, ``gfql_index_node_props``, and the
-planner entry ``maybe_index_hop``.
+``gfql_index_edges``, ``gfql_index_all``, ``gfql_index_node_props``,
+``gfql_index_node_types``, and the planner entry ``maybe_index_hop``.
 These are wired onto Plottable via ComputeMixin.
 """
 from .types import (
@@ -12,13 +12,15 @@
 )
 from .registry import (
     GfqlIndexRegistry, EMPTY_REGISTRY,
-    EDGE_OUT_ADJ, EDGE_IN_ADJ, NODE_ID, NODE_PROP, EDGE_PROP, ADJ_KINDS, ALL_KINDS,
+    EDGE_OUT_ADJ, EDGE_IN_ADJ, NODE_ID, NODE_PROP, EDGE_PROP,
+    NODE_TYPE_SEGMENT, ADJ_KINDS, ALL_KINDS,
     AdjacencyIndex, NodeIdIndex, NodePropIndex,
 )
 from .api import (
     create_index, drop_index, show_indexes, gfql_index_edges, gfql_index_all,
     gfql_index_col_stats,
     gfql_index_node_props,
+    gfql_index_node_types,
     GfqlIndexUnsupportedError,
     GfqlIndexNotImplementedError,
     get_registry, set_registry, get_index_policy, with_index_policy, maybe_index_hop,
@@ -36,10 +38,11 @@
     "AdjacencyIndexKind", "ArrayLike", "ArrayNamespace", "EdgeIndexDirection",
     "HopDirection", "IndexBackend", "IndexKind", "IndexTraceStep",
     "GfqlIndexRegistry", "EMPTY_REGISTRY",
-    "EDGE_OUT_ADJ", "EDGE_IN_ADJ", "NODE_ID", "NODE_PROP", "EDGE_PROP", "ADJ_KINDS", "ALL_KINDS",
+    "EDGE_OUT_ADJ", "EDGE_IN_ADJ", "NODE_ID", "NODE_PROP", "EDGE_PROP",
+    "NODE_TYPE_SEGMENT", "ADJ_KINDS", "ALL_KINDS",
     "AdjacencyIndex", "NodeIdIndex", "NodePropIndex",
     "create_index", "drop_index", "show_indexes", "gfql_index_edges",
-    "gfql_index_all", "gfql_index_col_stats", "gfql_index_node_props", "GfqlIndexUnsupportedError", "GfqlIndexNotImplementedError",
+    "gfql_index_all", "gfql_index_col_stats", "gfql_index_node_props", "gfql_index_node_types", "GfqlIndexUnsupportedError", "GfqlIndexNotImplementedError",
     "get_registry", "with_index_policy", "set_registry", "get_index_policy", "maybe_index_hop", "index_name",
     "index_trace",
     "CreateIndex", "DropIndex", "ShowIndexes", "IndexOp", "apply_index_op",

graphistry/compute/gfql/index/api.py

@@ -17,9 +17,10 @@
 from graphistry.Plottable import Plottable
 from .registry import (
     AdjacencyIndex, ColStatsFact, PartitionValue, ColStatsRole, GfqlIndexRegistry, EMPTY_REGISTRY, NodeIdIndex,
-    EDGE_OUT_ADJ, EDGE_IN_ADJ, NODE_ID, NODE_PROP, EDGE_PROP, ADJ_KINDS, ALL_KINDS, PROPERTY_ROLES,
+    EDGE_OUT_ADJ, EDGE_IN_ADJ, NODE_ID, NODE_PROP, EDGE_PROP, NODE_TYPE_SEGMENT,
+    ADJ_KINDS, ALL_KINDS, PROPERTY_ROLES,
 )
-from .build import build_adjacency_index, build_node_id_index, build_property_index, build_typed_adjacency_indexes
+from .build import build_adjacency_index, build_node_id_index, build_node_type_segment, build_property_index, build_typed_adjacency_indexes
 from .traverse import index_seeded_hop
 from .cost import cost_gate_frac, seed_deg_sum, seed_id_array
 from .policy import IndexPolicy, validate_index_policy
@@ -450,14 +451,21 @@ def create_index(
 
 
 def drop_index(
-    g: Plottable, kind: Optional[IndexKind] = None, *, column: Optional[str] = None
+    g: Plottable, kind: Optional[IndexKind] = None, *,
+    column: Optional[str] = None, value: Optional[PartitionValue] = None,
 ) -> Plottable:
     """Drop one index (by kind, or one property index by column) or all (kind=None).
 
-    Idempotent."""
+    For ``node_type_segment``, omit ``column`` to drop all segments, pass
+    ``column`` to drop that column's segments, or pass ``column`` and ``value``
+    to drop one segment. Idempotent."""
     registry = get_registry(g)
     if kind is None:
         return _attach(g, EMPTY_REGISTRY)
+    if kind == NODE_TYPE_SEGMENT:
+        if value is not None and column is None:
+            raise ValueError("dropping one node type segment requires type_column")
+        return _attach(g, registry.without_node_segments(column, value))
     if kind in (NODE_PROP, EDGE_PROP) and column is not None:
         return _attach(g, registry.without_property("nodes" if kind == NODE_PROP else "edges", column))
     return _attach(g, registry.without(kind))
@@ -593,6 +601,27 @@ def show_indexes(
             "nbytes": nbytes, "valid": valid, "query_engine": query_engine.value,
             "usable": usable, "reason": reason,
         })
+    for column, value in sorted(
+        registry.node_type_segments, key=lambda key: (key[0], str(key[1])),
+    ):
+        segment = registry.node_type_segments[(column, value)]
+        valid = (
+            g._nodes is not None
+            and registry.get_node_segment_valid(column, value, g._nodes, segment.engine)
+            is segment
+        )
+        usable, reason = _index_usability(NODE_TYPE_SEGMENT, segment.engine, valid, query_engine)
+        nbytes = sum(int(getattr(segment, name).nbytes) for name in (
+            "keys_sorted", "group_offsets", "row_positions",
+        ))
+        rows.append({
+            "name": f"typed_node:{column}={value}",
+            "kind": NODE_TYPE_SEGMENT, "key_col": segment.key_col,
+            "engine": segment.engine.value, "backend": segment.backend,
+            "n_keys": segment.n_keys, "n_rows": int(segment.row_positions.shape[0]),
+            "nbytes": nbytes, "valid": valid, "query_engine": query_engine.value,
+            "usable": usable, "reason": reason,
+        })
     for kind, role in PROPERTY_ROLES:
         frame = g._nodes if role == "nodes" else g._edges
         for column in sorted(registry.property_indexes(role)):
@@ -621,6 +650,31 @@ def show_indexes(
     return pd.DataFrame(rows, columns=cols)
 
 
+def _parse_type_partition(
+    label: str, type_column: Optional[str],
+    type_values: Optional[Sequence[PartitionValue]], allow_bool: bool,
+) -> Optional[List[PartitionValue]]:
+    """Shared type_column/type_values validation for ``gfql_index_edges`` /
+    ``gfql_index_node_types``: returns the parsed values, or None if untyped."""
+    if (type_column is None) != (type_values is None):
+        raise ValueError(f"{label} type index requires both type_column and type_values")
+    if type_column is None or type_values is None:
+        return None
+    if not isinstance(type_column, str) or not type_column:
+        raise TypeError(f"{label} type column must be a non-empty string")
+    if isinstance(type_values, (str, bytes)):
+        raise TypeError(f"{label} type index values must be a sequence, not a scalar string")
+    values = list(type_values)
+    if not values:
+        raise ValueError(f"{label} type index requires at least one value")
+    if allow_bool:
+        if any(not isinstance(value, (str, int, bool)) for value in values):
+            raise TypeError(f"{label} type index values must be strings, integers, or booleans")
+    elif any(isinstance(value, bool) or not isinstance(value, (str, int)) for value in values):
+        raise TypeError(f"{label} type index values must be strings or integers")
+    return values
+
+
 def gfql_index_edges(
     g: Plottable, direction: EdgeIndexDirection = "both",
     engine: EngineAbstractType = EngineAbstract.AUTO, *,
@@ -638,21 +692,9 @@ def gfql_index_edges(
     canonical edge predicate filtering after candidate lookup.
     """
     typed_column: Optional[str] = None
-    if (type_column is None) != (type_values is None):
-        raise ValueError("edge type index requires both type_column and type_values")
-    if type_column is not None and type_values is not None:
-        if not isinstance(type_column, str) or not type_column:
-            raise TypeError("edge type column must be a non-empty string")
-        if isinstance(type_values, (str, bytes)):
-            raise TypeError("edge type index values must be a sequence, not a scalar string")
-        values = list(type_values)
-        if not values:
-            raise ValueError("edge type index requires at least one value")
-        if any(isinstance(value, bool) or not isinstance(value, (str, int)) for value in values):
-            raise TypeError("edge type index values must be strings or integers")
+    values = _parse_type_partition("edge", type_column, type_values, allow_bool=False)
+    if values is not None:
         typed_column = type_column
-    else:
-        values = None
     if direction in ("forward", "both"):
         g = create_index(g, EDGE_OUT_ADJ, engine=engine)
     if direction in ("reverse", "both"):
@@ -708,6 +750,42 @@ def gfql_index_edges(
     return _attach(g, registry)
 
 
+def gfql_index_node_types(
+    g: Plottable, engine: EngineAbstractType = EngineAbstract.AUTO, *,
+    type_column: Optional[str] = None,
+    type_values: Optional[Sequence[PartitionValue]] = None,
+) -> Plottable:
+    """Build node type/label segments: the dual of ``gfql_index_edges``'s typed
+    edge sidecars.
+
+    With ``type_column`` and ``type_values``, register per-value row-position
+    segments over the node frame — a scalar type column gives partition semantics,
+    Boolean ``label__<Label>`` columns give membership semantics (one node may
+    appear in several segments). Membership is decided once at setup by the
+    canonical engine predicate, so segment selection is parity-exact with the
+    scan's node filter, including null rows. Segments are an explicit setup cost:
+    ``gfql_index_all()`` does not build them, and queries without a matching,
+    live segment retain the ordinary scan path.
+    """
+    values = _parse_type_partition("node", type_column, type_values, allow_bool=True)
+    if not values or type_column is None:
+        return g
+    from graphistry.compute.ComputeMixin import _coerce_input_formats
+
+    eng = resolve_index_engine(engine, g)
+    g = _coerce_input_formats(g, eng)
+    nodes = g._nodes
+    if nodes is None or type_column not in nodes.columns:
+        raise ValueError("node type index requires bound nodes and a present type column")
+    registry = get_registry(g)
+    for value in dict.fromkeys(values):
+        if registry.get_node_segment_valid(type_column, value, nodes, eng) is not None:
+            continue
+        segment = build_node_type_segment(nodes, type_column, value, eng)
+        registry = registry.with_node_segment(type_column, value, segment)
+    return _attach(g, registry)
+
+
 def gfql_index_node_props(g: Plottable, columns: Sequence[str],
                           engine: EngineAbstractType = EngineAbstract.AUTO) -> Plottable:
     """Convenience: build node property indexes for ``columns`` (skips unindexable).
@@ -1078,10 +1156,17 @@ def _hop_is_index_coverable(
     if any(x is not None for x in (
         min_hops if (min_hops is not None and min_hops > 1) else None,
         output_min_hops, output_max_hops, label_node_hops, label_edge_hops,
-        source_node_match, destination_node_match,
         source_node_query, destination_node_query, edge_query, target_wave_front,
     )):
         return False
+    # Node segments: simple scalar equality only; residency declines later to scan.
+    from .traverse import is_simple_equality_node_match
+    if (source_node_match is not None or destination_node_match is not None) and not (
+        return_as_wave_front
+        and is_simple_equality_node_match(source_node_match)
+        and is_simple_equality_node_match(destination_node_match)
+    ):
+        return False
     # Typed-edge edge_match: coverable only as a simple scalar-equality filter on the
     # wavefront (chain/Cypher) path, where index_seeded_hop applies it per-hop parity-
     # exact with the scan's filter_edges_by_dict. Predicate/membership forms and the
@@ -1148,6 +1233,25 @@ def _ensure_indexes(
     return registry
 
 
+def _node_segment_trace_state(
+    registry: GfqlIndexRegistry,
+    g: Plottable,
+    node_col: Optional[str],
+    engine: Engine,
+    node_match: Optional[object],
+) -> str:
+    if node_match is None:
+        return "not_present"
+    from .traverse import is_simple_equality_node_match
+    if not is_simple_equality_node_match(node_match):
+        return "unsupported"
+    assert isinstance(node_match, dict)
+    column, value = next(iter(node_match.items()))
+    if not isinstance(value, (str, int)) or node_col is None or g._nodes is None:
+        return "unsupported"
+    return "resident" if registry.get_node_segment_valid(column, value, g._nodes, engine) is not None else "missing_or_stale"
+
+
 def maybe_index_hop(
     g: Plottable, engine: Engine, *, nodes: Optional[DataFrameT], hops: Optional[int], direction: HopDirection, return_as_wave_front: bool,
     to_fixed_point: bool = False, policy: Optional[IndexPolicy] = "use", **rest: object,
@@ -1270,12 +1374,19 @@ def _bail(reason: str, decision_code: IndexDecisionCode) -> Optional[Plottable]:
     # (compute/hop.py); the index must run the SAME number of accumulating hops.
     # (regression: max_hops was passed through *rest and silently ignored — the index ran
     # `hops` (default 1) while the scan ran max_hops → wrong answer.)
+    if trace:
+        diag["node_segments"] = {
+            "source": _node_segment_trace_state(registry, g, node_col, engine, rest.get("source_node_match")),
+            "destination": _node_segment_trace_state(registry, g, node_col, engine, rest.get("destination_node_match")),
+        }
     eff_hops = max_hops if max_hops is not None else hops
     result = index_seeded_hop(
         g, registry, nodes=nodes, node_col=node_col, src=src, dst=dst, engine=engine,
         hops=eff_hops, to_fixed_point=to_fixed_point, direction=direction,
         return_as_wave_front=return_as_wave_front,
         edge_match=cast(Optional[dict], rest.get("edge_match")),
+        source_node_match=rest.get("source_node_match"),
+        destination_node_match=rest.get("destination_node_match"),
     )
     if trace:
         engine_mismatch_reason = (