diff --git a/CHANGELOG.md b/CHANGELOG.md
index b0f84c6804..3cb1419f31 100644
--- a/CHANGELOG.md
+++ b/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
diff --git a/bin/test-polars.sh b/bin/test-polars.sh
index 895d79a87c..517b3922e6 100755
--- a/bin/test-polars.sh
+++ b/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
diff --git a/docs/source/gfql/indexing.rst b/docs/source/gfql/indexing.rst
index 0bfa191f7e..b8e3f35c9b 100644
--- a/docs/source/gfql/indexing.rst
+++ b/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
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 
diff --git a/graphistry/compute/ComputeMixin.py b/graphistry/compute/ComputeMixin.py
index 4687f921ad..52d1fc3000 100644
--- a/graphistry/compute/ComputeMixin.py
+++ b/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.
 
diff --git a/graphistry/compute/gfql/index/__init__.py b/graphistry/compute/gfql/index/__init__.py
index 6e3aec5371..c93c9b90f7 100644
--- a/graphistry/compute/gfql/index/__init__.py
+++ b/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",
diff --git a/graphistry/compute/gfql/index/api.py b/graphistry/compute/gfql/index/api.py
index 816407eb56..ab19c8a58f 100644
--- a/graphistry/compute/gfql/index/api.py
+++ b/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 = (
diff --git a/graphistry/compute/gfql/index/build.py b/graphistry/compute/gfql/index/build.py
index 73e554fa65..472fb94709 100644
--- a/graphistry/compute/gfql/index/build.py
+++ b/graphistry/compute/gfql/index/build.py
@@ -178,6 +178,68 @@ def build_typed_adjacency_indexes(
     return out, into
 
 
+def build_node_type_segment(
+    nodes: DataFrameT,
+    type_column: str,
+    type_value: PartitionValue,
+    engine: Engine,
+) -> NodePropIndex:
+    """Row positions of nodes matching one scalar/Boolean type value, as a
+    single-bucket :class:`NodePropIndex`.
+
+    The dual of the typed edge sidecar: the canonical engine predicate
+    (``filter_by_dict``) decides membership ONCE at setup, so segment selection is
+    parity-exact with the scan's node filter by construction — including null rows
+    (null == value filters out on the scan path, and the canonical predicate encodes
+    that here). Original node row positions are retained through a position column,
+    mirroring ``build_typed_adjacency_indexes``; the user's ``.nodes`` frame is never
+    reordered. Boolean label columns give MEMBERSHIP semantics (a node appears in
+    several segments); a scalar type column gives partition semantics — the storage
+    is the same either way.
+    """
+    from graphistry.compute.filter_by_dict import filter_by_dict
+
+    xp, backend = array_namespace(engine)
+    position_column = "__gfql_typed_node_position__"
+    while position_column in nodes.columns:
+        position_column += "_"
+    if engine in (Engine.POLARS, Engine.POLARS_GPU):
+        import polars as pl
+
+        if not isinstance(nodes, pl.DataFrame):
+            raise TypeError("typed Polars node segment requires an eager Polars node frame")
+        positioned: DataFrameT = nodes.select([type_column]).with_row_index(position_column)
+    else:
+        positioned = nodes[[type_column]].assign(
+            **{position_column: xp.arange(len(nodes), dtype=xp.int64)}
+        )
+    matching = filter_by_dict(positioned, {type_column: type_value}, engine=engine.value)
+    # Stable int64 positions preserve cross-hop node identities.
+    row_positions = xp.asarray(
+        col_to_array(matching, position_column, engine), dtype=xp.int64,
+    )
+    if engine == Engine.CUDF and isinstance(type_value, str):
+        # cupy cannot encode text scalars; segments are looked up by registry key, so retain string values natively.
+        import cudf  # type: ignore[import]
+
+        keys = cast(ArrayLike, cudf.Series([type_value]))
+    else:
+        keys = xp.asarray([type_value])
+    return NodePropIndex(
+        key_col=type_column,
+        keys_sorted=keys,
+        group_offsets=xp.asarray([0, int(row_positions.shape[0])], dtype=xp.int64),
+        row_positions=row_positions,
+        backend=backend,
+        engine=engine,
+        fingerprint=frame_fingerprint(nodes, (type_column,), engine),
+        source_ref=nodes,
+        n_nodes=len(nodes),
+        n_keys=1,
+        min_group_count=int(row_positions.shape[0]),
+    )
+
+
 def build_node_id_index(
     nodes: DataFrameT,
     node_col: str,
@@ -210,7 +272,7 @@ def build_node_id_index(
         backend=backend,
         engine=engine,
         fingerprint=frame_fingerprint(nodes, (node_col,), engine),
-        source_ref=cast(DataFrameT, nodes),
+        source_ref=nodes,
         n_nodes=n_keys,
     )
 
@@ -283,7 +345,7 @@ def build_property_index(
         backend=backend,
         engine=engine,
         fingerprint=frame_fingerprint(nodes, (column,), engine),
-        source_ref=cast(DataFrameT, nodes),
+        source_ref=nodes,
         string_keys=dictionary,
         string_key_positions=string_key_positions,
         string_key_positions_bytes=string_key_positions_bytes,
diff --git a/graphistry/compute/gfql/index/cypher_ddl.py b/graphistry/compute/gfql/index/cypher_ddl.py
index b751eedf5e..3a3056f20e 100644
--- a/graphistry/compute/gfql/index/cypher_ddl.py
+++ b/graphistry/compute/gfql/index/cypher_ddl.py
@@ -23,7 +23,7 @@
 from .errors import is_not_yet_implemented_kind, not_implemented_kind_error
 from .registry import ALL_KINDS
 
-_KIND = r"(?P<kind>edge_out_adj|edge_in_adj|node_id|node_prop|edge_prop)"
+_KIND = r"(?P<kind>edge_out_adj|edge_in_adj|node_id|node_prop|edge_prop|node_type_segment)"
 
 _ON_COL = r"(?:\s+ON(?:\s*\(\s*(?P<col>[A-Za-z_]\w*)\s*\)|\s+(?P<col2>[A-Za-z_]\w*)))?"
 _CREATE_PATTERN = (
diff --git a/graphistry/compute/gfql/index/registry.py b/graphistry/compute/gfql/index/registry.py
index 30852ecc9a..a8658bfc7a 100644
--- a/graphistry/compute/gfql/index/registry.py
+++ b/graphistry/compute/gfql/index/registry.py
@@ -23,9 +23,12 @@
 NODE_ID: IndexKind = "node_id"
 NODE_PROP: IndexKind = "node_prop"
 EDGE_PROP: IndexKind = "edge_prop"
+NODE_TYPE_SEGMENT: IndexKind = "node_type_segment"
 
 ADJ_KINDS: Tuple[AdjacencyIndexKind, ...] = (EDGE_OUT_ADJ, EDGE_IN_ADJ)
-ALL_KINDS: Tuple[IndexKind, ...] = (EDGE_OUT_ADJ, EDGE_IN_ADJ, NODE_ID, NODE_PROP, EDGE_PROP)
+ALL_KINDS: Tuple[IndexKind, ...] = (
+    EDGE_OUT_ADJ, EDGE_IN_ADJ, NODE_ID, NODE_PROP, EDGE_PROP, NODE_TYPE_SEGMENT,
+)
 
 FrameFingerprint = Tuple[int, Tuple[str, ...], str]
 
@@ -152,7 +155,7 @@ class CategoryIndex:
 #: column are the same registry key. That is unreachable in practice (a column
 #: is bool-dtyped or int-dtyped, not both) and harmless where it is reachable
 #: (a query asking ``flag == 1`` of a bool column does select the True rows).
-PartitionValue = Union[str, int]
+PartitionValue = Union[str, int, bool]
 
 
 @dataclass(frozen=True)
@@ -263,6 +266,8 @@ class GfqlIndexRegistry:
     """Immutable kind -> index map. ``with_index`` / ``without`` return copies."""
     indexes: Dict[IndexKind, Union[AdjacencyIndex, NodeIdIndex]] = field(default_factory=dict)
     typed_adjacencies: Dict[Tuple[str, PartitionValue, AdjacencyIndexKind], AdjacencyIndex] = field(default_factory=dict)
+    # Unlike typed edge sidecars (per-value instances of adjacency kinds), these row-position facts keyed (column, value) need their own kind and selective drop.
+    node_type_segments: Dict[Tuple[str, PartitionValue], NodePropIndex] = field(default_factory=dict)
     # Property indexes are keyed by COLUMN, not kind: a graph may carry several.
     node_props: Dict[str, NodePropIndex] = field(default_factory=dict)
     edge_props: Dict[str, NodePropIndex] = field(default_factory=dict)
@@ -302,6 +307,31 @@ def get_typed_adjacency_valid(
             return None
         return index
 
+    def with_node_segment(
+        self, column: str, value: PartitionValue, index: "NodePropIndex",
+    ) -> "GfqlIndexRegistry":
+        segments = dict(self.node_type_segments)
+        segments[(column, value)] = index
+        return replace(self, node_type_segments=segments)
+
+    def get_node_segment_valid(
+        self, column: str, value: PartitionValue,
+        nodes: Optional[DataFrameT], engine: Engine,
+    ) -> Optional["NodePropIndex"]:
+        """Live node type segment for (column, value), or None.
+
+        A miss (never built, stale frame, foreign engine) declines to the scan —
+        never a wrong answer. Residency matters for parity: an EMPTY segment is a
+        real zero-row answer, distinct from an absent one."""
+        index = self.node_type_segments.get((column, value))
+        if index is None or nodes is None or index.engine != engine:
+            return None
+        if index.source_ref is not None and index.source_ref is not nodes:
+            return None
+        if index.fingerprint != frame_fingerprint(nodes, (column,), engine):
+            return None
+        return index
+
     def with_node_prop(self, column: str, index: "NodePropIndex") -> "GfqlIndexRegistry":
         props = dict(self.node_props)
         props[column] = index
@@ -463,6 +493,19 @@ def get_temporal_text_valid(
             return None
         return fact
 
+    def without_node_segments(
+        self, column: Optional[str] = None, value: Optional[PartitionValue] = None,
+    ) -> "GfqlIndexRegistry":
+        segments = dict(self.node_type_segments)
+        if column is None:
+            segments.clear()
+        else:
+            if value is None:
+                segments = {key: idx for key, idx in segments.items() if key[0] != column}
+            else:
+                segments.pop((column, value), None)
+        return replace(self, node_type_segments=segments)
+
     def without_temporal_text(self) -> "GfqlIndexRegistry":
         return replace(self, temporal_text={})
 
@@ -590,7 +633,7 @@ def kinds(self) -> Tuple[IndexKind, ...]:
         return cast(Tuple[IndexKind, ...], tuple(sorted(self.indexes.keys())))
 
     def is_empty(self) -> bool:
-        return not self.indexes and not self.typed_adjacencies and not self.node_props and not self.edge_props and not self.col_stats
+        return not self.indexes and not self.typed_adjacencies and not self.node_props and not self.edge_props and not self.col_stats and not self.node_type_segments
 
     def get_valid(self, kind: IndexKind, df: DataFrameT, cols: Tuple[str, ...], engine: Engine) -> Optional[Union[AdjacencyIndex, NodeIdIndex]]:
         """Return the index for ``kind`` only if its fingerprint still matches the
diff --git a/graphistry/compute/gfql/index/traverse.py b/graphistry/compute/gfql/index/traverse.py
index 139282245c..fcfaa7dae6 100644
--- a/graphistry/compute/gfql/index/traverse.py
+++ b/graphistry/compute/gfql/index/traverse.py
@@ -7,10 +7,10 @@
 
 Covered (v1): seeded (nodes given), integer ``hops`` >= 1 or ``to_fixed_point``,
 direction forward/reverse/undirected, ``return_as_wave_front``, and a simple
-scalar-equality ``edge_match`` (typed edges, e.g. Cypher ``-[:KNOWS]->``) applied on
-the wavefront path. Not covered (returns None): predicate/membership edge_match,
-source/destination match or query, edge_query, target_wave_front, min_hops>1,
-output_min/max_hops, labeling, missing node table.
+scalar-equality ``edge_match`` or source/destination node match on the wavefront
+path. Not covered (returns None): predicate/membership edge_match, node queries,
+edge_query, target_wave_front, min_hops>1, output_min/max_hops, labeling,
+missing node table.
 """
 from __future__ import annotations
 
@@ -28,7 +28,7 @@
 from .lookup import lookup_edge_rows, lookup_node_rows
 from .registry import EDGE_OUT_ADJ, EDGE_IN_ADJ, NODE_ID, AdjacencyIndex, GfqlIndexRegistry, NodeIdIndex
 from .types import (
-    ArrayLike, EdgeMatch, HopDirection, ScalarMatchValue, SimpleEqualityEdgeMatch,
+    ArrayLike, ArrayNamespace, EdgeMatch, HopDirection, ScalarMatchValue, SimpleEqualityEdgeMatch,
 )
 
 _EAGER_MASK_SWITCH_DIVISOR = 8
@@ -113,6 +113,58 @@ def is_simple_equality_edge_match(
     return True
 
 
+def is_simple_equality_node_match(
+    node_match: Optional[object],
+) -> bool:
+    """True iff ``node_match`` is None or a SINGLE-key plain scalar-equality dict.
+
+    The only node-match shape the node type-segment path accelerates parity-exact
+    in v1: the segment covers one (column, value) pair, and residual keys would
+    need id->row candidate filtering the segment path does not yet plumb. Boolean
+    ``True``/``False`` values ARE scalar here (the canonical Cypher ``label__X``
+    shape), unlike the typed-edge value check which rejects them.
+    """
+    if node_match is None:
+        return True
+    from graphistry.compute.predicates.ASTPredicate import ASTPredicate
+    from graphistry.compute.filter_by_dict import _is_membership_filter_value
+    if not isinstance(node_match, dict) or len(node_match) != 1:
+        return False
+    for column, value in node_match.items():
+        if not isinstance(column, str):
+            return False
+        if isinstance(value, ASTPredicate) or _is_membership_filter_value(value) or isinstance(value, dict):
+            return False
+    return True
+
+
+def _sorted_id_mask(values: ArrayLike, sorted_ids: ArrayLike, xp: ArrayNamespace) -> ArrayLike:
+    """Boolean mask over ``values`` marking membership in sorted-unique ``ids``.
+
+    Promotes BOTH sides to a common dtype (never narrows the query side — the
+    int64-cast lesson from the typed edge positions applies to id comparisons too).
+    """
+    n = int(values.shape[0])
+    if n == 0 or int(sorted_ids.shape[0]) == 0:
+        return xp.zeros(n, dtype=bool)
+    v = values
+    keys = sorted_ids
+    if v.dtype != keys.dtype:
+        common = xp.promote_types(v.dtype, keys.dtype)
+        v = v.astype(common)
+        keys = keys.astype(common)
+    pos = xp.searchsorted(keys, v)
+    pos_clipped = xp.where(pos < int(keys.shape[0]), pos, int(keys.shape[0]) - 1)
+    return keys[pos_clipped] == v
+
+
+def _sorted_intersection(a: ArrayLike, b: ArrayLike, xp: ArrayNamespace) -> ArrayLike:
+    """Sorted-unique intersection of two sorted-unique id arrays."""
+    if int(a.shape[0]) == 0 or int(b.shape[0]) == 0:
+        return a[:0]
+    return a[_sorted_id_mask(a, b, xp)]
+
+
 class _EdgeMatchRowFilter:
     """Evaluates a simple-equality ``edge_match`` on the CSR-matched edge rows only.
 
@@ -255,6 +307,36 @@ def _build_edge_row_filter(
         return None
 
 
+def _node_match_allowed_ids(
+    g: Plottable, registry: GfqlIndexRegistry, node_match: Optional[object],
+    node_col: Optional[str], engine: Engine,
+) -> Optional[ArrayLike]:
+    """Sorted-unique ids admitted by one single-key scalar node match, or None to
+    decline the indexed path (caller falls back to the scan).
+
+    Membership comes from the LIVE node type segment, whose rows were decided by
+    the canonical filter_by_dict predicate at build time — parity with the scan's
+    allowed-id construction by construction. An EMPTY segment is a real zero-row
+    answer (empty sorted array), NOT a decline."""
+    from .engine_arrays import array_namespace, col_to_array, sorted_unique, take_rows
+
+    if not is_simple_equality_node_match(node_match) or node_col is None or g._nodes is None:
+        return None
+    assert isinstance(node_match, dict)
+    xp, _backend = array_namespace(engine)
+    (column, value), = node_match.items()
+    if not isinstance(value, (str, int)):  # bool is an int subclass; guards exotic literals
+        return None
+    segment = registry.get_node_segment_valid(column, value, g._nodes, engine)
+    if segment is None:
+        return None
+    segment_rows = segment.row_positions
+    if int(segment_rows.shape[0]) == 0:
+        return segment_rows[:0]
+    ids = col_to_array(take_rows(g._nodes, segment_rows, engine), node_col, engine)
+    return sorted_unique(xp, ids)
+
+
 def index_seeded_hop(
     g: Plottable,
     registry: GfqlIndexRegistry,
@@ -269,6 +351,8 @@ def index_seeded_hop(
     direction: HopDirection,
     return_as_wave_front: bool,
     edge_match: Optional[EdgeMatch] = None,
+    source_node_match: Optional[object] = None,
+    destination_node_match: Optional[object] = None,
 ) -> Optional[Plottable]:
     if nodes is None or g._edges is None or g._nodes is None:
         return None
@@ -297,6 +381,22 @@ def index_seeded_hop(
         edge_filter = _build_edge_row_filter(edges, edge_match, engine)
         if edge_filter is None:
             return None
+
+    # Node matches resolve to allowed ids; empty is an answer, missing is a decline.
+    source_allowed: Optional[ArrayLike] = None
+    dest_allowed: Optional[ArrayLike] = None
+    if source_node_match is not None:
+        source_allowed = _node_match_allowed_ids(
+            g, registry, source_node_match, node_col, engine,
+        )
+        if source_allowed is None:
+            return None
+    if destination_node_match is not None:
+        dest_allowed = _node_match_allowed_ids(
+            g, registry, destination_node_match, node_col, engine,
+        )
+        if dest_allowed is None:
+            return None
     # Guard for the candidate-row form (see _EdgeMatchRowFilter.full_mask): once cumulative
     # gathered rows cross the threshold, build the whole-column mask once and reuse it.
     gathered_rows = 0
@@ -323,6 +423,11 @@ def index_seeded_hop(
             break
         if int(frontier.shape[0]) == 0:
             break
+        # Source restriction applies to seeds and every carried-forward frontier.
+        if source_allowed is not None:
+            frontier = _sorted_intersection(frontier, source_allowed, xp)
+            if int(frontier.shape[0]) == 0:
+                break
         hop_count += 1
 
         matched_parts: List[ArrayLike] = []
@@ -352,6 +457,9 @@ def index_seeded_hop(
                         # observable has been mutated, so this stays parity-safe.
                         return None
                     rows = rows[keep]
+            if dest_allowed is not None:
+                # Destination matching drops both frontier ids and edge rows.
+                rows = rows[_sorted_id_mask(ix.other_values[rows], dest_allowed, xp)]
             edge_rows_parts.append(rows)
             neigh_parts.append(ix.other_values[rows])
             matched_parts.append(matched)
diff --git a/graphistry/compute/gfql/index/types.py b/graphistry/compute/gfql/index/types.py
index e59854a7d6..3f485b3910 100644
--- a/graphistry/compute/gfql/index/types.py
+++ b/graphistry/compute/gfql/index/types.py
@@ -12,7 +12,10 @@
 if TYPE_CHECKING:
     from graphistry.compute.predicates.ASTPredicate import ASTPredicate
 
-IndexKind = Literal["edge_out_adj", "edge_in_adj", "node_id", "node_prop", "edge_prop"]
+IndexKind = Literal[
+    "edge_out_adj", "edge_in_adj", "node_id", "node_prop", "edge_prop",
+    "node_type_segment",
+]
 #: Build-time facts that `show_indexes` reports but that are not registry `indexes`.
 SidecarFactKind = Literal["category", "endpoint_rows", "temporal_text"]
 AdjacencyIndexKind = Literal["edge_out_adj", "edge_in_adj"]
@@ -94,6 +97,7 @@ class IndexTraceStep(TypedDict, total=False):
     seed_deg_sum: Optional[int]
     est_result_rows: Optional[int]
     threshold_frac: float
+    node_segments: Dict[str, str]
     # column-stat fact consults
     role: str
     column: str
diff --git a/graphistry/compute/gfql/index/wire.py b/graphistry/compute/gfql/index/wire.py
index fce0beab35..89fccc8821 100644
--- a/graphistry/compute/gfql/index/wire.py
+++ b/graphistry/compute/gfql/index/wire.py
@@ -19,7 +19,7 @@
 
 from typing_extensions import TypeGuard
 
-from .registry import ALL_KINDS
+from .registry import ALL_KINDS, NODE_TYPE_SEGMENT
 from .errors import is_not_yet_implemented_kind, not_implemented_kind_error
 from .types import IndexKind
 
@@ -157,7 +157,12 @@ def apply_index_op(g: Any, op: IndexOp, *, engine: Any = "auto") -> Any:
             return drop_index(g, kind, column=column)
         if kind is not None and not op.missing_ok:
             reg = get_registry(g)
-            if kind in (NODE_PROP, EDGE_PROP):
+            if kind == NODE_TYPE_SEGMENT:
+                is_resident = (
+                    any(key[0] == op.column for key in reg.node_type_segments)
+                    if op.column is not None else bool(reg.node_type_segments)
+                )
+            elif kind in (NODE_PROP, EDGE_PROP):
                 props = reg.property_indexes("nodes" if kind == NODE_PROP else "edges")
                 is_resident = op.column in props if op.column is not None else bool(props)
             else:
diff --git a/graphistry/tests/compute/gfql/index/test_node_type_segments.py b/graphistry/tests/compute/gfql/index/test_node_type_segments.py
new file mode 100644
index 0000000000..a4d45446ce
--- /dev/null
+++ b/graphistry/tests/compute/gfql/index/test_node_type_segments.py
@@ -0,0 +1,318 @@
+"""Node type-segment index must preserve canonical GFQL results (edge typed-index dual)."""
+from __future__ import annotations
+
+import pandas as pd
+import pytest
+
+import graphistry
+from graphistry.Engine import Engine
+from graphistry.compute.ast import e_forward, n, rows
+from graphistry.compute.gfql.index import (
+    get_registry, index_trace, maybe_index_hop, with_index_policy,
+)
+from graphistry.compute.gfql.index.cypher_ddl import parse_index_ddl
+from graphistry.compute.gfql.index.wire import apply_index_op
+
+
+def _graph(engine: str):
+    # Node 1 has TWO labels (membership); node 5 has a null type; node 4 is
+    # label-only-False; a null type never matches any scalar segment.
+    nodes = pd.DataFrame({
+        "id": [1, 2, 3, 4, 5],
+        "type": ["P", "Q", "P", "Q", None],
+        "label__Person": [True, True, False, False, True],
+        "label__Robot": [False, True, False, True, False],
+    })
+    edges = pd.DataFrame({
+        "src": [1, 1, 2, 3, 4, 5],
+        "dst": [2, 3, 4, 4, 5, 1],
+        "etype": ["K", "X", "K", "T", "K", "T"],
+    })
+    if engine == "polars":
+        pl = pytest.importorskip("polars")
+        nodes, edges = pl.from_pandas(nodes), pl.from_pandas(edges)
+    elif engine == "cudf":
+        cudf = pytest.importorskip("cudf")
+        nodes, edges = cudf.from_pandas(nodes), cudf.from_pandas(edges)
+    return graphistry.bind(node="id", source="src", destination="dst").nodes(nodes).edges(edges)
+
+
+def _pandas(frame):
+    return frame.to_pandas() if hasattr(frame, "to_pandas") else frame
+
+
+def _assert_same_graph(actual, expected) -> None:
+    if len(_pandas(actual._nodes)) == 0 and len(_pandas(expected._nodes)) == 0:
+        # Pre-existing master gap (also on the typed-edge path): the wavefront
+        # index path returns a raw-schema empty frame while the scan returns the
+        # projected empty frame. Zero rows either way; exact frame equality is
+        # asserted everywhere results are nonempty.
+        assert len(_pandas(actual._edges)) == len(_pandas(expected._edges))
+        return
+    pd.testing.assert_frame_equal(_pandas(actual._nodes), _pandas(expected._nodes))
+    assert (actual._edges is None) == (expected._edges is None)
+    if actual._edges is not None:
+        pd.testing.assert_frame_equal(_pandas(actual._edges), _pandas(expected._edges))
+
+
+def _hop_sig(g):
+    nodes, edges = _pandas(g._nodes), _pandas(g._edges)
+    return (
+        sorted(nodes[g._node].tolist()),
+        sorted(map(tuple, edges[[g._source, g._destination]].itertuples(index=False, name=None))),
+    )
+
+
+def _indexed(graph, engine: str):
+    out = graph.gfql_index_all(engine=engine)
+    out = out.gfql_index_node_types(
+        type_column="type", type_values=["P", "Q"], engine=engine)
+    out = out.gfql_index_node_types(
+        type_column="label__Person", type_values=[True], engine=engine)
+    out = out.gfql_index_node_types(
+        type_column="label__Robot", type_values=[True], engine=engine)
+    return out
+
+
+@pytest.mark.parametrize("engine", ["pandas", "polars", "cudf"])
+@pytest.mark.parametrize("source_match,dest_match", [
+    ({"type": "P"}, {"type": "Q"}),                       # scalar partition both ends
+    ({"label__Person": True}, {"label__Person": True}),   # Boolean membership
+    ({"label__Person": True}, {"label__Robot": True}),    # cross-label
+    ({"type": "P"}, {"label__Robot": True}),              # mixed scalar/Boolean
+    ({"type": "Z"}, {"type": "P"}),                       # EMPTY source segment
+    ({"type": "P"}, {"type": "Z"}),                       # EMPTY destination segment
+])
+def test_node_segment_matches_scan_single_hop(engine: str, source_match, dest_match) -> None:
+    graph = _graph(engine)
+    query = [
+        n({**{"id": 1}, **source_match}, name="a"),
+        e_forward(name="r"),
+        n(dest_match, name="b"),
+        rows(),
+    ]
+    expected = graph.gfql(query, engine=engine, index_policy="off")
+    indexed = _indexed(graph, engine)
+    actual = indexed.gfql(query, engine=engine, index_policy="force")
+    _assert_same_graph(actual, expected)
+
+
+DIRECT_HOP_CASES = [
+    ({"type": "P"}, {"type": "Q"}),
+    ({"label__Person": True}, {"label__Person": True}),
+    ({"type": "P"}, {"label__Robot": True}),
+    ({"type": "Z"}, {"type": "P"}),
+]
+
+
+@pytest.mark.parametrize("engine", ["pandas", "polars", "cudf"])
+@pytest.mark.parametrize("direction", ["forward", "reverse", "undirected"])
+@pytest.mark.parametrize("source_match,dest_match", DIRECT_HOP_CASES)
+def test_node_segment_direct_hop_matches_scan(
+    engine: str, direction: str, source_match, dest_match,
+) -> None:
+    """Direct hop parity: gfql may delegate small chains to connected-bindings."""
+    graph = _graph(engine)
+    hop_kw = dict(
+        hops=1, direction=direction, return_as_wave_front=True,
+        source_node_match=source_match, destination_node_match=dest_match,
+    )
+    expected = graph.hop(nodes=graph._nodes, engine=engine, **hop_kw)
+    indexed = _indexed(graph, engine)
+    indexed = indexed.gfql_index_node_types(
+        type_column="type", type_values=["Z"], engine=engine)
+    indexed = with_index_policy(indexed, "force")
+    actual = indexed.hop(nodes=graph._nodes, engine=engine, **hop_kw)
+    assert actual is not None
+    assert _hop_sig(actual) == _hop_sig(expected)
+
+
+@pytest.mark.route_engaged("index-hop")
+@pytest.mark.parametrize("engine", ["pandas", "polars", "cudf"])
+@pytest.mark.parametrize("direction", ["forward", "reverse", "undirected"])
+@pytest.mark.parametrize("source_match,dest_match", DIRECT_HOP_CASES)
+def test_node_segment_direct_hop_engagement(
+    engine: str, direction: str, source_match, dest_match,
+) -> None:
+    """Engagement pin for direct hops, replayed separately by the routes-off lane."""
+    graph = _graph(engine)
+    hop_kw = dict(
+        hops=1, direction=direction, return_as_wave_front=True,
+        source_node_match=source_match, destination_node_match=dest_match,
+    )
+    indexed = _indexed(graph, engine)
+    indexed = indexed.gfql_index_node_types(
+        type_column="type", type_values=["Z"], engine=engine)
+    indexed = with_index_policy(indexed, "force")
+    with index_trace() as steps:
+        actual = indexed.hop(nodes=graph._nodes, engine=engine, **hop_kw)
+    assert actual is not None
+    assert any(step["path"] == "index" for step in steps)
+    assert any(step.get("node_segments") == {
+        "source": "resident", "destination": "resident"
+    } for step in steps)
+
+
+@pytest.mark.parametrize("engine", ["pandas", "polars", "cudf"])
+def test_node_segment_two_hop_chain_matches_scan(engine: str) -> None:
+    """The IC4-realistic shape: labeled nodes on every pattern of a two-hop chain."""
+    graph = _graph(engine)
+    query = [
+        n({"id": 1, "label__Person": True}, name="a"),
+        e_forward(name="r1"), n({"type": "Q"}, name="b"),
+        e_forward(name="r2"), n({"label__Person": True}, name="c"), rows(),
+    ]
+    expected = graph.gfql(query, engine=engine, index_policy="off")
+    indexed = _indexed(graph, engine)
+    actual = indexed.gfql(query, engine=engine, index_policy="force")
+    _assert_same_graph(actual, expected)
+
+
+@pytest.mark.parametrize("engine", ["pandas", "polars", "cudf"])
+def test_node_segment_plus_typed_edge_combined(engine: str) -> None:
+    """Labeled nodes AND typed edges in the same two-hop query."""
+    graph = _graph(engine)
+    query = [
+        n({"id": 1, "type": "P"}, name="a"),
+        e_forward({"etype": "K"}, name="r1"), n({"label__Person": True}, name="b"),
+        e_forward(name="r2"), n({"type": "Q"}, name="c"), rows(),
+    ]
+    expected = graph.gfql(query, engine=engine, index_policy="off")
+    indexed = graph.gfql_index_all(engine=engine)
+    indexed = indexed.gfql_index_edges(type_column="etype", type_values=["K", "T"], engine=engine)
+    indexed = indexed.gfql_index_node_types(type_column="type", type_values=["P", "Q"], engine=engine)
+    indexed = indexed.gfql_index_node_types(type_column="label__Person", type_values=[True], engine=engine)
+    actual = indexed.gfql(query, engine=engine, index_policy="force")
+    _assert_same_graph(actual, expected)
+
+
+@pytest.mark.parametrize("engine", ["pandas", "polars"])
+def test_node_segment_null_type_never_matches_scalar_segment(engine: str) -> None:
+    graph = _graph(engine)
+    # Destination segment {type: 'P'} must exclude node 5 (null type) even though
+    # label__Person=True would admit it under a different column.
+    query = [
+        n({"id": 1}, name="a"), e_forward(name="r"), n({"type": "P"}, name="b"), rows(),
+    ]
+    expected = graph.gfql(query, engine=engine, index_policy="off")
+    indexed = _indexed(graph, engine)
+    actual = indexed.gfql(query, engine=engine, index_policy="force")
+    _assert_same_graph(actual, expected)
+
+
+@pytest.mark.parametrize("engine", ["pandas", "polars"])
+def test_stale_node_segment_declines_to_scan(engine: str) -> None:
+    graph = _graph(engine)
+    indexed = _indexed(graph, engine)
+    # Rebind nodes -> segment identity guard misses -> scan fallback, same answer.
+    pl_or_pd_nodes = indexed._nodes
+    rebound = indexed.nodes(_pandas(pl_or_pd_nodes), "id") if not hasattr(pl_or_pd_nodes, "clone") else indexed.nodes(pl_or_pd_nodes.clone(), "id")
+    query = [
+        n({"id": 1, "type": "P"}, name="a"), e_forward(name="r"), n({"type": "Q"}, name="b"), rows(),
+    ]
+    expected = graph.gfql(query, engine=engine, index_policy="off")
+    actual = rebound.gfql(query, engine=engine, index_policy="force")
+    _assert_same_graph(actual, expected)
+
+
+def test_multi_key_node_match_still_parity_via_scan() -> None:
+    """v1 covers single-key matches; multi-key declines to scan (same answer)."""
+    engine = "pandas"
+    graph = _graph(engine)
+    query = [
+        n({"id": 1, "type": "P"}, name="a"), e_forward(name="r"),
+        n({"type": "Q", "label__Robot": True}, name="b"), rows(),
+    ]
+    expected = graph.gfql(query, engine=engine, index_policy="off")
+    indexed = _indexed(graph, engine)
+    actual = indexed.gfql(query, engine=engine, index_policy="force")
+    _assert_same_graph(actual, expected)
+
+
+@pytest.mark.parametrize("engine", ["pandas", "polars"])
+def test_node_segment_lifecycle_show_and_selective_drop(engine: str) -> None:
+    graph = _graph(engine)
+    indexed = graph.gfql_index_all(engine=engine)
+    indexed = indexed.create_index("node_prop", column="id", engine=engine)
+    indexed = indexed.gfql_index_node_types(
+        type_column="type", type_values=["P", "Q"], engine=engine)
+    indexed = indexed.gfql_index_node_types(
+        type_column="label__Person", type_values=[True], engine=engine)
+    si = indexed.show_indexes()
+    typed_row = si[si["name"] == "typed_node:type=P"].iloc[0]
+    assert typed_row["kind"] == "node_type_segment"
+    assert bool(typed_row["valid"]) and bool(typed_row["usable"])
+    assert int(typed_row["n_rows"]) == 2
+
+    one = indexed.drop_index("node_type_segment", column="type", value="P")
+    registry = get_registry(one)
+    assert ("type", "P") not in registry.node_type_segments
+    assert ("type", "Q") in registry.node_type_segments
+    assert ("label__Person", True) in registry.node_type_segments
+    assert "id" in registry.node_props
+
+    column = one.drop_index("node_type_segment", column="type")
+    registry = get_registry(column)
+    assert not any(key[0] == "type" for key in registry.node_type_segments)
+    assert ("label__Person", True) in registry.node_type_segments
+    assert "id" in registry.node_props
+
+    ddl = parse_index_ddl('DROP GFQL INDEX FOR node_type_segment ON type')
+    assert ddl is not None and ddl.kind == "node_type_segment" and ddl.column == "type"
+    ddl_drop = apply_index_op(one, ddl)
+    assert not any(key[0] == "type" for key in get_registry(ddl_drop).node_type_segments)
+
+    none = column.drop_index("node_type_segment")
+    registry = get_registry(none)
+    assert registry.node_type_segments == {}
+    assert "id" in registry.node_props
+
+
+def test_gfql_index_node_types_validation() -> None:
+    graph = _graph("pandas")
+    with pytest.raises(ValueError):
+        graph.gfql_index_node_types(type_column="type")
+    with pytest.raises(ValueError):
+        graph.gfql_index_node_types(type_values=["P"])
+    with pytest.raises(TypeError):
+        graph.gfql_index_node_types(type_column="type", type_values="P")
+    with pytest.raises(ValueError):
+        graph.gfql_index_node_types(type_column="type", type_values=[])
+    with pytest.raises(TypeError):
+        graph.gfql_index_node_types(type_column="type", type_values=[1.5])
+    with pytest.raises(ValueError):
+        graph.gfql_index_node_types(type_column="nope", type_values=["P"])
+    with pytest.raises(ValueError):
+        graphistry.nodes(None).gfql_index_node_types(type_column="type", type_values=["P"])
+
+
+@pytest.mark.parametrize("engine", ["pandas", "polars"])
+def test_node_segment_trace_reports_residency(engine: str) -> None:
+    """Trace the planner entry directly: GFQL may dispatch the same chain to a
+    higher-level indexed seam, which does not consume node segments yet."""
+    indexed = _indexed(_graph(engine), engine)
+    engine_enum = Engine.PANDAS if engine == "pandas" else Engine.POLARS
+    seeds = indexed._nodes.head(1)
+
+    with index_trace() as steps:
+        served = maybe_index_hop(
+            indexed, engine_enum, nodes=seeds, hops=1, direction="forward",
+            return_as_wave_front=True, policy="force",
+            source_node_match={"type": "P"}, destination_node_match={"type": "Q"},
+        )
+    assert served is not None
+    assert steps[-1]["node_segments"] == {"source": "resident", "destination": "resident"}
+    assert steps[-1]["path"] == "index"
+
+    bare = _graph(engine).gfql_index_all(engine=engine)
+    seeds = bare._nodes.head(1)
+    with index_trace() as steps:
+        declined = maybe_index_hop(
+            bare, engine_enum, nodes=seeds, hops=1, direction="forward",
+            return_as_wave_front=True, policy="force",
+            source_node_match={"type": "P"}, destination_node_match={"type": "Q"},
+        )
+    assert declined is None
+    assert steps[-1]["node_segments"] == {
+        "source": "missing_or_stale", "destination": "missing_or_stale"}
+    assert steps[-1]["path"] == "scan"