diff --git a/CHANGELOG.md b/CHANGELOG.md index a0da6619bd..ef993819b4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,9 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm ## [Development] +### 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. diff --git a/docs/source/gfql/indexing.rst b/docs/source/gfql/indexing.rst index 457dc28064..d63dc535db 100644 --- a/docs/source/gfql/indexing.rst +++ b/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 ------------------------- diff --git a/graphistry/compute/ComputeMixin.py b/graphistry/compute/ComputeMixin.py index f8956663e9..736d5853f8 100644 --- a/graphistry/compute/ComputeMixin.py +++ b/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__ diff --git a/graphistry/compute/dataframe/join.py b/graphistry/compute/dataframe/join.py index dc8fc97aa2..62dc13336a 100644 --- a/graphistry/compute/dataframe/join.py +++ b/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])]) diff --git a/graphistry/compute/gfql/index/__init__.py b/graphistry/compute/gfql/index/__init__.py index 1d2386923d..774dc5149a 100644 --- a/graphistry/compute/gfql/index/__init__.py +++ b/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", diff --git a/graphistry/compute/gfql/index/api.py b/graphistry/compute/gfql/index/api.py index 6d7a38915d..97ac61a632 100644 --- a/graphistry/compute/gfql/index/api.py +++ b/graphistry/compute/gfql/index/api.py @@ -8,7 +8,7 @@ from __future__ import annotations import copy -from typing import Dict, List, Literal, Optional, cast +from typing import Dict, List, Literal, Optional, Sequence, cast import pandas as pd @@ -17,9 +17,9 @@ from graphistry.Plottable import Plottable from .registry import ( AdjacencyIndex, GfqlIndexRegistry, EMPTY_REGISTRY, NodeIdIndex, - EDGE_OUT_ADJ, EDGE_IN_ADJ, NODE_ID, ADJ_KINDS, ALL_KINDS, + EDGE_OUT_ADJ, EDGE_IN_ADJ, NODE_ID, NODE_PROP, ADJ_KINDS, ALL_KINDS, ) -from .build import build_adjacency_index, build_node_id_index +from .build import build_adjacency_index, build_node_id_index, build_node_prop_index 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 @@ -32,6 +32,17 @@ POLICY_ATTR = "_gfql_index_policy" REGISTRY_ATTR = "_gfql_index_registry" + +class GfqlIndexUnsupportedError(ValueError): + """The DATA cannot support this index (duplicate node ids, an unindexable + property dtype). Distinct from a caller mistake — a missing column, an unknown + kind, unbound edges — which stays a plain ``ValueError`` and must propagate. + + Subclasses ``ValueError`` so existing ``except ValueError`` callers keep + working; the convenience builders catch only THIS type, so a real failure is + never silently skipped. + """ + # --- lightweight, thread-local index decision trace (for gfql_explain) ------- import threading as _threading _TRACE = _threading.local() @@ -113,6 +124,13 @@ def get_index_policy(g: Plottable) -> IndexPolicy: return g._gfql_index_policy +def with_index_policy(g: Plottable, policy: IndexPolicy) -> Plottable: + """A copy of ``g`` carrying ``policy`` (never mutates ``g``).""" + out = g.bind() + out._gfql_index_policy = policy + return out + + def _attach(g: Plottable, registry: GfqlIndexRegistry) -> Plottable: res = copy.copy(g) res._gfql_index_registry = registry @@ -208,7 +226,7 @@ def create_index( _check_column(column, node_col, kind) node_idx = build_node_id_index(g2._nodes, node_col, eng) if node_idx is None: - raise ValueError( + raise GfqlIndexUnsupportedError( f"Cannot build a {NODE_ID!r} index: node id column {node_col!r} has " f"duplicate values (a node-id index requires unique ids). Seeded " f"traversal still works via the un-indexed node materialization path." @@ -217,14 +235,43 @@ def create_index( registry = registry.with_index(NODE_ID, node_idx) return _attach(g2, registry) + if kind == NODE_PROP: + if not column: + raise ValueError( + f"A {NODE_PROP!r} index indexes one node PROPERTY column; pass " + f"column=''." + ) + g2 = g.materialize_nodes() if g._nodes is None else g + assert g2._nodes is not None + if column not in g2._nodes.columns: + raise ValueError( + f"Cannot build a {NODE_PROP!r} index: node column {column!r} not found." + ) + prop_idx = build_node_prop_index(g2._nodes, column, eng) + if prop_idx is None: + raise GfqlIndexUnsupportedError( + f"Cannot build a {NODE_PROP!r} index on {column!r}: only integer " + f"columns without nulls are indexable today. Seeded queries still " + f"work via the un-indexed scan path." + ) + prop_idx = replace(prop_idx, name=name or index_name(kind, column)) + registry = registry.with_node_prop(column, prop_idx) + return _attach(g2, registry) + raise ValueError(f"Unknown GFQL index kind: {kind!r}. Expected one of {ALL_KINDS}.") -def drop_index(g: Plottable, kind: Optional[IndexKind] = None) -> Plottable: - """Drop one index (by kind) or all indexes (kind=None). Idempotent.""" +def drop_index( + g: Plottable, kind: Optional[IndexKind] = None, *, column: Optional[str] = None +) -> Plottable: + """Drop one index (by kind, or one property index by column) or all (kind=None). + + Idempotent.""" registry = get_registry(g) if kind is None: return _attach(g, EMPTY_REGISTRY) + if kind == NODE_PROP and column is not None: + return _attach(g, registry.without_node_prop(column)) return _attach(g, registry.without(kind)) @@ -264,6 +311,19 @@ def show_indexes(g: Plottable) -> pd.DataFrame: "nbytes": index_nbytes(idx), "valid": valid, }) + for column in registry.node_prop_cols(): + prop = registry.node_props[column] + rows.append({ + "name": prop.name or index_name(NODE_PROP, column), + "kind": NODE_PROP, + "key_col": column, + "engine": prop.engine.value, + "backend": prop.backend, + "n_keys": prop.n_keys, + "n_rows": prop.n_nodes, + "nbytes": index_nbytes(prop), + "valid": registry.get_node_prop_valid(column, g._nodes, prop.engine) is not None, + }) cols = ["name", "kind", "key_col", "engine", "backend", "n_keys", "n_rows", "nbytes", "valid"] return pd.DataFrame(rows, columns=cols) @@ -278,6 +338,23 @@ def gfql_index_edges(g: Plottable, direction: EdgeIndexDirection = "both", return g +def gfql_index_node_props(g: Plottable, columns: Sequence[str], + engine: EngineAbstractType = EngineAbstract.AUTO) -> Plottable: + """Convenience: build node property indexes for ``columns`` (skips unindexable). + + Skipping mirrors ``gfql_index_all``'s node_id behaviour — a column whose dtype + this index cannot serve keeps the correct scan path. ONLY that case is skipped: + a missing column, an unknown kind, or any unexpected failure propagates. + ``create_index(NODE_PROP, column=...)`` still raises for everything, since the + caller asked for that column specifically.""" + for column in columns: + try: + g = create_index(g, NODE_PROP, column=column, engine=engine) + except GfqlIndexUnsupportedError: + continue # dtype this index cannot serve -> keep the correct scan path + return g + + def gfql_index_all(g: Plottable, engine: EngineAbstractType = EngineAbstract.AUTO) -> Plottable: """Convenience: build out+in adjacency + (when ids are unique) node_id indexes. @@ -290,7 +367,7 @@ def gfql_index_all(g: Plottable, g = gfql_index_edges(g, "both", engine=engine) try: g = create_index(g, NODE_ID, engine=engine) - except ValueError: + except GfqlIndexUnsupportedError: pass # non-unique node ids -> skip the node_id accelerator (adjacency still built) return g diff --git a/graphistry/compute/gfql/index/bindings.py b/graphistry/compute/gfql/index/bindings.py index 2dfbd72f90..0cb5487b21 100644 --- a/graphistry/compute/gfql/index/bindings.py +++ b/graphistry/compute/gfql/index/bindings.py @@ -10,9 +10,9 @@ from dataclasses import dataclass from numbers import Integral -from typing import Any, Dict, List, Mapping, Optional, Sequence, Tuple, cast +from typing import Any, Dict, List, Mapping, Optional, Sequence, Tuple, Union, cast -from graphistry.Engine import Engine +from graphistry.Engine import Engine, df_concat from graphistry.Plottable import Plottable from graphistry.compute.typing import DataFrameT @@ -21,11 +21,32 @@ _trace_active, get_index_policy, get_registry, + with_index_policy, ) +from graphistry.compute.dataframe.join import ( + estimate_inner_join_rows, + path_ordered_expand_join, + semijoin_by_column, +) + from .cost import cost_gate_frac from .engine_arrays import array_namespace, col_to_array, take_rows -from .lookup import lookup_edge_rows, lookup_node_rows -from .registry import EDGE_IN_ADJ, EDGE_OUT_ADJ, NODE_ID, AdjacencyIndex, NodeIdIndex +from .lookup import ( + lookup_degree, + lookup_edge_rows, + lookup_node_rows, + lookup_prop_rows, + prop_match_count, +) +from .registry import ( + EDGE_IN_ADJ, + EDGE_OUT_ADJ, + NODE_ID, + AdjacencyIndex, + GfqlIndexRegistry, + NodeIdIndex, + NodePropIndex, +) from .traverse import _indices_for_direction @@ -76,13 +97,16 @@ def _simple_filter_dict(value: Any, *, allow_empty: bool = True) -> bool: ) -def _integer_index(index: Any) -> bool: - key_dtype = getattr(getattr(index, "keys_sorted", None), "dtype", None) - other_dtype = getattr(getattr(index, "other_values", None), "dtype", None) - key_ok = getattr(key_dtype, "kind", None) in ("i", "u") - if isinstance(index, NodeIdIndex): +def _integer_index(index: Union[AdjacencyIndex, NodeIdIndex, NodePropIndex]) -> bool: + """Whether an index's keys (and, for adjacency, its neighbor ids) are integral. + + The vectorized gather promotes dtypes rather than narrowing, so non-integral + keys are declined rather than risking a lossy compare. + """ + key_ok = index.keys_sorted.dtype.kind in ("i", "u") + if not isinstance(index, AdjacencyIndex): return key_ok - return key_ok and getattr(other_dtype, "kind", None) in ("i", "u") + return key_ok and index.other_values.dtype.kind in ("i", "u") def _filter_compatible(frame: DataFrameT, filter_dict: Optional[dict]) -> bool: @@ -137,23 +161,7 @@ def _with_marker(frame: DataFrameT, name: Optional[str], engine: Engine) -> Data if engine == Engine.POLARS: # Native Polars intentionally omits pandas' alias-marker residue. return frame - out = frame.copy() - out[name] = True - return cast(DataFrameT, out) - - -def _concat(frames: Sequence[DataFrameT], engine: Engine) -> DataFrameT: - if engine == Engine.POLARS: - import polars as pl - - return cast(DataFrameT, pl.concat(list(frames), how="vertical")) # type: ignore[type-var] - if engine == Engine.CUDF: - import cudf # type: ignore - - return cast(DataFrameT, cudf.concat(list(frames), ignore_index=True)) - import pandas as pd - - return cast(DataFrameT, pd.concat(list(frames), ignore_index=True)) + return cast(DataFrameT, frame.assign(**{name: True})) def _frame_with_positions( @@ -167,9 +175,7 @@ def _frame_with_positions( DataFrameT, frame.with_columns(pl.Series(_EDGE_ORD, np.asarray(positions))), # type: ignore[operator] ) - out = frame.copy() - out[_EDGE_ORD] = positions - return cast(DataFrameT, out) + return cast(DataFrameT, frame.assign(**{_EDGE_ORD: positions})) def _orient_edges( @@ -205,9 +211,9 @@ def one(from_col: str, to_col: str, orient: int) -> DataFrameT: ) else: - work = gathered.copy() + work = gathered if isinstance(alias, str): - work[alias] = True + work = gathered.assign(**{alias: True}) payload = payload + [alias] renames = {col: f"{alias}.{col}" for col in payload} else: @@ -217,151 +223,72 @@ def one(from_col: str, to_col: str, orient: int) -> DataFrameT: out = work.rename(columns={from_col: _FROM, to_col: _TO}) if renames: out = out.rename(columns=renames) - out[_ORIENT_ORD] = orient - return cast(DataFrameT, out) + return cast(DataFrameT, out.assign(**{_ORIENT_ORD: orient})) if direction == "undirected": forward = one(src, dst, 0) reverse = one(dst, src, 1) if engine == Engine.POLARS: reverse = reverse.select(forward.columns) # type: ignore[operator] - return _concat([forward, reverse], engine) + return cast(DataFrameT, df_concat(engine)([forward, reverse], ignore_index=True)) if direction == "reverse": return one(dst, src, 0) return one(src, dst, 0) -def _estimate_join_rows( - state: DataFrameT, oriented: DataFrameT, engine: Engine, -) -> int: - if len(state) == 0 or len(oriented) == 0: - return 0 - if engine == Engine.POLARS: - import polars as pl - - left = state.group_by(_CURRENT).len().rename({"len": _LEFT_N}) # type: ignore[operator] - right = oriented.group_by(_FROM).len().rename({"len": _RIGHT_N}) # type: ignore[operator] - value = ( - left.join(right, left_on=_CURRENT, right_on=_FROM, how="inner") - .select((pl.col(_LEFT_N) * pl.col(_RIGHT_N)).sum()) - .item() - ) - return 0 if value is None else int(value) - - left = state.groupby(_CURRENT, sort=False).size().reset_index() - left.columns = [_CURRENT, _LEFT_N] - right = oriented.groupby(_FROM, sort=False).size().reset_index() - right.columns = [_FROM, _RIGHT_N] - counts = left.merge( - right, left_on=_CURRENT, right_on=_FROM, how="inner", sort=False, - ) - if len(counts) == 0: - return 0 - return int((counts[_LEFT_N] * counts[_RIGHT_N]).sum()) - - -def _join_state( - state: DataFrameT, - oriented: DataFrameT, - *, - node_alias: Optional[str], - engine: Engine, -) -> DataFrameT: - if engine == Engine.POLARS: - import polars as pl - - joined = ( - state.with_row_index(_PATH_ORD) # type: ignore[operator] - .join(oriented, left_on=_CURRENT, right_on=_FROM, how="inner") - .sort([_PATH_ORD, _ORIENT_ORD, _EDGE_ORD]) - .drop(_CURRENT) - .rename({_TO: _CURRENT}) - ) - if isinstance(node_alias, str): - joined = joined.with_columns(pl.col(_CURRENT).alias(node_alias)) - return cast( - DataFrameT, - joined.drop([ - col for col in (_FROM, _PATH_ORD, _EDGE_ORD, _ORIENT_ORD) - if col in joined.columns - ]), - ) - - out = state.copy() - xp, _ = array_namespace(engine) - out[_PATH_ORD] = xp.arange(len(out)) # type: ignore[call-overload] - out = out.merge( - oriented, left_on=_CURRENT, right_on=_FROM, how="inner", sort=False, - ) - if len(out): - if engine == Engine.PANDAS: - out = out.sort_values( - [_PATH_ORD, _ORIENT_ORD, _EDGE_ORD], kind="stable", - ) - else: - out = out.sort_values([_PATH_ORD, _ORIENT_ORD, _EDGE_ORD]) - out = out.drop(columns=[_CURRENT]).rename(columns={_TO: _CURRENT}) - if isinstance(node_alias, str): - out[node_alias] = out[_CURRENT] - return cast( - DataFrameT, - out.drop( - columns=[ - col for col in (_FROM, _PATH_ORD, _EDGE_ORD, _ORIENT_ORD) - if col in out.columns - ], - ), - ) - - def _policy_is_active() -> bool: from graphistry.compute.gfql.call.executor import _thread_local return getattr(_thread_local, "policy", None) is not None -def _filter_oriented_endpoints( - oriented: DataFrameT, - next_nodes: DataFrameT, - node_id: str, +def _seed_rows_via_property_index( + registry: GfqlIndexRegistry, + nodes: DataFrameT, + first_filter: Mapping[str, Any], engine: Engine, -) -> DataFrameT: - if engine == Engine.POLARS: - return cast( - DataFrameT, - oriented.join( # type: ignore[call-arg] - next_nodes.select(node_id).unique(), # type: ignore[operator] - left_on=_TO, - right_on=node_id, - how="semi", # type: ignore[arg-type] - ), - ) - return cast( - DataFrameT, - oriented[oriented[_TO].isin(next_nodes[node_id])].copy(), - ) - - + xp: Any, + *, + policy: str, +) -> Optional[Any]: + """Node row positions for the most selective indexed scalar seed predicate. + + The seed of a fixed-hop pattern is usually a high-selectivity equality on a + business key (``{id: 42}``) that is NOT the graph's node-id binding, which + otherwise costs a full node scan. When a resident, still-valid property index + covers such a column, gather its candidates instead; the caller re-applies the + WHOLE filter to them, so the result is identical to the scan either way. + + Returns None (keep scanning) when nothing is indexed, no predicate is a plain + integer scalar, or the estimated candidate count is not selective enough to + beat the scan (``force`` skips the cost gate). + """ + if not first_filter: + return None + best_rows = None + best_count: Optional[int] = None + for column in registry.node_prop_cols(): + value = first_filter.get(column) + if value is None or isinstance(value, bool) or not isinstance(value, Integral): + continue + index = registry.get_node_prop_valid(column, nodes, engine) + if index is None: + continue + values = xp.asarray([value]) + count = prop_match_count(index, values, xp) + if best_count is not None and count >= best_count: + continue + best_count = count + best_rows = (index, values) + if best_rows is None or best_count is None: + return None + if policy != "force": + n_nodes = int(nodes.shape[0]) + if best_count >= cost_gate_frac(engine) * n_nodes: + return None # not selective enough to beat one vectorized scan + index, values = best_rows + return xp.sort(lookup_prop_rows(index, values, xp)) -def _lookup_degree(index: AdjacencyIndex, frontier: Any, xp: Any) -> int: - """Native O(frontier) degree estimate before CSR range expansion.""" - keys = index.keys_sorted - if int(keys.shape[0]) == 0 or int(frontier.shape[0]) == 0: - return 0 - values = frontier - if values.dtype != keys.dtype: - common = xp.promote_types(values.dtype, keys.dtype) - values = values.astype(common) - keys = keys.astype(common) - positions = xp.searchsorted(keys, values) - clipped = xp.where( - positions < keys.shape[0], positions, keys.shape[0] - 1, - ) - hits = clipped[keys[clipped] == values] - if int(hits.shape[0]) == 0: - return 0 - counts = index.group_offsets[hits + 1] - index.group_offsets[hits] - return int(counts.sum()) def _try_indexed_connected_bindings_state( base_graph: Plottable, @@ -395,9 +322,8 @@ def _try_indexed_connected_bindings_state( node_id, src, dst = str(node_id), str(src), str(dst) aliases = [ - getattr(op, "_name", None) - for op in ops - if isinstance(getattr(op, "_name", None), str) + op._name for op in ops + if isinstance(op, (ASTNode, ASTEdge)) and isinstance(op._name, str) ] if ( len(aliases) != len(set(aliases)) @@ -501,10 +427,20 @@ def _try_indexed_connected_bindings_state( first_nodes = take_rows(nodes, seed_rows, engine) first_nodes = _filter_frame(first_nodes, first_filter, engine) else: - hop_count = (len(ops) - 1) // 2 - if int(nodes.shape[0]) >= hop_count * int(edges.shape[0]): - return None - first_nodes = _filter_frame(nodes, first_filter, engine) + prop_rows = _seed_rows_via_property_index( + registry, nodes, first_filter, engine, xp, policy=get_index_policy(base_graph), + ) + if prop_rows is not None: + # Secondary index hit: gather the candidates, then let the UNCHANGED + # filter apply every remaining predicate to that small frame. + first_nodes = _filter_frame( + take_rows(nodes, prop_rows, engine), first_filter, engine, + ) + else: + hop_count = (len(ops) - 1) // 2 + if int(nodes.shape[0]) >= hop_count * int(edges.shape[0]): + return None + first_nodes = _filter_frame(nodes, first_filter, engine) first_alias = first_op._name alias_frames: Dict[str, DataFrameT] = {} @@ -519,9 +455,9 @@ def _try_indexed_connected_bindings_state( if isinstance(first_alias, str): state = state.with_columns(pl.col(_CURRENT).alias(first_alias)) else: - state = first_nodes[[node_id]].copy().rename(columns={node_id: _CURRENT}) + state = first_nodes[[node_id]].rename(columns={node_id: _CURRENT}) if isinstance(first_alias, str): - state[first_alias] = state[_CURRENT] + state = state.assign(**{first_alias: state[_CURRENT]}) estimated_rows = int(state.shape[0]) policy = get_index_policy(base_graph) @@ -538,7 +474,7 @@ def _try_indexed_connected_bindings_state( if int(frontier.shape[0]) >= threshold: return None gather_estimate = sum( - _lookup_degree(index, frontier, xp) for index in edge_indexes + lookup_degree(index, frontier, xp) for index in edge_indexes ) if ( policy != "force" @@ -574,17 +510,24 @@ def _try_indexed_connected_bindings_state( next_nodes = take_rows(nodes, node_rows, engine) next_nodes = _filter_frame(next_nodes, next_op.filter_dict, engine) next_alias_frame = _with_marker(next_nodes, next_op._name, engine) - oriented = _filter_oriented_endpoints( - oriented, next_nodes, node_id, engine, + oriented = semijoin_by_column( + oriented, next_nodes, left_on=_TO, right_on=node_id, engine=engine, ) - estimated_rows = _estimate_join_rows(state, oriented, engine) + estimated_rows = estimate_inner_join_rows( + state, oriented, left_on=_CURRENT, right_on=_FROM, engine=engine, + ) if policy != "force" and estimated_rows > 0 and estimated_rows >= n_edges: return None - state = _join_state( + state = path_ordered_expand_join( state, oriented, - node_alias=next_op._name, + current_col=_CURRENT, + from_col=_FROM, + to_col=_TO, + path_order_col=_PATH_ORD, + tiebreak_cols=(_ORIENT_ORD, _EDGE_ORD), + alias=next_op._name, engine=engine, ) if isinstance(next_op._name, str): @@ -606,7 +549,7 @@ def _connected_decline_reason( alias_prefilters: Optional[Any], ) -> str: """Classify a completed safe decline without changing canonical behavior.""" - from graphistry.compute.ast import ASTEdge + from graphistry.compute.ast import ASTEdge, ASTNode if engine not in (Engine.PANDAS, Engine.CUDF, Engine.POLARS): return "unsupported_engine" @@ -640,21 +583,19 @@ def _connected_decline_reason( if not _integer_index(valid): return "unsupported_dtype" - if get_index_policy(base_graph) != "force" and ops: - first_filter = getattr(ops[0], "filter_dict", None) - if isinstance(first_filter, Mapping) and first_filter: - first_nodes = _filter_frame(nodes, cast(dict, first_filter), engine) - frontier_n = int(first_nodes.shape[0]) - adjacency = [ - cast(AdjacencyIndex, valid) - for kind, frame, columns in required - for valid in [registry.get_valid(kind, frame, columns, engine)] - if kind != NODE_ID and valid is not None - ] - if adjacency and frontier_n >= cost_gate_frac(engine) * min( - index.n_keys for index in adjacency - ): - return "cost_frontier" + if get_index_policy(base_graph) != "force": + # Was cost the ONLY thing in the way? Re-run the real gate with the cost + # checks disabled rather than re-deriving them here: a second copy of the + # thresholds drifts, and on engines with a tighter gate it mislabelled + # structurally-unsupported shapes as "cost_frontier". Trace-only path. + if _try_indexed_connected_bindings_state( + with_index_policy(base_graph, "force"), + ops, + engine=engine, + start_nodes=start_nodes, + alias_prefilters=alias_prefilters, + ) is not None: + return "cost_frontier" return "unsupported_shape" @@ -667,8 +608,13 @@ def try_indexed_connected_bindings_state( alias_prefilters: Optional[Any] = None, ) -> Optional[IndexedBindingsState]: """Attempt the indexed path; only explicit safe declines fall through.""" + from graphistry.compute.ast import ASTNode + hop_count = max(0, (len(ops) - 1) // 2) - first_filter = getattr(ops[0], "filter_dict", None) if ops else None + first_op_any = ops[0] if ops else None + first_filter = ( + first_op_any.filter_dict if isinstance(first_op_any, ASTNode) else None + ) node_id = base_graph._node public_seed_scan = not ( isinstance(first_filter, Mapping) diff --git a/graphistry/compute/gfql/index/build.py b/graphistry/compute/gfql/index/build.py index 344c4c9df4..9657463342 100644 --- a/graphistry/compute/gfql/index/build.py +++ b/graphistry/compute/gfql/index/build.py @@ -11,7 +11,7 @@ from graphistry.Engine import Engine from graphistry.compute.typing import DataFrameT from .engine_arrays import array_namespace, col_to_array -from .registry import AdjacencyIndex, NodeIdIndex, frame_fingerprint +from .registry import AdjacencyIndex, NodeIdIndex, NodePropIndex, frame_fingerprint from .types import AdjacencyIndexKind, ArrayLike, ArrayNamespace @@ -99,3 +99,40 @@ def build_node_id_index( source_ref=cast(DataFrameT, nodes), n_nodes=n_keys, ) + + +def build_node_prop_index( + nodes: DataFrameT, + column: str, + engine: Engine, +) -> Optional[NodePropIndex]: + """Sorted property value -> node row positions (CSR), or None when unindexable. + + Duplicates are fine (CSR keeps every row per value) — this is the secondary + index, so the caller still applies the remaining predicates to the gathered + candidates. Declines (None) for anything whose ordering/equality is not + unambiguous under a vectorized ``searchsorted`` on BOTH backends: non-integer + dtypes (float NaN ordering, object/string on cupy) and null-bearing columns. + Widening that gate later is additive — a decline only means "scan", never a + wrong answer. + """ + xp, backend = array_namespace(engine) + try: + keys = col_to_array(nodes, column, engine) + except (AttributeError, KeyError, TypeError, ValueError): + return None + if getattr(getattr(keys, "dtype", None), "kind", None) not in ("i", "u"): + return None + unique_keys, group_offsets, row_positions = _csr_from_keys(keys, xp) + return NodePropIndex( + key_col=column, + keys_sorted=unique_keys, + group_offsets=group_offsets, + row_positions=row_positions, + backend=backend, + engine=engine, + fingerprint=frame_fingerprint(nodes, (column,), engine), + source_ref=cast(DataFrameT, nodes), + n_nodes=int(keys.shape[0]), + n_keys=int(unique_keys.shape[0]), + ) diff --git a/graphistry/compute/gfql/index/cypher_ddl.py b/graphistry/compute/gfql/index/cypher_ddl.py index e289a93e49..b1c03f000a 100644 --- a/graphistry/compute/gfql/index/cypher_ddl.py +++ b/graphistry/compute/gfql/index/cypher_ddl.py @@ -19,7 +19,7 @@ from .types import IndexKind from .wire import CreateIndex, DropIndex, ShowIndexes, IndexOp -_KIND = r"(?Pedge_out_adj|edge_in_adj|node_id)" +_KIND = r"(?Pedge_out_adj|edge_in_adj|node_id|node_prop)" _CREATE_PATTERN = ( r"^\s*CREATE\s+GFQL\s+INDEX\s+(?:(?P[A-Za-z_]\w*)\s+)?FOR\s+" + _KIND diff --git a/graphistry/compute/gfql/index/lookup.py b/graphistry/compute/gfql/index/lookup.py index 14fa469021..61e4b0eb3e 100644 --- a/graphistry/compute/gfql/index/lookup.py +++ b/graphistry/compute/gfql/index/lookup.py @@ -6,9 +6,9 @@ """ from __future__ import annotations -from typing import Tuple +from typing import Any, Tuple -from .registry import AdjacencyIndex, NodeIdIndex +from .registry import AdjacencyIndex, NodeIdIndex, NodePropIndex from .types import ArrayLike, ArrayNamespace @@ -93,3 +93,69 @@ def lookup_node_rows(index: NodeIdIndex, ids: ArrayLike, xp: ArrayNamespace) -> pos_clipped = xp.where(pos < U, pos, U - 1) hit = keys[pos_clipped] == f return index.row_positions[pos_clipped[hit]] + + +def _csr_hit_positions(keys: ArrayLike, values: ArrayLike, xp: ArrayNamespace) -> ArrayLike: + """Group positions in ``keys`` for the ``values`` that are actually present. + + The shared front half of every CSR probe: promote to a common dtype (never + narrow — an int64 id cast to int32 keys wraps and false-matches), searchsorted, + then verify membership. Returns the matched group positions. + """ + U = int(keys.shape[0]) + if U == 0 or int(values.shape[0]) == 0: + return xp.zeros(0, dtype=xp.int64) + if values.dtype != keys.dtype: + common = xp.promote_types(values.dtype, keys.dtype) + values = values.astype(common) + keys = keys.astype(common) + pos = xp.searchsorted(keys, values) + clipped = xp.where(pos < U, pos, U - 1) + return clipped[keys[clipped] == values] + + +def _csr_group_sizes(index: Any, positions: ArrayLike) -> ArrayLike: + """Row count of each CSR group named by ``positions``.""" + return index.group_offsets[positions + 1] - index.group_offsets[positions] + + +def csr_match_count(index: Any, values: ArrayLike, xp: ArrayNamespace) -> int: + """How many rows a CSR gather of ``values`` would return — offsets only, no + gather. The planner's free selectivity/degree estimate.""" + positions = _csr_hit_positions(index.keys_sorted, values, xp) + if int(positions.shape[0]) == 0: + return 0 + return int(_csr_group_sizes(index, positions).sum()) + + +def csr_gather_rows(index: Any, values: ArrayLike, xp: ArrayNamespace) -> ArrayLike: + """Row positions of every row whose key is in ``values`` (CSR range expansion).""" + positions = _csr_hit_positions(index.keys_sorted, values, xp) + empty = index.row_positions[:0] + if int(positions.shape[0]) == 0: + return empty + start = index.group_offsets[positions] + counts = _csr_group_sizes(index, positions) + total = int(counts.sum()) + if total == 0: + return empty + return index.row_positions[_expand_ranges(start, counts, total, xp)] + + +def lookup_prop_rows(index: NodePropIndex, values: ArrayLike, xp: ArrayNamespace) -> ArrayLike: + """values -> node row positions of every row holding one of them. + + Order is unspecified here; callers that need frame order sort. + """ + return csr_gather_rows(index, values, xp) + + +def prop_match_count(index: NodePropIndex, values: ArrayLike, xp: ArrayNamespace) -> int: + """Rows ``lookup_prop_rows`` would return — the free selectivity estimate.""" + return csr_match_count(index, values, xp) + + +def lookup_degree(index: AdjacencyIndex, frontier: ArrayLike, xp: ArrayNamespace) -> int: + """Total incident-edge count for a frontier — the hop's fanout estimate, + computed from CSR offsets before any range expansion.""" + return csr_match_count(index, frontier, xp) diff --git a/graphistry/compute/gfql/index/registry.py b/graphistry/compute/gfql/index/registry.py index b5c2754e24..799f97ca1c 100644 --- a/graphistry/compute/gfql/index/registry.py +++ b/graphistry/compute/gfql/index/registry.py @@ -21,9 +21,10 @@ EDGE_OUT_ADJ: AdjacencyIndexKind = "edge_out_adj" EDGE_IN_ADJ: AdjacencyIndexKind = "edge_in_adj" NODE_ID: IndexKind = "node_id" +NODE_PROP: IndexKind = "node_prop" ADJ_KINDS: Tuple[AdjacencyIndexKind, ...] = (EDGE_OUT_ADJ, EDGE_IN_ADJ) -ALL_KINDS: Tuple[IndexKind, ...] = (EDGE_OUT_ADJ, EDGE_IN_ADJ, NODE_ID) +ALL_KINDS: Tuple[IndexKind, ...] = (EDGE_OUT_ADJ, EDGE_IN_ADJ, NODE_ID, NODE_PROP) FrameFingerprint = Tuple[int, Tuple[str, ...], str] @@ -81,20 +82,74 @@ class NodeIdIndex: name: Optional[str] = None +@dataclass(frozen=True) +class NodePropIndex: + """Sorted node PROPERTY value -> node row positions (CSR, duplicates allowed). + + The secondary index: a seed predicate on a non-key column (``{id: 42}`` where + the graph's node id is some other column) otherwise costs a full node scan. + Unlike :class:`NodeIdIndex` this keeps ALL rows per key in CSR form, so + non-unique properties are indexable — the caller applies any residual + predicates to the gathered candidates, so results are identical either way. + """ + key_col: str + keys_sorted: ArrayLike # distinct values, ascending (len U) + group_offsets: ArrayLike # CSR offsets into row_positions (len U+1) + row_positions: ArrayLike # node row indices grouped by value (len N) + backend: IndexBackend + engine: Engine + fingerprint: FrameFingerprint = field(compare=False, default=(-1, (), "")) + source_ref: Optional[DataFrameT] = field(compare=False, default=None) + n_nodes: int = 0 + n_keys: int = 0 + name: Optional[str] = None + + @dataclass(frozen=True) class GfqlIndexRegistry: """Immutable kind -> index map. ``with_index`` / ``without`` return copies.""" indexes: Dict[IndexKind, Union[AdjacencyIndex, NodeIdIndex]] = 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) def with_index(self, kind: IndexKind, index: Union[AdjacencyIndex, NodeIdIndex]) -> "GfqlIndexRegistry": new = dict(self.indexes) new[kind] = index - return GfqlIndexRegistry(new) + return GfqlIndexRegistry(new, dict(self.node_props)) + + def with_node_prop(self, column: str, index: "NodePropIndex") -> "GfqlIndexRegistry": + props = dict(self.node_props) + props[column] = index + return GfqlIndexRegistry(dict(self.indexes), props) + + def node_prop_cols(self) -> Tuple[str, ...]: + return tuple(sorted(self.node_props.keys())) + + def get_node_prop_valid( + self, column: str, df: Optional[DataFrameT], engine: Engine + ) -> Optional["NodePropIndex"]: + """The property index for ``column``, only while it still matches the live + frame + engine (same identity/fingerprint contract as ``get_valid``).""" + idx = self.node_props.get(column) + if idx is None or df is None or idx.engine != engine: + return None + if idx.source_ref is not None and idx.source_ref is not df: + return None + if idx.fingerprint != frame_fingerprint(df, (column,), engine): + return None + return idx def without(self, kind: IndexKind) -> "GfqlIndexRegistry": + if kind == NODE_PROP: + return GfqlIndexRegistry(dict(self.indexes), {}) new = dict(self.indexes) new.pop(kind, None) - return GfqlIndexRegistry(new) + return GfqlIndexRegistry(new, dict(self.node_props)) + + def without_node_prop(self, column: str) -> "GfqlIndexRegistry": + props = dict(self.node_props) + props.pop(column, None) + return GfqlIndexRegistry(dict(self.indexes), props) def rebind_edges(self, new_edges: DataFrameT) -> "GfqlIndexRegistry": """Re-point the EDGE adjacency indexes' identity guard at ``new_edges``. @@ -134,7 +189,7 @@ def rebind_edges(self, new_edges: DataFrameT) -> "GfqlIndexRegistry": new[kind] = replace(idx, source_ref=new_edges) else: new.pop(kind, None) - return GfqlIndexRegistry(new) + return GfqlIndexRegistry(new, dict(self.node_props)) def get(self, kind: IndexKind) -> Optional[Union[AdjacencyIndex, NodeIdIndex]]: return self.indexes.get(kind) @@ -146,7 +201,7 @@ def kinds(self) -> Tuple[IndexKind, ...]: return cast(Tuple[IndexKind, ...], tuple(sorted(self.indexes.keys()))) def is_empty(self) -> bool: - return not self.indexes + return not self.indexes and not self.node_props 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 @@ -166,7 +221,7 @@ def get_valid(self, kind: IndexKind, df: DataFrameT, cols: Tuple[str, ...], engi return idx -def index_nbytes(idx: Union[AdjacencyIndex, NodeIdIndex]) -> int: +def index_nbytes(idx: Union[AdjacencyIndex, NodeIdIndex, "NodePropIndex"]) -> int: """Approximate resident memory of an index's sidecar arrays (bytes).""" total = 0 for attr in ("keys_sorted", "group_offsets", "row_positions", "other_values"): diff --git a/graphistry/compute/gfql/index/types.py b/graphistry/compute/gfql/index/types.py index 43a7fe722f..3d6faa2547 100644 --- a/graphistry/compute/gfql/index/types.py +++ b/graphistry/compute/gfql/index/types.py @@ -11,7 +11,7 @@ if TYPE_CHECKING: from graphistry.compute.predicates.ASTPredicate import ASTPredicate -IndexKind = Literal["edge_out_adj", "edge_in_adj", "node_id"] +IndexKind = Literal["edge_out_adj", "edge_in_adj", "node_id", "node_prop"] AdjacencyIndexKind = Literal["edge_out_adj", "edge_in_adj"] IndexBackend = Literal["numpy", "cupy"] HopDirection = Literal["forward", "reverse", "undirected"] diff --git a/graphistry/compute/gfql/index/wire.py b/graphistry/compute/gfql/index/wire.py index 12322e3edf..23ccf2e793 100644 --- a/graphistry/compute/gfql/index/wire.py +++ b/graphistry/compute/gfql/index/wire.py @@ -96,12 +96,23 @@ def apply_index_op(g: Any, op: IndexOp, *, engine: Any = "auto") -> Any: CreateIndex/DropIndex -> new Plottable; ShowIndexes -> pandas DataFrame. """ + from graphistry.Engine import resolve_engine from .api import create_index, drop_index, show_indexes, get_registry, _is_resident_index_valid + from .registry import NODE_PROP + if isinstance(op, CreateIndex): if not op.replace: reg = get_registry(g) - if reg.has(op.kind) and _is_resident_index_valid(g, op.kind, engine): + if op.kind == NODE_PROP: + # Property indexes are keyed by COLUMN, not kind: reuse only the + # index for THIS column, and only while it is still valid. + if op.column is not None and reg.get_node_prop_valid( + op.column, getattr(g, "_nodes", None), + resolve_engine(engine, g), + ) is not None: + return g + elif reg.has(op.kind) and _is_resident_index_valid(g, op.kind, engine): return g # valid resident index reuse return create_index(g, op.kind, column=op.column, name=op.name, engine=engine) if isinstance(op, DropIndex): @@ -113,16 +124,35 @@ def apply_index_op(g: Any, op: IndexOp, *, engine: Any = "auto") -> Any: reg = get_registry(g) kind = next((k for k, ix in reg.indexes.items() if getattr(ix, "name", None) == op.name), None) + column = op.column + if kind is None: + # Property indexes are column-keyed, so resolve names there too. + prop_col = next((c for c in reg.node_prop_cols() + if reg.node_props[c].name == op.name), None) + if prop_col is not None: + kind, column = NODE_PROP, prop_col if kind is None: if op.missing_ok: return g # IF EXISTS semantics: dropping a missing index is a no-op + resident = sorted( + [getattr(ix, 'name', k) for k, ix in reg.indexes.items()] + + [reg.node_props[c].name or c for c in reg.node_prop_cols()] + ) raise ValueError( f"DROP GFQL INDEX: no resident index named {op.name!r} " - f"(resident: {sorted(getattr(ix, 'name', k) for k, ix in reg.indexes.items())})" + f"(resident: {resident})" ) - if kind is not None and not op.missing_ok and not get_registry(g).has(kind): - raise ValueError(f"DROP GFQL INDEX: no resident index of kind {kind!r}") - return drop_index(g, kind) + return drop_index(g, kind, column=column) + if kind is not None and not op.missing_ok: + reg = get_registry(g) + is_resident = ( + (op.column in reg.node_prop_cols() if op.column is not None + else bool(reg.node_prop_cols())) + if kind == NODE_PROP else reg.has(kind) + ) + if not is_resident: + raise ValueError(f"DROP GFQL INDEX: no resident index of kind {kind!r}") + return drop_index(g, kind, column=op.column) if isinstance(op, ShowIndexes): return show_indexes(g) raise ValueError(f"Unknown index op: {op!r}") diff --git a/graphistry/tests/compute/gfql/index/test_index.py b/graphistry/tests/compute/gfql/index/test_index.py index 0303ce0495..baab166c2f 100644 --- a/graphistry/tests/compute/gfql/index/test_index.py +++ b/graphistry/tests/compute/gfql/index/test_index.py @@ -145,6 +145,46 @@ def test_wire_roundtrip(graph): show = g.gfql({"type": "ShowIndexes"}) assert show.shape[0] == 1 +def test_node_prop_ddl_and_wire_surfaces(graph): + """The property index must be reachable from all three surfaces (review of #1777): + Python, Cypher DDL, and the JSON wire protocol — including targeted drops.""" + from graphistry.compute.gfql.index import NODE_PROP + + # Cypher DDL + assert parse_index_ddl("CREATE GFQL INDEX FOR node_prop ON lab").column == "lab" + g = graph.gfql("CREATE GFQL INDEX FOR node_prop ON lab") + assert get_registry(g).node_prop_cols() == ("lab",) + assert g.gfql("SHOW GFQL INDEXES").shape[0] == 1 + + # JSON wire, incl. a second column, then a TARGETED drop of just one of them + g = g.gfql({"type": "CreateIndex", "kind": "node_prop", "column": "id"}) + assert get_registry(g).node_prop_cols() == ("id", "lab") + g2 = g.gfql({"type": "DropIndex", "kind": "node_prop", "column": "lab"}) + assert get_registry(g2).node_prop_cols() == ("id",) + + # kind-wide drop, and idempotent re-create of a still-valid resident index + assert get_registry(g.gfql({"type": "DropIndex", "kind": "node_prop"})).node_prop_cols() == () + again = g.gfql({"type": "CreateIndex", "kind": "node_prop", "column": "id"}) + assert get_registry(again).node_props["id"] is get_registry(g).node_props["id"] + + +def test_node_prop_drop_resident_does_not_raise(graph): + """`has()` only knows kind-keyed indexes, so a resident property index must not + report itself missing (review of #1777).""" + g = graph.gfql("CREATE GFQL INDEX FOR node_prop ON lab") + assert get_registry(g.gfql("DROP GFQL INDEX FOR node_prop")).node_prop_cols() == () + with pytest.raises(ValueError): + graph.gfql("DROP GFQL INDEX FOR node_prop") # none resident, no IF EXISTS + assert graph.gfql("DROP GFQL INDEX IF EXISTS FOR node_prop") is not None + + +def test_node_prop_drop_by_name(graph): + """A custom-named property index resolves by name, like the kind-keyed ones.""" + g = graph.create_index("node_prop", column="lab", name="by_lab") + assert get_registry(g).node_props["lab"].name == "by_lab" + assert get_registry(g.gfql("DROP GFQL INDEX by_lab")).node_prop_cols() == () + + def test_create_rebuilds_stale_resident_index(): g = graphistry.edges(pd.DataFrame({"src": [0, 1], "dst": [1, 2]}), "src", "dst").materialize_nodes() gi = g.gfql("CREATE GFQL INDEX FOR edge_out_adj") diff --git a/graphistry/tests/compute/gfql/index/test_indexed_bindings.py b/graphistry/tests/compute/gfql/index/test_indexed_bindings.py index acfbd28166..55e9a6d367 100644 --- a/graphistry/tests/compute/gfql/index/test_indexed_bindings.py +++ b/graphistry/tests/compute/gfql/index/test_indexed_bindings.py @@ -526,6 +526,293 @@ def record_filter(frame: Any, *args: Any, **kwargs: Any) -> Any: assert filtered_lengths[0] == 1 +# --- secondary (node property) index ----------------------------------------- +# CONNECTED_QUERY seeds on ``public``, which is NOT the graph's node-id binding +# (``id``), so without a property index the seed costs a full node scan. + + +def _prop_frames() -> Tuple[pd.DataFrame, pd.DataFrame]: + nodes, edges = _base_frames() + # ``grp`` repeats (4 rows per value): the duplicate-key case a node-id index + # cannot express but a CSR property index can. + nodes = nodes.assign(grp=(nodes["rank"] % 3).astype("int64")) + return nodes, edges + + +def _prop_graph(engine: str, columns: Sequence[str] = ("public",)) -> Any: + nodes, edges = _prop_frames() + return _graph_from_frames(engine, nodes, edges).gfql_index_node_props( + list(columns), engine=engine + ) + + +def _seed_filter_widths( + g: Any, query: Any, engine: str, monkeypatch: pytest.MonkeyPatch +) -> Tuple[Any, List[Dict[str, Any]], List[int]]: + """Run traced, recording how many rows each helper filter had to look at.""" + import graphistry.compute.gfql.index.bindings as indexed_bindings + + widths: List[int] = [] + original = indexed_bindings._filter_frame + + def record(frame: Any, *args: Any, **kwargs: Any) -> Any: + widths.append(int(frame.shape[0])) + return original(frame, *args, **kwargs) + + monkeypatch.setattr(indexed_bindings, "_filter_frame", record) + actual, steps = _trace_run(g, query, engine) + return actual, steps, widths + + +@pytest.mark.parametrize("engine", ENGINES) +def test_node_property_index_seeds_without_scanning( + engine: str, + monkeypatch: pytest.MonkeyPatch, +) -> None: + g = _prop_graph(engine) + with monkeypatch.context() as m: + expected = _run(g, CONNECTED_QUERY, engine, m, generic=True) + actual, steps, widths = _seed_filter_widths( + g, CONNECTED_QUERY, engine, monkeypatch + ) + _assert_result_exact(actual, expected, engine) + decisions = [s for s in steps if s.get("seam") == "connected_bindings"] + assert len(decisions) == 1 + _assert_decision(decisions[0], seam="connected_bindings", served=True) + assert widths and widths[0] == 1 # one indexed candidate, not the node table + + +@pytest.mark.parametrize("engine", ENGINES) +def test_node_property_index_absent_matches_indexed( + engine: str, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Same query, same answer, with and without the secondary index.""" + nodes, edges = _prop_frames() + plain = _graph_from_frames(engine, nodes, edges) + indexed = plain.gfql_index_node_props(["public"], engine=engine) + with monkeypatch.context() as m: + expected = _run(plain, CONNECTED_QUERY, engine, m, generic=True) + for g in (plain, indexed): + actual, _ = _trace_run(g, CONNECTED_QUERY, engine) + _assert_result_exact(actual, expected, engine) + + +def test_node_property_index_duplicate_values_match_scan( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A non-unique property still gathers EVERY matching row (CSR, not first-hit).""" + from graphistry.compute.ast import rows as rows_call + + g = _prop_graph("pandas", columns=("grp",)) + query = [n({"grp": 0}, name="a"), e_forward({"type": "A"}, name="r"), n(name="b"), rows_call()] + with monkeypatch.context() as m: + expected = _run(g, query, "pandas", m, generic=True) + actual, steps, widths = _seed_filter_widths(g, query, "pandas", monkeypatch) + _assert_result_exact(actual, expected, "pandas") + assert widths and widths[0] == 4 # every row with grp == 0, none of the others + assert [s for s in steps if s.get("seam") == "connected_bindings"] + + +@pytest.mark.parametrize("case", ["stale", "policy_off"]) +def test_node_property_index_lifecycle_falls_back( + case: str, + monkeypatch: pytest.MonkeyPatch, +) -> None: + g = _prop_graph("pandas") + index_policy = "off" if case == "policy_off" else "force" + if case == "stale": + g = g.nodes(_native_copy(g._nodes), "id") # rebind -> index treated as absent + with monkeypatch.context() as m: + expected = _run( + g, CONNECTED_QUERY, "pandas", m, generic=True, index_policy=index_policy + ) + actual, _ = _trace_run(g, CONNECTED_QUERY, "pandas", index_policy=index_policy) + _assert_result_exact(actual, expected, "pandas") + + +def test_node_property_index_declines_unindexable_columns() -> None: + """Unindexable DTYPE is skippable; a caller mistake is not. + + The convenience builder suppresses exactly one condition — a column whose dtype + this index cannot serve. A missing column, a missing argument, or any other + failure must propagate, or a typo would silently leave the query unindexed. + """ + from graphistry.compute.gfql.index import ( + NODE_PROP, GfqlIndexUnsupportedError, create_index, get_registry, + ) + + g = _prop_graph("pandas", columns=()) + for column in ("kind", "maybe"): # object dtype, float-with-null + with pytest.raises(GfqlIndexUnsupportedError): + create_index(g, NODE_PROP, column=column) + # caller mistakes are NOT the skippable kind + for bad in ({"column": "nosuch"}, {}): + with pytest.raises(ValueError) as excinfo: + create_index(g, NODE_PROP, **bad) + assert not isinstance(excinfo.value, GfqlIndexUnsupportedError) + + # the convenience wrapper skips the unindexable dtypes and indexes what it can + g2 = g.gfql_index_node_props(["kind", "maybe", "public"]) + assert get_registry(g2).node_prop_cols() == ("public",) + # ...but does NOT swallow a real failure + with pytest.raises(ValueError): + g.gfql_index_node_props(["nosuch"]) + + +def test_node_property_index_prefers_the_most_selective_column( + monkeypatch: pytest.MonkeyPatch, +) -> None: + from graphistry.compute.ast import rows as rows_call + + g = _prop_graph("pandas", columns=("public", "grp")) + query = [ + n({"public": 100, "grp": 0}, name="a"), + e_forward({"type": "A"}, name="r"), + n(name="b"), + rows_call(), + ] + with monkeypatch.context() as m: + expected = _run(g, query, "pandas", m, generic=True) + actual, _, widths = _seed_filter_widths(g, query, "pandas", monkeypatch) + _assert_result_exact(actual, expected, "pandas") + assert widths and widths[0] == 1 # 'public' (1 match) beats 'grp' (4 matches) + + +@pytest.mark.parametrize( + "seed,indexed_column,expect_gathered", + [ + pytest.param({"public": 100}, "public", True, id="selective-uses-index"), + pytest.param({"grp": 0}, "grp", False, id="unselective-keeps-scan"), + ], +) +def test_node_property_index_cost_gate_under_policy_use( + seed: Dict[str, Any], + indexed_column: str, + expect_gathered: bool, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Under the default `use` policy the gate must reject a non-selective predicate. + + `force` skips the gate, so the other property-index cases cannot exercise it. + The crossover is pinned here via the public knob rather than relying on the + engine default, so the test states the threshold it is testing: at 25%, + `public` (1 of 12 rows) gathers through the index and `grp` (4 of 12) does not. + Either way the answer is the canonical one. + """ + from graphistry.compute.ast import rows as rows_call + from graphistry.compute.gfql.index import reset_cost_gate_frac, set_cost_gate_frac + + set_cost_gate_frac(Engine.PANDAS, 0.25) + try: + g = _prop_graph("pandas", columns=(indexed_column,)) + query = [ + n(seed, name="a"), e_forward({"type": "A"}, name="r"), n(name="b"), rows_call(), + ] + with monkeypatch.context() as m: + expected = _run(g, query, "pandas", m, generic=True, index_policy="use") + + import graphistry.compute.gfql.index.bindings as indexed_bindings + + widths: List[int] = [] + original = indexed_bindings._filter_frame + + def record(frame: Any, *args: Any, **kwargs: Any) -> Any: + widths.append(int(frame.shape[0])) + return original(frame, *args, **kwargs) + + monkeypatch.setattr(indexed_bindings, "_filter_frame", record) + actual, _ = _trace_run(g, query, "pandas", index_policy="use") + finally: + reset_cost_gate_frac(Engine.PANDAS) + + _assert_result_exact(actual, expected, "pandas") + assert widths + n_nodes = int(g._nodes.shape[0]) + if expect_gathered: + assert widths[0] < n_nodes # indexed candidates only + else: + assert widths[0] == n_nodes # gate declined -> canonical scan + + +def test_node_property_index_shows_and_drops() -> None: + from graphistry.compute.gfql.index import NODE_PROP, get_registry + + g = _prop_graph("pandas", columns=("public", "grp")) + shown = g.show_indexes() + props = shown[shown["kind"] == NODE_PROP] + assert sorted(props["key_col"]) == ["grp", "public"] + assert bool(props["valid"].all()) + assert sorted(props["n_keys"]) == [3, 12] + assert get_registry(g.drop_index(NODE_PROP, column="grp")).node_prop_cols() == ("public",) + assert get_registry(g.drop_index(NODE_PROP)).node_prop_cols() == () + assert get_registry(g.drop_index()).is_empty() + + +@pytest.mark.parametrize( + "suffix_params,seeded,reason", + [ + pytest.param({"source": "a"}, False, "rows-with-source", id="rows-source"), + pytest.param({"alias_endpoints": {"a": "src"}}, False, "alias-endpoints", id="alias-endpoints"), + pytest.param({"alias_prefilters": {"b": {"rank": 1}}}, False, "prefiltered", id="alias-prefilters"), + pytest.param({}, True, "seeded re-entry", id="carried-seed"), + ], +) +def test_polars_early_gate_refuses_unsupported_boundaries( + suffix_params: Dict[str, Any], + seeded: bool, + reason: str, +) -> None: + """The polars bypass must not engage on shapes its rows call does not consume. + + Engaging on any of these would hand the materializer a compact state for a plan + it is not executing, so each refusal condition is asserted on the gate directly. + """ + pytest.importorskip("polars") + from graphistry.compute.ast import rows as rows_call + from graphistry.compute.gfql.lazy.engine.polars.chain import _try_indexed_middle_polars + + g = _graph("polars") + middle = [n({"public": 100}, name="a"), e_forward({"type": "A"}, name="r"), n(name="b")] + start_nodes = g._nodes.head(1) if seeded else None + + state, attempted = _try_indexed_middle_polars( + g, middle, [rows_call(**suffix_params)], start_nodes + ) + assert state is None, f"polars gate engaged on {reason}" + assert attempted is False, f"polars gate recorded an attempt on {reason}" + + +def test_polars_early_gate_requires_the_whole_middle() -> None: + """binding_ops that do not cover the middle must not take the bypass.""" + pytest.importorskip("polars") + from graphistry.compute.ast import rows as rows_call + from graphistry.compute.chain import serialize_binding_ops + from graphistry.compute.gfql.lazy.engine.polars.chain import _try_indexed_middle_polars + + g = _graph("polars") + middle = [n({"public": 100}, name="a"), e_forward({"type": "A"}, name="r"), n(name="b")] + + other = serialize_binding_ops([n({"public": 101}, name="a"), e_forward(), n(name="b")]) + state, attempted = _try_indexed_middle_polars( + g, middle, [rows_call(binding_ops=other)], None + ) + assert state is None and attempted is False # a different plan + + unnamed = [n({"public": 100}), e_forward({"type": "A"}), n()] + state, attempted = _try_indexed_middle_polars(g, unnamed, [rows_call()], None) + assert state is None and attempted is False # rewrite would not install them + + # the supported shape DOES engage (guards against a vacuous negative suite); + # `force` skips the cost gate, which a 12-row fixture would otherwise trip + from graphistry.compute.gfql.index import with_index_policy + + state, attempted = _try_indexed_middle_polars( + with_index_policy(g, "force"), middle, [rows_call()], None + ) + assert attempted is True and state is not None + + @pytest.mark.parametrize("engine", ENGINES) def test_indexed_execution_is_pure( engine: str, @@ -546,6 +833,7 @@ def test_indexed_execution_is_pure( pd.testing.assert_frame_equal(_to_pandas(g._edges), edges_before) +@pytest.mark.parametrize("engine", ENGINES) @pytest.mark.parametrize( "ops,kwargs", [ @@ -604,18 +892,25 @@ def test_indexed_execution_is_pure( ], ) def test_unsupported_shapes_decline_before_work( + engine: str, ops: Sequence[Any], kwargs: Dict[str, Any], ) -> None: import graphistry.compute.gfql.index.bindings as indexed_bindings from graphistry.compute.gfql.index import index_trace + call_kwargs = dict(kwargs) + if engine != "pandas" and "start_nodes" in call_kwargs: + native_seed, _ = _native_frames( + engine, call_kwargs["start_nodes"], _base_frames()[1] + ) + call_kwargs["start_nodes"] = native_seed with index_trace() as captured: out = indexed_bindings.try_indexed_connected_bindings_state( - _graph("pandas"), + _graph(engine), ops, - engine=Engine.PANDAS, - **kwargs, + engine=ENGINE_ENUM[engine], + **call_kwargs, ) assert out is None decisions = [