Merge pull request #1777 from graphistry/perf/gfql-node-property-index

perf(gfql): secondary node property indexes for seeded traversal

Author: lmeyerov

CHANGELOG.md

@@ -8,6 +8,9 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 ## [Development]
 <!-- Do Not Erase This Section - Used for tracking unreleased changes -->
 
+### Added
+- **GFQL secondary node property indexes (`create_index('node_prop', column=...)` / `g.gfql_index_node_props([...])`)**: a seed predicate on a NON-key column — `MATCH (m {id: 42})` where the graph's node id binding is some other column — previously cost a full node scan, because the registry only indexed the node-id binding and the CSR adjacencies. A property index is the same pay-as-you-go sidecar as the existing kinds: sorted distinct values over node **row positions** (CSR, so duplicate values are indexable), never reorders `.nodes`, fingerprint-validated so a `.nodes()` rebind is treated as absent (safe miss, never a wrong answer), engine-polymorphic (numpy host / cupy on-device), and policy-gated (`off`/`use`/`auto`/`force`). The seeded fixed-hop planner picks the **most selective** indexed scalar predicate in the seed filter using a free CSR-offset estimate, gathers those candidates, and applies every remaining predicate to them — so results are identical whether the index is present, absent, stale, or cost-gated out. `show_indexes()` lists property indexes; `drop_index('node_prop', column=...)` drops one. Only integer columns are indexable today (float NaN ordering, strings on cupy, and nulls all decline to the scan); widening that is additive. **Perf (dgx-spark, official LDBC SNB SF1, 3.18M nodes, warm median, value-identical 19-row result):** interactive-short IS7 `71.6 ms -> 19.5 ms` (**3.7x**), with a one-time `112 ms` build — the seed lookup itself goes from a `51.2 ms` scan to `0.096 ms`.
+
 ### Performance
 - **GFQL indexed fixed-hop bindings now dispatch BEFORE the canonical traversal (pandas / cuDF / native Polars)**: the resident-index fixed-hop path (`rows(binding_ops=...)`, the Cypher multi-alias `MATCH … RETURN` shape) was consulted only *inside* row materialization — after the engine had already run the full canonical traversal — so its compact path bag was computed on top of the graph-sized work it exists to avoid. Both the pandas/cuDF chain boundary and the native Polars chain now ask the same shared, structural gate first and, only when it can serve the WHOLE middle exactly, skip the canonical traversal and hand the compact state to the unchanged row materializer. Engagement stays operator/index/dtype/cost based — no query, schema, or hop-count recognition — and every unsupported, seeded, prefiltered, policy-bearing, shortest-path, or cost-gated shape still declines to canonical execution with identical results. On a standard LDBC SNB SF1 interactive-short profile this removed the redundant two-hop typed-mask and 16-merge buckets and cut the profiled call ~11.9× (1263 ms → 106 ms), exact 19-row oracle preserved.
 - **Indexed seed lookup now covers a unique node id PLUS extra scalar constraints**: `{node_id: v, other: w}` previously fell back to a full node-table scan even though the unique node-id index can gather at most one row; it now gathers that one row and applies the remaining constraints to it. Missing ids and non-matching extra constraints return the same empty result as before, and duplicate-id graphs (which cannot build the index) are unaffected.

docs/source/gfql/indexing.rst

@@ -39,6 +39,12 @@ carrying them:
      - Sorted node-id lookup: seed-row and endpoint materialization become positional
        gathers instead of ``O(N)`` scans. Requires unique node ids —
        ``gfql_index_all()`` silently skips it otherwise (adjacency is still built).
+   * - ``node_prop``
+     - Sorted lookup on a node **property** column (a secondary index): a seed
+       predicate like ``MATCH (m {id: 42})`` on a column that is not the node-id
+       binding becomes a positional gather instead of an ``O(N)`` scan. Duplicate
+       values are fine (all matching rows are gathered). Integer columns without
+       nulls only — anything else declines to the scan. Opt-in per column.
 
 They are **sidecars over row positions**: your ``.edges`` / ``.nodes`` frames are never
 reordered or copied, and the resident footprint is visible per index via
@@ -93,12 +99,36 @@ API):
    g = g.gfql_index_all()               # out+in adjacency + node_id (the one-liner)
    g = g.gfql_index_edges("forward")    # or just one direction: 'forward'|'reverse'|'both'
    g = g.create_index("edge_out_adj")   # or one kind: 'edge_out_adj'|'edge_in_adj'|'node_id'
+   g = g.gfql_index_node_props(["id"])  # secondary indexes on node property columns
    g.show_indexes()                     # pandas DataFrame: kind, engine, n_keys, nbytes, valid
    g = g.drop_index()                   # drop all (or drop_index("edge_out_adj"))
 
 Unlike ``gfql_index_all()``, an explicit ``create_index("node_id")`` **raises** on
 non-unique node ids rather than skipping.
 
+Seeding on a property (secondary index)
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+The ``node_id`` index covers the column bound as the node id. A query that seeds on
+a *different* column — a business key such as ``MATCH (m {id: 42})`` when the graph
+is keyed by something else — otherwise scans the whole node table to find its seed.
+``node_prop`` indexes that column instead:
+
+.. code-block:: python
+
+   g = g.gfql_index_all().gfql_index_node_props(["id"])   # skips unindexable columns
+   g = g.create_index("node_prop", column="id")           # or one column, raising if it cannot
+
+   # equivalently over the Cypher DDL / JSON surfaces
+   g.gfql('CREATE GFQL INDEX FOR node_prop ON id')
+   g = g.drop_index("node_prop", column="id")             # or drop_index("node_prop") for all
+
+When several indexed columns appear in one seed predicate, the planner gathers on the
+**most selective** one (estimated for free from the index's own offsets) and applies
+the remaining predicates to those candidates, so results never depend on which index
+happens to be resident. As with every kind, a missing, stale, or cost-gated-out index
+simply falls back to the scan.
+
 What uses the index today
 -------------------------
 

graphistry/compute/ComputeMixin.py

@@ -1,6 +1,6 @@
 import numpy as np
 import pandas as pd
-from typing import TYPE_CHECKING, Any, Dict, List, Optional, Union
+from typing import TYPE_CHECKING, Any, Dict, List, Optional, Sequence, Union
 from typing_extensions import Literal
 from graphistry.Engine import Engine, EngineAbstract, EngineAbstractType, POLARS_ENGINES, resolve_engine, df_to_engine, df_concat, safe_merge
 from graphistry.Plottable import Plottable
@@ -29,6 +29,7 @@
 )
 
 if TYPE_CHECKING:
+    from graphistry.compute.gfql.index.types import IndexKind
     from graphistry.compute.gfql.index.explain import GfqlExplainReport
     from graphistry.compute.gfql.index.policy import IndexPolicy
     from graphistry.compute.gfql.query_types import GFQLQuery
@@ -638,10 +639,10 @@ 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=None):
-        """Drop one resident GFQL index (by kind) or all (kind=None). Idempotent; returns a new Plottable."""
+    def drop_index(self, kind: Optional['IndexKind'] = None, *, column: Optional[str] = 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)
+        return _di(self, kind, column=column)
 
     def show_indexes(self):
         """Return a pandas DataFrame describing resident GFQL indexes (name, kind, column, valid). Empty if none; ``valid=False`` marks a stale index after a frame rebind."""
@@ -658,6 +659,15 @@ def gfql_index_all(self, engine='auto'):
         from graphistry.compute.gfql.index import gfql_index_all as _gia
         return _gia(self, engine=engine)
 
+    def gfql_index_node_props(self, columns: Sequence[str], engine: EngineAbstractType = 'auto') -> 'Plottable':
+        """Convenience: build node PROPERTY indexes for ``columns`` (secondary indexes).
+
+        A seed predicate on a non-key column (``{id: 42}`` when the graph's node id
+        is some other column) otherwise costs a full node scan. Unindexable columns
+        are skipped, keeping the correct scan path. Returns a new Plottable."""
+        from graphistry.compute.gfql.index import gfql_index_node_props as _ginp
+        return _ginp(self, columns, engine=engine)
+
     def filter_nodes_by_dict(self, *args, **kwargs):
         return filter_nodes_by_dict_base(self, *args, **kwargs)
     filter_nodes_by_dict.__doc__ = filter_nodes_by_dict_base.__doc__

graphistry/compute/dataframe/join.py

@@ -294,3 +294,124 @@ def _uniq(df: DataFrameT, col: str, name: str) -> DataFrameT:
     if right_keep:
         right_eval = right_eval[list(right_keep)]
     return left_eval, right_eval, mid_values
+
+
+def estimate_inner_join_rows(
+    left: DataFrameT,
+    right: DataFrameT,
+    *,
+    left_on: str,
+    right_on: str,
+    engine: Engine,
+) -> int:
+    """Rows an inner join WOULD emit, without materializing it.
+
+    Sum over matched keys of (left count * right count) — a group-by on each side
+    instead of the join itself, so a planner can cost-gate a hop before paying for
+    it.
+    """
+    if len(left) == 0 or len(right) == 0:
+        return 0
+    left_n, right_n = "__gfql_join_left_n__", "__gfql_join_right_n__"
+    if engine in POLARS_ENGINES:
+        import polars as pl
+
+        left_counts = left.group_by(left_on).len().rename({"len": left_n})  # type: ignore[operator]
+        right_counts = right.group_by(right_on).len().rename({"len": right_n})  # type: ignore[operator]
+        value = (
+            left_counts.join(right_counts, left_on=left_on, right_on=right_on, how="inner")
+            .select((pl.col(left_n) * pl.col(right_n)).sum())
+            .item()
+        )
+        return 0 if value is None else int(value)
+
+    left_counts = left.groupby(left_on, sort=False).size().reset_index()
+    left_counts.columns = [left_on, left_n]
+    right_counts = right.groupby(right_on, sort=False).size().reset_index()
+    right_counts.columns = [right_on, right_n]
+    counts = left_counts.merge(
+        right_counts, left_on=left_on, right_on=right_on, how="inner", sort=False,
+    )
+    if len(counts) == 0:
+        return 0
+    return int((counts[left_n] * counts[right_n]).sum())
+
+
+def path_ordered_expand_join(
+    state: DataFrameT,
+    step: DataFrameT,
+    *,
+    current_col: str,
+    from_col: str,
+    to_col: str,
+    path_order_col: str,
+    tiebreak_cols: Sequence[str],
+    alias: Optional[str],
+    engine: Engine,
+) -> DataFrameT:
+    """Expand a path-bag by one step, preserving (path, tiebreak...) row order.
+
+    ``state[current_col]`` joins ``step[from_col]``; ``step[to_col]`` becomes the new
+    ``current_col``. A synthetic ``path_order_col`` pins the incoming row order so the
+    result is deterministic under any engine's join, and the bookkeeping columns are
+    dropped on the way out. ``alias``, when given, also exposes the new current value
+    under that name.
+    """
+    drop_after = [from_col, path_order_col, *tiebreak_cols]
+    if engine in POLARS_ENGINES:
+        import polars as pl
+
+        joined = (
+            state.with_row_index(path_order_col)  # type: ignore[operator]
+            .join(step, left_on=current_col, right_on=from_col, how="inner")
+            .sort([path_order_col, *tiebreak_cols])
+            .drop(current_col)
+            .rename({to_col: current_col})
+        )
+        if isinstance(alias, str):
+            joined = joined.with_columns(pl.col(current_col).alias(alias))
+        return cast(
+            DataFrameT,
+            joined.drop([col for col in drop_after if col in joined.columns]),
+        )
+
+    import numpy as np
+
+    out = state.assign(**{path_order_col: np.arange(len(state))}).merge(
+        step, left_on=current_col, right_on=from_col, how="inner", sort=False,
+    )
+    if len(out):
+        sort_cols = [path_order_col, *tiebreak_cols]
+        out = (
+            out.sort_values(sort_cols, kind="stable") if engine == Engine.PANDAS
+            else out.sort_values(sort_cols)
+        )
+    out = out.drop(columns=[current_col]).rename(columns={to_col: current_col})
+    if isinstance(alias, str):
+        out = out.assign(**{alias: out[current_col]})
+    return cast(
+        DataFrameT,
+        out.drop(columns=[col for col in drop_after if col in out.columns]),
+    )
+
+
+def semijoin_by_column(
+    frame: DataFrameT,
+    keys: DataFrameT,
+    *,
+    left_on: str,
+    right_on: str,
+    engine: Engine,
+) -> DataFrameT:
+    """Rows of ``frame`` whose ``left_on`` value appears in ``keys[right_on]``."""
+    if engine in POLARS_ENGINES:
+        return cast(
+            DataFrameT,
+            frame.join(  # type: ignore[call-arg]
+                keys.select(right_on).unique(),  # type: ignore[operator]
+                left_on=left_on,
+                right_on=right_on,
+                how="semi",  # type: ignore[arg-type]
+            ),
+        )
+    return cast(DataFrameT, frame[frame[left_on].isin(keys[right_on])])

graphistry/compute/gfql/index/__init__.py

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