diff --git a/CHANGELOG.md b/CHANGELOG.md index 346f1d8214..a0da6619bd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,18 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm ## [Development] +### 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. +- **Seeded typed-hop property projection keeps its fast path through DISTINCT / ORDER BY / SKIP / LIMIT**: those trailing row ops are plain frame operations, so the fast path now delegates them to the canonical chain instead of declining the whole query. + +### Changed +- **GFQL execution context is declared rather than attached at runtime**: the private `_gfql_*` per-execution fields (index policy and registry, row-pipeline base graph, carried seed nodes, edge aliases, shortest-path backend, and the indexed-bindings handoff) are now declared on `Plottable` with defaults on `PlotterBase`, instead of being set with `setattr` and read back with `getattr(..., default)`. No public API or behaviour change; internal call sites are typed, and hand-rolled `Plottable` stand-ins must now construct the full context. + +### Fixed +- **Indexed bypass could return every node for `rows()` over an unnamed pattern**: the boundary accepted a `rows()` call carrying no binding ops over a middle with no aliases. That call reads the traversal-narrowed node table, but the bypass hands it the full graph, so the query would have returned all nodes. The gate now requires that the rows call actually consumes the whole middle as binding ops — either it already carries them, or the named-middle rewrite installs exactly them. +- **Seeded property-RETURN dtype divergence on cuDF**: the lean projection applied the pandas rows-pivot artifact (int → float64, bool → object) on every engine, but cuDF's canonical pivot preserves the source dtypes — so the fast path returned `float64`/`object` where cuDF's own canonical path returns `int64`/`bool`. The cast rule is now engine-aware; the dtype-class decline guard is unchanged. + ### Documentation - **GFQL pay-as-you-go resident indexing user guide**: New :doc:`Pay-As-You-Go Resident Indexing ` page — the lifecycle guide to resident indexes (`gfql_index_all()` / `gfql_index_edges()` / `create_index()` / `show_indexes()` / `drop_index()`): what the node-id + CSR in/out adjacency sidecars are, what engages them on 0.58.0 (seeded typed-hop fast paths incl. property RETURNs and property-seeded lookups per #1768/#1770, direct `g.hop()`; the general polars chain traversal honestly noted as not yet covered), the staleness/validity contract (identity + fingerprint; rebind invalidates; declines are safe — identical results either way), engine notes (polars needs `gfql_index_all(engine='polars')` until #1767), 0.58.0-tag measured numbers, and a runnable end-to-end example. Wired into the GFQL toctree + recommended paths alongside :doc:`Seeded Traversal Indexes `. - **GFQL performance docs: 0.58.0 release-tag-verified numbers, siloed in one page**: `gfql/performance.rst` is now the canonical benchmark-numbers page (alongside `gfql/index_adjacency.rst` for the index benchmarks) — a benchmark rerun updates it alone. It carries the 0.58.0 tag sweep (DGX Spark GB10, warm medians N=30; four-engine numbers cross-engine parity-verified, competitor pairs validated against expected result rows): seeded typed-hop fast path across all four engines (e.g. pandas 29.9→2.46ms, 12.1×), native chain form, resident-index covered-shape lookups (with the `gfql_index_all(engine='polars')` caveat / PR #1767), flat seeded-hop scaling on pandas (0.159–0.164ms from 0.25M to 32M edges), the one-keyword `engine='polars'` LDBC SNB SF1 seed-lookup win (1,299.6→106.1ms, 12.3×), LDBC SNB interactive SF1 vs Neo4j 5.26 same-box pairs (GFQL 4 of 5; Neo4j wins recent-replies — reported as-is), OLAP multi-join vs embedded Kuzu (q8 200×, q9 14.2×) with the honest inverse (Kuzu wins single-table aggregates 2–4×, seeded property-projection lookups 2.4–64×), plus the prior Orkut/LiveJournal bulk sweep (moved from `engines.rst`, dated once) and its methodology. All other pages — `engines.rst`, `quick.rst`, `about.rst`, `overview.rst`, `index.rst` — now carry stable qualitative claims (e.g. "often an order of magnitude faster on query-heavy workloads") that link into `performance.rst` instead of inline figures, replacing the stale "up to ~38×" headlines and avoiding scattered per-claim version labels. diff --git a/agents/skills/review/SKILL.md b/agents/skills/review/SKILL.md index 64b3c1f184..376f95e591 100644 --- a/agents/skills/review/SKILL.md +++ b/agents/skills/review/SKILL.md @@ -211,6 +211,20 @@ GFQL is vectorization-first and pure-functional. The row pipeline runs on pandas Vectorized mutation (`df.loc[mask, col] = v`) is still mutation — prefer `df.assign(...)`. +**Batch the assign** (SUGGESTION; IMPORTANT on a hot row path): copy-then-mutate +(`out = df.copy()` followed by one or more `out[col] = ...`) should be a single +`df.assign(col_a=..., col_b=...)`. It is the functional phrasing the rest of the +codebase uses, it drops the explicit copy (assign already returns a new frame), and +one assign builds the result once instead of copying and then writing column by +column. + +| Pattern | Fix | +|---|---| +| `out = df.copy()` + `out[c] = v` | `df.assign(c=v)` | +| `out = df.copy()` + several `out[c] = v` | one `df.assign(c1=v1, c2=v2)` | +| `df.copy().rename(columns=...)` | `df.rename(columns=...)` — rename already copies | +| `out = df[mask].copy()` with no later mutation | `df[mask]` — masking already copies | + **cuDF compatibility** — flag IMPORTANT unless noted: | Pattern | Fix | diff --git a/bin/ci_cypher_surface_guard_baseline.json b/bin/ci_cypher_surface_guard_baseline.json index 615454cca5..36b69ed92c 100644 --- a/bin/ci_cypher_surface_guard_baseline.json +++ b/bin/ci_cypher_surface_guard_baseline.json @@ -13,5 +13,5 @@ "max_properties": 0 } }, - "lowering_py_max_lines": 9249 + "lowering_py_max_lines": 9255 } diff --git a/graphistry/Plottable.py b/graphistry/Plottable.py index 970dd228ec..52fdb5a03a 100644 --- a/graphistry/Plottable.py +++ b/graphistry/Plottable.py @@ -1,4 +1,4 @@ -from typing import TYPE_CHECKING, Any, Callable, Dict, List, Optional, Set, Tuple, Union, Protocol, overload +from typing import TYPE_CHECKING, Any, Callable, Dict, Iterable, List, Optional, Set, Tuple, Union, Protocol, overload from typing_extensions import Literal, runtime_checkable import pandas as pd @@ -22,6 +22,10 @@ from graphistry.models.surfaces.graphistry_frontend.url_params import URLParamsDict if TYPE_CHECKING: + from graphistry.compute.typing import DataFrameT + from graphistry.compute.gfql.index.handoff import IndexedBindingsHandoff + from graphistry.compute.gfql.index.policy import IndexPolicy + from graphistry.compute.gfql.index.registry import GfqlIndexRegistry try: from umap import UMAP except: @@ -56,6 +60,17 @@ class Plottable(Protocol): session: ClientSession _pygraphistry: AuthManagerProtocol + # GFQL execution context. Declared here (rather than smuggled onto instances + # with setattr) so every access is typed and the defaults live in one place — + # see PlotterBase for the values. Internal: not part of the user surface. + _gfql_index_policy: "IndexPolicy" + _gfql_index_registry: Optional["GfqlIndexRegistry"] + _gfql_indexed_bindings_handoff: Optional["IndexedBindingsHandoff"] + _gfql_rows_base_graph: Optional["Plottable"] + _gfql_start_nodes: Optional["DataFrameT"] + _gfql_rows_edge_aliases: Optional[Iterable[str]] + _gfql_shortest_path_backend: str + _edges : Any _nodes : Any _source : Optional[str] diff --git a/graphistry/PlotterBase.py b/graphistry/PlotterBase.py index c512b734e0..ba0920e8ae 100644 --- a/graphistry/PlotterBase.py +++ b/graphistry/PlotterBase.py @@ -1,5 +1,11 @@ from graphistry.Plottable import Plottable, RenderModes, RenderModesConcrete -from typing import Any, Callable, Dict, List, Optional, Union, Tuple, cast, overload, TYPE_CHECKING +from typing import Any, Callable, Dict, Iterable, List, Optional, Union, Tuple, cast, overload, TYPE_CHECKING + +if TYPE_CHECKING: + from graphistry.compute.typing import DataFrameT + from graphistry.compute.gfql.index.handoff import IndexedBindingsHandoff + from graphistry.compute.gfql.index.policy import IndexPolicy + from graphistry.compute.gfql.index.registry import GfqlIndexRegistry from typing_extensions import Literal from graphistry.io.types import ComplexEncodingsDict from graphistry.models.collections import CollectionsInput @@ -213,6 +219,18 @@ class PlotterBase(Plottable): The class supports convenience methods for mixing calls across Pandas, NetworkX, and IGraph. """ + # GFQL execution context defaults (fields declared on Plottable). Immutable + # scalars / None only, so sharing the class attribute is safe; `bind()` is a + # shallow copy, so a per-graph value set on an instance travels with it. + _gfql_index_policy: "IndexPolicy" = "use" + _gfql_index_registry: Optional["GfqlIndexRegistry"] = None + _gfql_indexed_bindings_handoff: Optional["IndexedBindingsHandoff"] = None + _gfql_rows_base_graph: Optional["Plottable"] = None + _gfql_start_nodes: Optional["DataFrameT"] = None + _gfql_rows_edge_aliases: Optional[Iterable[str]] = None + _gfql_shortest_path_backend: str = "auto" + _g: Optional["Plottable"] = None # set only on row-pipeline adapters + _defaultNodeId = NODE _defaultEdgeSourceId = SRC _defaultEdgeDestinationId = DST diff --git a/graphistry/compute/chain.py b/graphistry/compute/chain.py index 18a9b3ebb4..b62bd23918 100644 --- a/graphistry/compute/chain.py +++ b/graphistry/compute/chain.py @@ -8,7 +8,7 @@ from graphistry.Engine import safe_merge from graphistry.util import setup_logger from graphistry.utils.json import JSONVal -from .ast import ASTObject, ASTNode, ASTEdge, Direction, from_json as ASTObject_from_json, serialize_binding_ops +from .ast import ASTObject, ASTNode, ASTEdge, ASTCall, Direction, from_json as ASTObject_from_json, serialize_binding_ops from .typing import DataFrameT, SeriesT from .util import generate_safe_column_name from .chain_fast_paths import _seeded_typed_hop_pandas_cudf @@ -25,6 +25,7 @@ if TYPE_CHECKING: from graphistry.compute.exceptions import GFQLSchemaError, GFQLValidationError + from graphistry.compute.gfql.index.handoff import IndexedBindingsHandoff logger = setup_logger(__name__) @@ -556,6 +557,71 @@ def _get_boundary_calls(ops: List[ASTObject]) -> Tuple[List[ASTObject], List[AST return (prefix, middle, suffix) + +def _plan_indexed_middle( + g: Plottable, + prefix: List[ASTObject], + middle: List[ASTObject], + suffix: List[ASTObject], + engine: Union[EngineAbstract, str], + policy: Optional[Any], + start_nodes: Optional[DataFrameT], +) -> Optional["IndexedBindingsHandoff"]: + """Decide ONCE whether the resident indexes can serve this boundary's middle. + + Returns ``None`` when the indexed path does not apply to this shape at all, a + handoff carrying a state when it serves, and a handoff without one when it was + tried and declined. The twin of ``_try_indexed_middle_polars`` on the polars + chain; keeping both as a single predicate-plus-plan keeps the two boundaries + comparable instead of two inline condition chains that drift. + """ + from .gfql.index.handoff import IndexedBindingsHandoff + + if not (middle and suffix) or prefix or start_nodes is not None or policy: + return None + call = suffix[0] + if not isinstance(call, ASTCall) or call.function != "rows": + return None + if ( + call.params.get("source") is not None + or call.params.get("alias_endpoints") is not None + or call.params.get("alias_prefilters") + or not all(isinstance(op, (ASTNode, ASTEdge)) for op in middle) + ): + return None + plan = serialize_binding_ops(middle) + # The bypass is only sound when the rows call actually consumes the WHOLE + # middle as binding ops: either it already carries them, or the named-middle + # rewrite installs exactly them. A rows call with no binding ops over an + # UNNAMED middle instead reads the traversal-narrowed node table, which the + # bypass would not produce. + if not ( + call.params.get("binding_ops") == plan + or ( + call.params.get("binding_ops") is None + and any(op._name is not None for op in middle + if isinstance(op, (ASTNode, ASTEdge))) + ) + ): + return None + + engine_concrete = resolve_engine(engine, g) # type: ignore[arg-type] + if engine_concrete not in (Engine.PANDAS, Engine.CUDF): + return None + + from .gfql.index.bindings import try_indexed_connected_bindings_state + + state = try_indexed_connected_bindings_state(g, middle, engine=engine_concrete) + return IndexedBindingsHandoff( + binding_ops=plan, + state=state, + edge_aliases=tuple( + op._name for op in middle + if isinstance(op, ASTEdge) and isinstance(op._name, str) + ) if state is not None else (), + ) + + def _handle_boundary_calls( self: Plottable, ops: List[ASTObject], @@ -606,6 +672,10 @@ def _handle_boundary_calls( logger.debug('Boundary call pattern detected: prefix=%s, middle=%s, suffix=%s', len(prefix), len(middle), len(suffix)) + # Function-scope import: `gfql.index` transitively imports this module, so a + # module-scope import would be a cycle. + from .gfql.index.handoff import attach_handoff + g_temp = self suffix_base_graph = g_temp @@ -622,24 +692,40 @@ def _handle_boundary_calls( ) suffix_base_graph = g_temp - if middle: - logger.debug('Executing middle operations: %s', middle) - g_temp = _chain_impl( - g_temp, - middle, - engine, - validate_schema, - policy, - context, - start_nodes - ) + # ONE value carries the whole decision: None = the indexed path does not apply + # to this boundary, a handoff WITH state = it serves, a handoff WITHOUT state = + # it was tried and declined (which the row materializer must know, so it does + # not re-attempt the same plan after the canonical traversal). + handoff = _plan_indexed_middle( + self, prefix, middle, suffix, engine, policy, start_nodes, + ) + served = handoff is not None and handoff.state is not None + + if served: + assert handoff is not None # narrowed by `served` + g_temp = attach_handoff(self, handoff) + else: + if middle: + logger.debug('Executing middle operations: %s', middle) + g_temp = _chain_impl( + g_temp, + middle, + engine, + validate_schema, + policy, + context, + start_nodes + ) + if handoff is not None: + # attach (not mutate): `g_temp` may still be the caller's graph on + # paths where nothing rebuilt it, and this is internal plumbing. + g_temp = attach_handoff(g_temp, handoff) if suffix: logger.debug('Executing boundary suffix calls: %s', suffix) if start_nodes is not None: - setattr(g_temp, "_gfql_start_nodes", start_nodes) - setattr(g_temp, "_gfql_rows_base_graph", suffix_base_graph) - setattr(g_temp, "_gfql_shortest_path_backend", getattr(g_temp, "_gfql_shortest_path_backend", "auto")) + g_temp._gfql_start_nodes = start_nodes + g_temp._gfql_rows_base_graph = suffix_base_graph if ( middle and any(getattr(op, "_name", None) is not None for op in middle) @@ -898,6 +984,14 @@ def chain( # undirected multi-edge); that honest signal propagates to the caller. _tgt = ExecutionTarget.GPU if engine_concrete_early == Engine.POLARS_GPU else ExecutionTarget.CPU with target_mode(_tgt): + if policy: + from graphistry.compute.gfql.call.executor import _thread_local as call_thread_local + old_policy = getattr(call_thread_local, 'policy', None) + try: + call_thread_local.policy = policy + return chain_polars(self, ops, start_nodes=start_nodes) + finally: + call_thread_local.policy = old_policy return chain_polars(self, ops, start_nodes=start_nodes) if policy: diff --git a/graphistry/compute/chain_fast_paths.py b/graphistry/compute/chain_fast_paths.py index 70066f6460..45dd638b64 100644 --- a/graphistry/compute/chain_fast_paths.py +++ b/graphistry/compute/chain_fast_paths.py @@ -54,9 +54,11 @@ def _resident_seed_indexes( identity via get_valid), else None — callers keep the scan path, so a stale or absent index can never change results, only speed.""" from graphistry.Engine import Engine, is_polars_df - from graphistry.compute.gfql.index import get_registry + from graphistry.compute.gfql.index import get_index_policy, get_registry from graphistry.compute.gfql.index.registry import EDGE_OUT_ADJ, EDGE_IN_ADJ, NODE_ID from graphistry.compute.gfql.index.engine_arrays import array_namespace + if get_index_policy(g) == "off": + return None registry = get_registry(g) if registry.is_empty(): return None diff --git a/graphistry/compute/gfql/cypher/lowering.py b/graphistry/compute/gfql/cypher/lowering.py index 65895bd6fa..91b0e0c09a 100644 --- a/graphistry/compute/gfql/cypher/lowering.py +++ b/graphistry/compute/gfql/cypher/lowering.py @@ -2130,8 +2130,14 @@ def __init__(self, table_df: Any) -> None: self._destination = None self._edge = None self._g = self + # Full GFQL execution context (see Plottable): this stand-in flows through + # the same row pipeline, so every field must exist or typed access breaks. self._gfql_start_nodes = None self._gfql_rows_base_graph = None + self._gfql_rows_edge_aliases = None + self._gfql_indexed_bindings_handoff = None + self._gfql_index_policy = "use" + self._gfql_index_registry = None self._gfql_shortest_path_backend = "auto" def bind(self) -> "_SyntheticRowGraph": diff --git a/graphistry/compute/gfql/index/api.py b/graphistry/compute/gfql/index/api.py index 0ed74c3d4c..6d7a38915d 100644 --- a/graphistry/compute/gfql/index/api.py +++ b/graphistry/compute/gfql/index/api.py @@ -70,21 +70,52 @@ def _trace_active() -> bool: return _get_trace_steps() is not None +def _record_indexed_traversal( + *, + seam: str, + engine: Engine, + served: bool, + reason: str, + hop_count: int, + public_seed_scan: bool, + hop_details: Optional[List[Dict[str, object]]] = None, +) -> None: + """Record one backward-compatible indexed traversal decision when tracing.""" + if not _trace_active(): + return + path = "index" if served else "scan" + _record(cast(IndexTraceStep, { + "op": "indexed_traversal", + "operation": "indexed_traversal", + "seam": seam, + "engine": engine.value, + "served": served, + "reason": reason, + "hops": hop_count, + "hop_count": hop_count, + "public_seed_scan": public_seed_scan, + "hop_details": [] if hop_details is None else hop_details, + "path": path, + "decision_reason": reason, + })) + + # Back-compat for existing private tests while helpers live in cost.py. _seed_id_array = seed_id_array _seed_deg_sum = seed_deg_sum def get_registry(g: Plottable) -> GfqlIndexRegistry: - return cast(GfqlIndexRegistry, getattr(g, REGISTRY_ATTR, EMPTY_REGISTRY)) + registry = g._gfql_index_registry + return registry if registry is not None else EMPTY_REGISTRY def get_index_policy(g: Plottable) -> IndexPolicy: - return cast(IndexPolicy, getattr(g, POLICY_ATTR, "use")) + return g._gfql_index_policy def _attach(g: Plottable, registry: GfqlIndexRegistry) -> Plottable: res = copy.copy(g) - setattr(res, REGISTRY_ATTR, registry) + res._gfql_index_registry = registry return res diff --git a/graphistry/compute/gfql/index/bindings.py b/graphistry/compute/gfql/index/bindings.py new file mode 100644 index 0000000000..2dfbd72f90 --- /dev/null +++ b/graphistry/compute/gfql/index/bindings.py @@ -0,0 +1,722 @@ +"""Index-driven fixed-hop connected binding rows. + +This module is deliberately semantic rather than query-shaped: it recognizes a +small, exact subset of alternating node/edge ASTs and returns a same-engine path +bag, or ``None`` before visible mutation when the resident indexes, semantics, +or cost are unsuitable. Canonical row builders remain responsible for property +attachment and every suffix operation. +""" +from __future__ import annotations + +from dataclasses import dataclass +from numbers import Integral +from typing import Any, Dict, List, Mapping, Optional, Sequence, Tuple, cast + +from graphistry.Engine import Engine +from graphistry.Plottable import Plottable +from graphistry.compute.typing import DataFrameT + +from .api import ( + _record_indexed_traversal, + _trace_active, + get_index_policy, + get_registry, +) +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 .traverse import _indices_for_direction + + +_CURRENT = "__current__" +_FROM = "__gfql_ib_from__" +_TO = "__gfql_ib_to__" +_PATH_ORD = "__gfql_ib_path_ord__" +_EDGE_ORD = "__gfql_ib_edge_ord__" +_ORIENT_ORD = "__gfql_ib_orient_ord__" +_LEFT_N = "__gfql_ib_left_n__" +_RIGHT_N = "__gfql_ib_right_n__" +_INTERNAL = { + _CURRENT, _FROM, _TO, _PATH_ORD, _EDGE_ORD, _ORIENT_ORD, + _LEFT_N, _RIGHT_N, +} + + +@dataclass(frozen=True) +class IndexedBindingsState: + """A complete fixed-hop path bag ready for canonical materialization.""" + + state: DataFrameT + alias_frames: Dict[str, DataFrameT] + engine: Engine + hop_count: int + estimated_rows: int + + +def _plain_scalar_filter(value: Any) -> bool: + from graphistry.compute.filter_by_dict import _is_membership_filter_value + from graphistry.compute.predicates.ASTPredicate import ASTPredicate + + return ( + value is not None + and not isinstance(value, (ASTPredicate, dict)) + and not _is_membership_filter_value(value) + ) + + +def _simple_filter_dict(value: Any, *, allow_empty: bool = True) -> bool: + if value is None: + return allow_empty + return ( + isinstance(value, Mapping) + and (allow_empty or bool(value)) + and all(isinstance(key, str) and _plain_scalar_filter(item) + for key, item in value.items()) + ) + + +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): + return key_ok + return key_ok and getattr(other_dtype, "kind", None) in ("i", "u") + + +def _filter_compatible(frame: DataFrameT, filter_dict: Optional[dict]) -> bool: + """Cheap schema/type parity gate; decline so canonical filtering raises.""" + if not filter_dict: + return True + from graphistry.compute.filter_by_dict import ( + _is_numeric_dtype_safe, + _is_string_dtype_safe, + resolve_filter_column, + ) + + try: + for col, value in filter_dict.items(): + resolved, resolved_value = resolve_filter_column(frame, col, value) + series = ( + frame.get_column(resolved) # type: ignore[operator] + if "polars" in type(frame).__module__ + else frame[resolved] + ) + dtype = series.dtype + if _is_numeric_dtype_safe(dtype) and isinstance(resolved_value, str): + return False + if ( + _is_string_dtype_safe(dtype) + and isinstance(resolved_value, (int, float)) + and not isinstance(resolved_value, bool) + ): + return False + return True + except (AttributeError, KeyError, TypeError, ValueError): + return False + + +def _filter_frame( + frame: DataFrameT, filter_dict: Optional[dict], engine: Engine, +) -> DataFrameT: + if engine == Engine.POLARS: + from graphistry.compute.gfql.lazy.engine.polars.predicates import ( + filter_by_dict_polars, + ) + + return cast(DataFrameT, filter_by_dict_polars(frame, filter_dict)) # type: ignore[arg-type] + from graphistry.compute.filter_by_dict import filter_by_dict + + return filter_by_dict(frame, filter_dict, engine) # type: ignore[arg-type] + + +def _with_marker(frame: DataFrameT, name: Optional[str], engine: Engine) -> DataFrameT: + if not isinstance(name, str): + return frame + 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)) + + +def _frame_with_positions( + frame: DataFrameT, positions: Any, engine: Engine, +) -> DataFrameT: + if engine == Engine.POLARS: + import numpy as np + import polars as pl + + return cast( + 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) + + +def _orient_edges( + gathered: DataFrameT, + *, + src: str, + dst: str, + direction: str, + alias: Optional[str], + engine: Engine, +) -> DataFrameT: + payload = [ + col for col in gathered.columns + if col not in (src, dst, _EDGE_ORD) + ] + if engine == Engine.POLARS: + import polars as pl + + work = gathered + if isinstance(alias, str): + renames = {col: f"{alias}.{col}" for col in payload} + else: + work = work.select([src, dst, _EDGE_ORD]) # type: ignore[operator] + renames = {} + + def one(from_col: str, to_col: str, orient: int) -> DataFrameT: + out = work.rename({from_col: _FROM, to_col: _TO}) + if renames: + out = out.rename(renames) + return cast( + DataFrameT, + out.with_columns(pl.lit(orient).alias(_ORIENT_ORD)), # type: ignore[operator] + ) + + else: + work = gathered.copy() + if isinstance(alias, str): + work[alias] = True + payload = payload + [alias] + renames = {col: f"{alias}.{col}" for col in payload} + else: + renames = {} + + 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) + + 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) + 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, + 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(), + ) + + + +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, + ops: Sequence[Any], + *, + engine: Engine, + start_nodes: Optional[DataFrameT] = None, + alias_prefilters: Optional[Any] = None, +) -> Optional[IndexedBindingsState]: + """Return an exact indexed fixed-hop path bag, or safely decline with ``None``.""" + from graphistry.compute.ast import ASTEdge, ASTNode + + if ( + engine not in (Engine.PANDAS, Engine.CUDF, Engine.POLARS) + or start_nodes is not None + or alias_prefilters + or _policy_is_active() + or get_index_policy(base_graph) == "off" + or len(ops) < 3 + or len(ops) % 2 == 0 + ): + return None + + nodes = base_graph._nodes + edges = base_graph._edges + node_id = base_graph._node + src = base_graph._source + dst = base_graph._destination + if nodes is None or edges is None or node_id is None or src is None or dst is None: + return None + 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) + ] + if ( + len(aliases) != len(set(aliases)) + or _INTERNAL.intersection(set(map(str, nodes.columns))) + or _INTERNAL.intersection(set(map(str, edges.columns))) + or _INTERNAL.intersection(set(cast(Sequence[str], aliases))) + ): + return None + + for index, op in enumerate(ops): + if index % 2 == 0: + if ( + not isinstance(op, ASTNode) + or op.query is not None + or op._name == node_id + or not _simple_filter_dict(op.filter_dict) + ): + return None + else: + if ( + not isinstance(op, ASTEdge) + or not op.is_simple_single_hop() + or op._name in (src, dst) + or op.direction not in ("forward", "reverse", "undirected") + or not _simple_filter_dict(op.edge_match) + or any(value is not None for value in ( + op.source_node_match, op.destination_node_match, + op.source_node_query, op.destination_node_query, + op.edge_query, + )) + or op.prune_to_endpoints + or op.include_zero_hop_seed + or not _filter_compatible(edges, op.edge_match) + ): + return None + if any( + isinstance(op, ASTEdge) and op.direction == "undirected" for op in ops + ) and len(ops) != 3: + return None + + if engine in (Engine.PANDAS, Engine.CUDF): + unaliased_edges = sum( + isinstance(op, ASTEdge) and not isinstance(op._name, str) + for op in ops + ) + edge_payload = set(map(str, edges.columns)).difference((src, dst)) + if unaliased_edges >= 4 and edge_payload: + # Preserve the canonical pandas/cuDF fallback boundary: its fourth + # repeated anonymous edge payload can collide with accumulated + # merge suffixes. Aliased edges and payload-free frames are safe. + return None + + first_op = ops[0] + if ( + not isinstance(first_op, ASTNode) + or not _simple_filter_dict(first_op.filter_dict, allow_empty=False) + ): + return None + + if any( + isinstance(op, ASTNode) + and not _filter_compatible(nodes, op.filter_dict) + for op in ops[::2] + ): + return None + + registry = get_registry(base_graph) + node_index = cast( + Optional[NodeIdIndex], + registry.get_valid(NODE_ID, nodes, (node_id,), engine), + ) + if node_index is None or not _integer_index(node_index): + return None + + if engine == Engine.POLARS and ( + nodes.schema[node_id] != edges.schema[src] + or nodes.schema[node_id] != edges.schema[dst] + ): + return None + + direction_indexes: Dict[int, Sequence[AdjacencyIndex]] = {} + for edge_index in range(1, len(ops), 2): + edge_op = cast(ASTEdge, ops[edge_index]) + indexes = _indices_for_direction( + registry, edge_op.direction, edges, (src, dst), engine, + ) + if indexes is None or not all(_integer_index(index) for index in indexes): + return None + direction_indexes[edge_index] = indexes + + first_filter = cast(dict, first_op.filter_dict) + xp, _ = array_namespace(engine) + if node_id in first_filter: + if ( + not isinstance(first_filter[node_id], Integral) + or isinstance(first_filter[node_id], bool) + ): + return None + seed_ids = xp.asarray([first_filter[node_id]]) + seed_rows = xp.sort(lookup_node_rows(node_index, seed_ids, xp)) + 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) + + first_alias = first_op._name + alias_frames: Dict[str, DataFrameT] = {} + first_alias_frame = _with_marker(first_nodes, first_alias, engine) + if isinstance(first_alias, str): + alias_frames[first_alias] = first_alias_frame + + if engine == Engine.POLARS: + import polars as pl + + state = first_nodes.select(pl.col(node_id).alias(_CURRENT)) # type: ignore[operator] + 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}) + if isinstance(first_alias, str): + state[first_alias] = state[_CURRENT] + + estimated_rows = int(state.shape[0]) + policy = get_index_policy(base_graph) + n_edges = int(edges.shape[0]) + + for edge_index in range(1, len(ops), 2): + edge_op = cast(ASTEdge, ops[edge_index]) + frontier = xp.unique(col_to_array(state, _CURRENT, engine)) + edge_indexes = direction_indexes[edge_index] + if policy != "force": + threshold = cost_gate_frac(engine) * min( + index.n_keys for index in edge_indexes + ) + if int(frontier.shape[0]) >= threshold: + return None + gather_estimate = sum( + _lookup_degree(index, frontier, xp) for index in edge_indexes + ) + if ( + policy != "force" + and gather_estimate >= cost_gate_frac(engine) * n_edges + ): + return None + + row_parts = [ + lookup_edge_rows(index, frontier, xp)[0] for index in edge_indexes + ] + rows = ( + row_parts[0] if len(row_parts) == 1 + else xp.concatenate(row_parts) + ) + rows = xp.unique(rows) + gathered = take_rows(edges, rows, engine) + gathered = _frame_with_positions(gathered, rows, engine) + gathered = _filter_frame(gathered, edge_op.edge_match, engine) + oriented = _orient_edges( + gathered, + src=src, + dst=dst, + direction=edge_op.direction, + alias=edge_op._name, + engine=engine, + ) + + next_op = ops[edge_index + 1] + if not isinstance(next_op, ASTNode): + return None + endpoint_ids = xp.unique(col_to_array(oriented, _TO, engine)) + node_rows = xp.sort(lookup_node_rows(node_index, endpoint_ids, xp)) + 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, + ) + + estimated_rows = _estimate_join_rows(state, oriented, engine) + if policy != "force" and estimated_rows > 0 and estimated_rows >= n_edges: + return None + state = _join_state( + state, + oriented, + node_alias=next_op._name, + engine=engine, + ) + if isinstance(next_op._name, str): + alias_frames[next_op._name] = next_alias_frame + + return IndexedBindingsState( + state, alias_frames, engine, (len(ops) - 1) // 2, estimated_rows, + ) + + + + +def _connected_decline_reason( + base_graph: Plottable, + ops: Sequence[Any], + *, + engine: Engine, + start_nodes: Optional[DataFrameT], + alias_prefilters: Optional[Any], +) -> str: + """Classify a completed safe decline without changing canonical behavior.""" + from graphistry.compute.ast import ASTEdge + + if engine not in (Engine.PANDAS, Engine.CUDF, Engine.POLARS): + return "unsupported_engine" + if _policy_is_active(): + return "policy_active" + if get_index_policy(base_graph) == "off": + return "index_policy_off" + if start_nodes is not None or alias_prefilters: + return "unsupported_shape" + nodes, edges = base_graph._nodes, base_graph._edges + node_id, src, dst = base_graph._node, base_graph._source, base_graph._destination + if nodes is None or edges is None or node_id is None or src is None or dst is None: + return "unsupported_shape" + + registry = get_registry(base_graph) + required: List[Tuple[Any, Any, Tuple[str, ...]]] = [(NODE_ID, nodes, (str(node_id),))] + for op in ops[1::2]: + if not isinstance(op, ASTEdge): + return "unsupported_shape" + if op.direction in ("forward", "undirected"): + required.append((EDGE_OUT_ADJ, edges, (str(src), str(dst)))) + if op.direction in ("reverse", "undirected"): + required.append((EDGE_IN_ADJ, edges, (str(src), str(dst)))) + for kind, frame, columns in required: + raw = registry.get(kind) + if raw is None: + return "index_missing" + valid = registry.get_valid(kind, frame, columns, engine) + if valid is None: + return "index_stale" + 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" + return "unsupported_shape" + + +def try_indexed_connected_bindings_state( + base_graph: Plottable, + ops: Sequence[Any], + *, + engine: Engine, + start_nodes: Optional[DataFrameT] = None, + alias_prefilters: Optional[Any] = None, +) -> Optional[IndexedBindingsState]: + """Attempt the indexed path; only explicit safe declines fall through.""" + hop_count = max(0, (len(ops) - 1) // 2) + first_filter = getattr(ops[0], "filter_dict", None) if ops else None + node_id = base_graph._node + public_seed_scan = not ( + isinstance(first_filter, Mapping) + and node_id is not None + and str(node_id) in first_filter + ) + result = _try_indexed_connected_bindings_state( + base_graph, + ops, + engine=engine, + start_nodes=start_nodes, + alias_prefilters=alias_prefilters, + ) + if result is None: + # Classifying WHY we declined re-validates index fingerprints and can even + # filter the seed frame — diagnostic work whose only consumer is the trace. + # Compute it strictly inside a trace context (api._trace_active contract: + # diagnostic enrichment costs the hot path nothing). + reason = ( + _connected_decline_reason( + base_graph, + ops, + engine=engine, + start_nodes=start_nodes, + alias_prefilters=alias_prefilters, + ) + if _trace_active() + else "not_traced" + ) + _record_indexed_traversal( + seam="connected_bindings", + engine=engine, + served=False, + reason=reason, + hop_count=hop_count, + public_seed_scan=public_seed_scan, + ) + return None + _record_indexed_traversal( + seam="connected_bindings", + engine=engine, + served=True, + reason="served", + hop_count=result.hop_count, + public_seed_scan=public_seed_scan, + hop_details=[ + {"hop": hop + 1, "estimated_rows": result.estimated_rows} + for hop in range(result.hop_count) + ], + ) + return result diff --git a/graphistry/compute/gfql/index/handoff.py b/graphistry/compute/gfql/index/handoff.py new file mode 100644 index 0000000000..bce38a9a86 --- /dev/null +++ b/graphistry/compute/gfql/index/handoff.py @@ -0,0 +1,72 @@ +"""The chain-boundary -> row-materializer handoff for indexed fixed-hop bindings. + +The boundary decides ONCE whether the resident indexes can serve a fixed-hop +pattern, then the canonical row materializer consumes that decision instead of +re-deriving it. This module owns the whole contract — the dataclass, the single +attribute it rides on, and the only attribute access — so callers stay typed and +no other module hand-rolls ``getattr``/``setattr`` for it. + +The field is DECLARED on Plottable (with its default on PlotterBase) rather than +smuggled on with ``setattr``, so every access here is ordinary typed attribute +access. It is internal, always attached to an internal copy, and cleared before +the result is handed back to the caller. +""" +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple + +from graphistry.Engine import Engine +from graphistry.utils.json import JSONVal + +if TYPE_CHECKING: + from graphistry.Plottable import Plottable + from .bindings import IndexedBindingsState + + +@dataclass(frozen=True) +class IndexedBindingsHandoff: + """One boundary decision about one exact plan. + + ``state is None`` records a DECLINE, which is as important as a serve: without + it the row materializer would re-attempt the same plan (and re-record the same + trace decision) after the canonical traversal already ran. + """ + + binding_ops: List[Dict[str, JSONVal]] + state: Optional["IndexedBindingsState"] = None + edge_aliases: Tuple[str, ...] = field(default=()) + + def serves(self, binding_ops: List[Dict[str, JSONVal]], engine: Engine) -> bool: + """True when this handoff carries a usable state for exactly this plan.""" + return ( + self.state is not None + and self.state.engine == engine + and self.binding_ops == binding_ops + ) + + def declined(self, binding_ops: List[Dict[str, JSONVal]]) -> bool: + """True when this exact plan was already tried and safely declined.""" + return self.state is None and self.binding_ops == binding_ops + + +def attach_handoff(g: "Plottable", handoff: IndexedBindingsHandoff) -> "Plottable": + """Return an internal copy of ``g`` carrying ``handoff`` (never mutates ``g``).""" + out = g.bind() + out._gfql_indexed_bindings_handoff = handoff + return out + + +def set_handoff(g: Any, handoff: IndexedBindingsHandoff) -> None: + """Attach ``handoff`` to an object the caller is itself constructing.""" + g._gfql_indexed_bindings_handoff = handoff + + +def read_handoff(g: Any) -> Optional[IndexedBindingsHandoff]: + """The boundary decision riding on ``g``, if any.""" + return g._gfql_indexed_bindings_handoff + + +def clear_handoff(g: Any) -> None: + """Drop the handoff so it never escapes on a user-visible result.""" + g._gfql_indexed_bindings_handoff = None diff --git a/graphistry/compute/gfql/index/types.py b/graphistry/compute/gfql/index/types.py index ab4f7543aa..43a7fe722f 100644 --- a/graphistry/compute/gfql/index/types.py +++ b/graphistry/compute/gfql/index/types.py @@ -40,10 +40,17 @@ class IndexTraceStep(TypedDict, total=False): op: str + operation: str + seam: str direction: HopDirection hops: Optional[int] + hop_count: int policy: str engine: str + served: bool + reason: str + public_seed_scan: bool + hop_details: List[Dict[str, Any]] frontier_n: int path: IndexPath decision_reason: str diff --git a/graphistry/compute/gfql/lazy/engine/polars/chain.py b/graphistry/compute/gfql/lazy/engine/polars/chain.py index 8f3e79e383..dd5dbf42bd 100644 --- a/graphistry/compute/gfql/lazy/engine/polars/chain.py +++ b/graphistry/compute/gfql/lazy/engine/polars/chain.py @@ -7,7 +7,7 @@ (no silent pandas fallback). Deferred: variable-length/multi-hop edge sub-cases, some undirected multi-edge combos, node query=. """ -from typing import Any, List, Optional, Tuple, cast +from typing import TYPE_CHECKING, Any, List, Optional, Tuple, cast from typing_extensions import TypedDict @@ -17,6 +17,9 @@ from graphistry.Plottable import Plottable from graphistry.compute.ast import ASTObject, ASTNode, ASTEdge + +if TYPE_CHECKING: + from graphistry.compute.gfql.index.bindings import IndexedBindingsState from .hop_eager import ensure_nodes_polars from .dtypes import is_lazy, colnames, endpoint_ids from .degrees import get_degrees_polars, get_indegrees_polars, get_outdegrees_polars @@ -373,9 +376,8 @@ def _run_calls_polars(g_cur, calls, start_nodes, base_graph, middle): return g_cur if start_nodes is not None: - setattr(g_cur, "_gfql_start_nodes", start_nodes) - setattr(g_cur, "_gfql_rows_base_graph", base_graph) - setattr(g_cur, "_gfql_shortest_path_backend", getattr(g_cur, "_gfql_shortest_path_backend", "auto")) + g_cur._gfql_start_nodes = start_nodes + g_cur._gfql_rows_base_graph = base_graph if ( middle @@ -582,12 +584,86 @@ def chain_polars(self: Plottable, ops, start_nodes: Optional[Any] = None) -> Plo "use engine='pandas' for this chain." ) - g_cur = _chain_traversal_polars(self, middle, start_nodes) + from graphistry.compute.chain import serialize_binding_ops + from graphistry.compute.gfql.index.handoff import ( + IndexedBindingsHandoff, attach_handoff, + ) + + indexed_state, indexed_attempted = _try_indexed_middle_polars(self, middle, suffix, start_nodes) + if indexed_state is not None: + # Skip the canonical traversal: the compact indexed path bag already IS the + # binding rows the suffix asks for. Hand it to the unchanged native rows + # materializer through an internal graph copy. + g_cur = attach_handoff(self, IndexedBindingsHandoff( + binding_ops=serialize_binding_ops(middle), + state=indexed_state, + edge_aliases=tuple( + op._name for op in middle + if isinstance(op, ASTEdge) and isinstance(op._name, str) + ), + )) + else: + g_cur = _chain_traversal_polars(self, middle, start_nodes) + if indexed_attempted: + # Record the exact declined plan so the rows materializer does not + # re-attempt (and re-record) the same decision. Attach, never mutate: + # the pandas twin does the same, so neither boundary can write onto a + # graph it does not own. + g_cur = attach_handoff(g_cur, IndexedBindingsHandoff( + binding_ops=serialize_binding_ops(middle), + )) if suffix: g_cur = _run_calls_polars(g_cur, suffix, start_nodes, base_graph=self, middle=middle) return g_cur +def _try_indexed_middle_polars( + g: Plottable, + middle: List[ASTObject], + suffix: List[ASTObject], + start_nodes: Optional[Any], +) -> Tuple[Optional["IndexedBindingsState"], bool]: + """Attempt the indexed fixed-hop path BEFORE the canonical polars traversal. + + Returns ``(state_or_None, attempted)``. The shape gate mirrors the pandas + boundary gate and ``_run_calls_polars``' named-middle rewrite: the bypass is + only sound when the leading rows call consumes exactly this middle as binding + ops. Everything else (prefiltered/seeded/aliased-endpoint rows, unnamed middle + without binding ops, non-traversal middle) keeps the canonical path. + """ + from graphistry.compute.ast import ASTCall + from graphistry.compute.chain import serialize_binding_ops + + if ( + not middle + or not suffix + or start_nodes is not None + or not isinstance(suffix[0], ASTCall) + or suffix[0].function != "rows" + or suffix[0].params.get("source") is not None + or suffix[0].params.get("alias_endpoints") is not None + or suffix[0].params.get("alias_prefilters") + or not all(isinstance(op, (ASTNode, ASTEdge)) for op in middle) + ): + return None, False + binding_ops = suffix[0].params.get("binding_ops") + if not ( + binding_ops == serialize_binding_ops(middle) + or ( + binding_ops is None + and any(op._name is not None for op in middle) + ) + ): + return None, False + + from graphistry.Engine import Engine + from graphistry.compute.gfql.index.bindings import try_indexed_connected_bindings_state + from graphistry.compute.gfql.lazy import active_target, ExecutionTarget + + engine = Engine.POLARS_GPU if active_target() == ExecutionTarget.GPU else Engine.POLARS + return try_indexed_connected_bindings_state(g, middle, engine=engine), True + + def _chain_traversal_polars(self: Plottable, ops, start_nodes: Optional[Any] = None) -> Plottable: import polars as pl from graphistry.compute.chain import Chain diff --git a/graphistry/compute/gfql/lazy/engine/polars/pattern_apply.py b/graphistry/compute/gfql/lazy/engine/polars/pattern_apply.py index 2b2b7e36e4..1c0f0bbfaf 100644 --- a/graphistry/compute/gfql/lazy/engine/polars/pattern_apply.py +++ b/graphistry/compute/gfql/lazy/engine/polars/pattern_apply.py @@ -26,7 +26,7 @@ def _rows_base_graph(g: Plottable) -> Plottable: """The ORIGINAL graph threaded by the boundary lane (`_gfql_rows_base_graph` setattr convention, mirrored from the pandas lane at compute/chain.py) — the one sanctioned dynamic read, contained here; falls back to the current graph.""" - base = getattr(g, "_gfql_rows_base_graph", None) + base = g._gfql_rows_base_graph return base if base is not None else g @@ -60,14 +60,14 @@ def rows_binding_ops_polars(g: Plottable, binding_ops: Sequence[Dict[str, JSONVa nodes = base_graph._nodes if nodes is None or node_id not in nodes.columns: return None - start_nodes = getattr(g, "_gfql_start_nodes", None) + start_nodes = g._gfql_start_nodes if start_nodes is not None: # start-nodes constrain the scan like the pandas prev_node_wavefront does. # Normalize to EAGER: an eager.join(LazyFrame) raises and would silently # decline the whole wavefront-constrained case (wave-2 W2-3). sn = start_nodes.collect() if isinstance(start_nodes, pl.LazyFrame) else start_nodes try: - nodes = nodes.join(sn.select(node_id).unique(), on=node_id, how="semi") + nodes = nodes.join(sn.select(node_id).unique(), on=node_id, how="semi") # type: ignore[operator] # polars op on a DataFrameT-typed seed except Exception: return None try: diff --git a/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py b/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py index bd98c02379..277e570991 100644 --- a/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py +++ b/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py @@ -22,6 +22,12 @@ from graphistry.compute.gfql.expr_parser import ExprNode, FunctionCall from graphistry.compute.ast import ASTObject + # Within ONE call the path bag and its per-alias frames are the same polars + # flavour — the generic builder works in LazyFrames, the indexed one in eager + # frames. A constrained TypeVar says that; a plain union does not, and then + # `state.join(lookup)` cannot type-check. + PolarsFrameT = TypeVar("PolarsFrameT", "pl.DataFrame", "pl.LazyFrame") + from graphistry.Plottable import Plottable from graphistry.utils.json import JSONVal # Engine-neutral wire-format payload types (ASTCall.params). Shapes are safelist-validated @@ -636,6 +642,86 @@ def _rewrap(g: Plottable, table_df: Any) -> Plottable: return frame_ops.row_table(_RowPipelineAdapter(g), table_df) +def _finish_binding_rows_polars( + g: Plottable, + ops: "Sequence[ASTObject]", + state: "PolarsFrameT", + alias_frames: Dict[str, "PolarsFrameT"], + node_id: str, + attach_prop_aliases: Optional[Sequence[str]], + *, + decline_on_schema_error: bool, +) -> Optional[Plottable]: + """Canonical property attachment/materialization for generic or indexed state. + + ``decline_on_schema_error`` reflects WHOSE state this is. The generic builder + joins frames polars will not unify the way pandas implicitly does (int vs float + join keys), so a ``SchemaError`` there is an honest decline. The indexed state + comes from a helper that already checked those dtypes, so a ``SchemaError`` + there is a BUG and must surface rather than become a silent slow-path fallback. + """ + import polars as pl + from graphistry.compute.ast import ASTEdge, ASTNode + from graphistry.compute.gfql.lazy import collect as _lazy_collect + + def names(frame: "PolarsFrameT") -> List[str]: + return ( + frame.collect_schema().names() + if isinstance(frame, pl.LazyFrame) + else list(frame.columns) + ) + + try: + attach_set = ( + None if attach_prop_aliases is None else set(attach_prop_aliases) + ) + node_aliases: List[str] = [ + op._name + for op in ops[::2] + if isinstance(op, (ASTNode, ASTEdge)) and isinstance(op._name, str) + ] + for alias in node_aliases: + if attach_set is not None and alias not in attach_set: + continue + lookup_src = alias_frames[alias] + lookup = lookup_src.select( + [ + pl.col(node_id), + pl.col(node_id).alias(f"{alias}.{node_id}"), + ] + + [ + pl.col(col).alias(f"{alias}.{col}") + for col in names(lookup_src) + if col != node_id + ] + ) + if (set(names(lookup)) - {node_id}) & set(names(state)): + return None + state = state.join( + lookup, left_on=alias, right_on=node_id, how="left", + ) + state = state.drop("__current__") + out_df = ( + _lazy_collect(state) + if isinstance(state, pl.LazyFrame) + else state + ) + except pl.exceptions.SchemaError: + if not decline_on_schema_error: + raise + return None + + out = _rewrap(g, out_df) + edge_aliases = { + alias + for op in ops[1::2] + for alias in [op._name] + if isinstance(alias, str) + } + out._gfql_rows_edge_aliases = edge_aliases + return out + + _LowerT = TypeVar("_LowerT") @@ -1051,7 +1137,7 @@ def _names(lf: pl.LazyFrame) -> List[str]: if nodes is None or edges is None or node_id is None or src is None or dst is None: return None seed_ids_lf: Optional[Any] = None # LazyFrame; Any avoids the union-typed seed_nodes.join mismatch - start_nodes = getattr(g, "_gfql_start_nodes", None) + start_nodes = g._gfql_start_nodes if start_nodes is not None: # Bounded WITH->MATCH re-entry (#1273): the carried WITH rows seed the first # alias. Constrain the first node alias to the carried ids via a semi-join — @@ -1065,7 +1151,7 @@ def _names(lf: pl.LazyFrame) -> List[str]: sn = _df_to_engine(sn, _Engine.POLARS) if node_id not in sn.columns: return None - seed_ids = sn.select(pl.col(node_id)).drop_nulls() + seed_ids = sn.select(pl.col(node_id)).drop_nulls() # type: ignore[operator] # polars op on a DataFrameT-typed seed if seed_ids.height != seed_ids.unique().height: return None seed_ids_lf = seed_ids.lazy() @@ -1086,6 +1172,50 @@ def _names(lf: pl.LazyFrame) -> List[str]: if RowPipelineMixin._gfql_is_shortest_path_scalar_binding_ops(ops): return None # shortestPath scalar contract: BFS/native backends, pandas-only + from graphistry.compute.gfql.index import bindings as indexed_bindings + from graphistry.compute.gfql.lazy import active_target, ExecutionTarget + from graphistry.Engine import Engine + base_graph = g._gfql_rows_base_graph + if base_graph is None: + base_graph = g + engine_concrete = ( + Engine.POLARS_GPU + if active_target() == ExecutionTarget.GPU + else Engine.POLARS + ) + # The chain boundary may already have decided this exact plan (served or + # declined) before the canonical traversal; reuse that decision instead of + # recomputing it — and re-recording a duplicate trace step. + from graphistry.compute.gfql.index.handoff import read_handoff + + handoff = read_handoff(g) + plan = list(binding_ops) + indexed_state = ( + handoff.state + if handoff is not None and handoff.serves(plan, engine_concrete) + else None + ) + if indexed_state is None and not (handoff is not None and handoff.declined(plan)): + indexed_state = indexed_bindings.try_indexed_connected_bindings_state( + base_graph, + ops, + engine=engine_concrete, + start_nodes=start_nodes, + ) + if indexed_state is not None: + return _finish_binding_rows_polars( + g, + ops, + # engine-polymorphic by declaration; on the polars branch the helper + # builds polars frames, so narrow once here rather than widening the + # finisher's contract back to Any + cast("pl.DataFrame", indexed_state.state), + cast(Dict[str, "pl.DataFrame"], indexed_state.alias_frames), + str(node_id), + attach_prop_aliases, + decline_on_schema_error=False, # our own state: a schema clash is a bug + ) + for idx, op in enumerate(ops): if idx % 2 == 0: if not isinstance(op, ASTNode) or op.query is not None: @@ -1327,7 +1457,7 @@ def _names(lf: pl.LazyFrame) -> List[str]: and edge_op.direction == "forward" and not RowPipelineMixin._gfql_node_filter_has_label(next_op.filter_dict) ): - base_graph = getattr(g, "_gfql_rows_base_graph", None) + base_graph = g._gfql_rows_base_graph base_nodes = getattr(base_graph, "_nodes", None) if base_nodes is None: return None @@ -1347,47 +1477,18 @@ def _names(lf: pl.LazyFrame) -> List[str]: alias_frames[next_alias] = next_nodes node_aliases.append(next_alias) - # #1711 projection-pushdown: attach_prop_aliases (from the cypher lowering) - # names node aliases whose PROPERTIES are referenced downstream; others skip - # the property join (their bare id column suffices). None = attach all. - attach_set = None if attach_prop_aliases is None else set(attach_prop_aliases) - for alias in node_aliases: - if attach_set is not None and alias not in attach_set: - continue # properties unreferenced — keep only the bare id column - lookup_src = alias_frames[alias] - lookup = lookup_src.select( - [ - pl.col(node_id), - pl.col(node_id).alias(f"{alias}.{node_id}"), - ] - + [ - pl.col(col).alias(f"{alias}.{col}") - for col in _names(lookup_src) - if col != node_id - ] - ) - if (set(_names(lookup)) - {node_id}) & set(_names(state)): - return None - state = state.join(lookup, left_on=alias, right_on=node_id, how="left") - state = state.drop("__current__") - # Single collect on the active target (CPU / GPU). Deferred SchemaError - # (int/float join-key dtype divergence pandas unifies implicitly) surfaces - # here → decline honestly; a GPU-incapable node raises NotImplementedError - # (from the lazy `collect` NO-CHEATING contract), which we let propagate. - out_df = _lazy_collect(state) + # The finisher's frame type is a constrained TypeVar so `state.join(lookup)` + # type-checks. The GENERIC builder above mixes eager and lazy frames across + # its (pre-existing) branches, so inference cannot pick one here; the + # indexed caller binds cleanly. Narrow just this call rather than widening + # the finisher back to `Any`. + return _finish_binding_rows_polars( # type: ignore[misc] + g, ops, state, alias_frames, node_id, attach_prop_aliases, + decline_on_schema_error=True, # pandas-vs-polars join-key dtype divergence + ) except pl.exceptions.SchemaError: return None - out = _rewrap(g, out_df) - edge_aliases = { - alias - for op in ops[1::2] - for alias in [op._name] - if isinstance(alias, str) - } - setattr(out, "_gfql_rows_edge_aliases", edge_aliases) - return out - def can_select_native(items: Sequence[SelectItem], columns: Sequence[str]) -> bool: return lower_select_items(items, columns) is not None diff --git a/graphistry/compute/gfql/row/frame_ops.py b/graphistry/compute/gfql/row/frame_ops.py index e734793096..000433334d 100644 --- a/graphistry/compute/gfql/row/frame_ops.py +++ b/graphistry/compute/gfql/row/frame_ops.py @@ -1,6 +1,6 @@ from __future__ import annotations -from typing import TYPE_CHECKING, Any, Dict, List, Optional, Sequence, cast +from typing import TYPE_CHECKING, Any, Dict, Iterable, List, Optional, Sequence, cast import pandas as pd @@ -11,6 +11,7 @@ if TYPE_CHECKING: from typing import Protocol from graphistry.Plottable import Plottable + from graphistry.compute.typing import DataFrameT class RowPipelineCtx(Protocol): """Structural contract the row-pipeline frame ops need from their host graph. @@ -19,9 +20,15 @@ class RowPipelineCtx(Protocol): former ``ctx: Any`` so the attribute/method access is type-checked instead of duck-typed. Type-check-only (annotations are strings under ``from __future__ import annotations``) — zero runtime effect, no runtime Protocol import.""" - _nodes: Any - _edges: Any - _edge: Any + _nodes: Optional["DataFrameT"] + _edges: Optional["DataFrameT"] + _edge: Optional[str] + _node: Optional[str] + # Back-reference to the graph an adapter wraps; None on a graph that IS one. + _g: Optional["Plottable"] + _gfql_rows_base_graph: Optional["Plottable"] + _gfql_start_nodes: Optional["DataFrameT"] + _gfql_rows_edge_aliases: Optional[Iterable[str]] def bind(self) -> "Plottable": ... def _gfql_binding_ops_row_table( self, @@ -56,7 +63,11 @@ def _alias_true_mask(table_df: Any, source: str) -> Any: def row_table(ctx: RowPipelineCtx, table_df: Any) -> "Plottable": """Return a plottable that treats ``table_df`` as the active row table.""" + from graphistry.compute.gfql.index.handoff import clear_handoff, read_handoff + + handoff = read_handoff(ctx) out = ctx.bind() + clear_handoff(out) # internal plumbing must never escape on a user-visible result # polars has no row index, so reset_index is both unnecessary and absent. if not _is_polars(table_df): table_df = table_df.reset_index(drop=True) @@ -65,22 +76,48 @@ def row_table(ctx: RowPipelineCtx, table_df: Any) -> "Plottable": out._edges = _empty_like(ctx._edges) else: out._edges = _empty_like(table_df) + if handoff is not None and handoff.state is not None and out._edges is not None: + # The canonical traversal we skipped is what would have added these alias + # marker columns to the (zero-row) edge frame; synthesize them for parity. + indexed_edge_aliases = handoff.edge_aliases + if indexed_edge_aliases: + missing = [ + alias + for alias in indexed_edge_aliases + if alias not in out._edges.columns + ] + if _is_polars(out._edges): + # polars' canonical traversal APPENDS alias flag columns; pandas' + # puts them first. Match each engine's own canonical column order. + import polars as pl + + if missing: + out._edges = out._edges.with_columns( + [pl.lit(True).alias(alias) for alias in missing] + ) + else: + original_cols = [ + col + for col in out._edges.columns + if col not in indexed_edge_aliases + ] + out._edges = out._edges.assign( + **{alias: True for alias in missing} + )[list(indexed_edge_aliases) + original_cols] out._source = None out._destination = None out._edge = ctx._edge if ctx._edge is not None and ctx._edge in table_df.columns else None if out._node is not None and out._node not in table_df.columns: out._node = None - base_graph = getattr(ctx, "_gfql_rows_base_graph", None) + base_graph = ctx._gfql_rows_base_graph if base_graph is None: - base_graph = getattr(ctx, "_g", None) + base_graph = ctx._g # adapter-only back-reference if base_graph is not None: - setattr(out, "_gfql_rows_base_graph", base_graph) - start_nodes = getattr(ctx, "_gfql_start_nodes", None) - if start_nodes is not None: - setattr(out, "_gfql_start_nodes", start_nodes) - edge_aliases = getattr(ctx, "_gfql_rows_edge_aliases", None) - if edge_aliases is not None: - setattr(out, "_gfql_rows_edge_aliases", edge_aliases) + out._gfql_rows_base_graph = base_graph + if ctx._gfql_start_nodes is not None: + out._gfql_start_nodes = ctx._gfql_start_nodes + if ctx._gfql_rows_edge_aliases is not None: + out._gfql_rows_edge_aliases = ctx._gfql_rows_edge_aliases return cast("Plottable", out) @@ -95,13 +132,13 @@ def empty_frame( elif ctx._edges is not None: template_df = ctx._edges else: - base_graph = getattr(ctx, "_gfql_rows_base_graph", None) + base_graph = ctx._gfql_rows_base_graph if base_graph is None: - base_graph = getattr(ctx, "_g", None) + base_graph = ctx._g if base_graph is not None: - template_df = getattr(base_graph, "_nodes", None) + template_df = base_graph._nodes if template_df is None: - template_df = getattr(base_graph, "_edges", None) + template_df = base_graph._edges if template_df is not None: if columns is None: @@ -187,7 +224,7 @@ def rows( raise ValueError(f"rows(source=...) alias column not found: {source!r}") if _is_polars(table_df): import polars as pl - table_df = table_df.filter(pl.col(source).fill_null(False).cast(pl.Boolean)) + table_df = table_df.filter(pl.col(source).fill_null(False).cast(pl.Boolean)) # type: ignore[arg-type] else: table_df = table_df.loc[_alias_true_mask(table_df, source)] @@ -231,7 +268,7 @@ def count_table( import polars as pl if source is not None: # LazyFrame lacks .columns without a resolve; collect_schema is lazy-safe. - cols = table_df.collect_schema().names() + cols = table_df.collect_schema().names() # type: ignore[operator] if source not in cols: raise ValueError( f"count_table(source=...) alias column not found: {source!r}" @@ -239,7 +276,7 @@ def count_table( count_expr = pl.col(source).fill_null(False).cast(pl.Boolean).sum() else: count_expr = pl.len() - res = table_df.select(count_expr.alias(alias)) + res = table_df.select(count_expr.alias(alias)) # type: ignore[operator] # eager DataFrame.select -> DataFrame (no collect); LazyFrame.select -> LazyFrame. if hasattr(res, "collect"): res = res.collect() diff --git a/graphistry/compute/gfql/row/pipeline.py b/graphistry/compute/gfql/row/pipeline.py index 7e9a0d9064..5aec80b971 100644 --- a/graphistry/compute/gfql/row/pipeline.py +++ b/graphistry/compute/gfql/row/pipeline.py @@ -5,7 +5,7 @@ import warnings from functools import lru_cache from types import ModuleType -from typing import TYPE_CHECKING, Any, Callable, Dict, List, Mapping, NoReturn, Optional, Sequence, Tuple, cast +from typing import TYPE_CHECKING, Any, Callable, Dict, Iterable, List, Mapping, NoReturn, Optional, Sequence, Tuple, cast from typing_extensions import Literal import pandas as pd @@ -81,6 +81,9 @@ if TYPE_CHECKING: from graphistry.Plottable import Plottable + from graphistry.compute.typing import DataFrameT + from graphistry.compute.gfql.index.handoff import IndexedBindingsHandoff + from graphistry.compute.gfql.index.registry import GfqlIndexRegistry from graphistry.compute.ast import ASTObject from graphistry.compute.gfql.expr_parser import ExprNode @@ -196,10 +199,18 @@ def is_row_pipeline_call(function: str) -> bool: class RowPipelineMixin: - _g: Any - _gfql_start_nodes: Any - _gfql_rows_base_graph: Any - _gfql_rows_edge_aliases: Any + # Mirrors the GFQL execution-context fields declared on Plottable: this mixin + # is also used by `_RowPipelineAdapter`, which is not a PlotterBase, so the + # defaults must exist here too for typed access to be total. + _gfql_index_policy: str = "use" + _gfql_index_registry: Optional["GfqlIndexRegistry"] = None + _gfql_indexed_bindings_handoff: Optional["IndexedBindingsHandoff"] = None + _gfql_shortest_path_backend: str = "auto" + + _g: Optional["Plottable"] = None + _gfql_start_nodes: Optional["DataFrameT"] = None + _gfql_rows_base_graph: Optional["Plottable"] = None + _gfql_rows_edge_aliases: Optional[Iterable[str]] = None _nodes: Any _edges: Any _node: Any @@ -2759,7 +2770,7 @@ def _gfql_resolve_token(self, table_df: Any, token: str) -> Any: # identity column (alias.{node_id_col}). This lets expressions like # count(post) work when the table has post.id, post.name, etc. (#880) if "." not in txt and RowPipelineMixin._gfql_has_bindings_alias_prefix(table_df, txt): - edge_aliases = getattr(self, "_gfql_rows_edge_aliases", None) + edge_aliases = self._gfql_rows_edge_aliases if edge_aliases is not None and txt in edge_aliases: # Relationship aliases should render as entities (parity with # Cypher RETURN ) instead of collapsing to id-like @@ -3581,7 +3592,7 @@ def _gfql_connected_bindings_state( alias_prefilters: Optional[AliasPrefilters] = None, ) -> Tuple[DataFrameT, Dict[str, DataFrameT]]: from graphistry.compute.gfql.same_path.edge_semantics import EdgeSemantics - from graphistry.compute.ast import ASTEdge, ASTNode + from graphistry.compute.ast import ASTEdge, ASTNode, serialize_binding_ops if self._nodes is None or self._edges is None: return self._gfql_empty_frame(), {} @@ -3607,7 +3618,29 @@ def _gfql_connected_bindings_state( base_df = self._nodes if self._nodes is not None else self._edges engine = resolve_engine(EngineAbstract.AUTO, base_df) - start_nodes = getattr(self, "_gfql_start_nodes", None) + start_nodes = self._gfql_start_nodes + from graphistry.compute.gfql.index.handoff import read_handoff + + handoff = read_handoff(self) + binding_ops = serialize_binding_ops(ops) + if handoff is not None and handoff.serves(binding_ops, engine) and not alias_prefilters: + assert handoff.state is not None # narrowed by serves() + return handoff.state.state, handoff.state.alias_frames + previously_declined = handoff is not None and handoff.declined(binding_ops) + if ( + not previously_declined + and not RowPipelineMixin._gfql_is_shortest_path_scalar_binding_ops(ops) + ): + from graphistry.compute.gfql.index import bindings as indexed_bindings + indexed_state = indexed_bindings.try_indexed_connected_bindings_state( + base_graph, + ops, + engine=engine, + start_nodes=start_nodes, + alias_prefilters=alias_prefilters, + ) + if indexed_state is not None: + return indexed_state.state, indexed_state.alias_frames first_op = ops[0] if not isinstance(first_op, ASTNode): self._gfql_bindings_error( @@ -3839,7 +3872,7 @@ def _gfql_connected_bindings_row_table_from_ops( for alias in [getattr(op, "_name", None)] if isinstance(op, ASTEdge) and isinstance(alias, str) } - setattr(out, "_gfql_rows_edge_aliases", edge_aliases) + out._gfql_rows_edge_aliases = edge_aliases return out def _gfql_add_missing_binding_columns( @@ -4097,7 +4130,7 @@ def _gfql_shortest_path_scalar_bindings_row_table( return self._gfql_row_table(self._gfql_empty_frame()) # Try native igraph/cugraph backend first; fall back to BFS - sp_backend = getattr(self, "_gfql_shortest_path_backend", "auto") + sp_backend = self._gfql_shortest_path_backend reachable_hops = self._gfql_shortest_path_scalar_native( seed_table, ops, start_alias, end_alias, hop_column, backend=sp_backend ) @@ -4168,7 +4201,7 @@ def _gfql_cartesian_node_bindings_row_table( base_df = self._nodes if self._nodes is not None else self._edges engine = resolve_engine(EngineAbstract.AUTO, base_df) - start_nodes = getattr(self, "_gfql_start_nodes", None) + start_nodes = self._gfql_start_nodes join_col = RowPipelineMixin._gfql_fresh_col_name(base_nodes.columns, "__gfql_bindings_join__") bindings: Optional[Any] = None anonymous_cols: List[str] = [] @@ -5158,10 +5191,13 @@ class _RowPipelineAdapter(RowPipelineMixin): """Adapter for row-pipeline calls without requiring global ComputeMixin inheritance.""" def __init__(self, g: "Plottable") -> None: + from graphistry.compute.gfql.index.handoff import read_handoff + self._g = g - self._gfql_start_nodes = getattr(g, "_gfql_start_nodes", None) - self._gfql_rows_base_graph = getattr(g, "_gfql_rows_base_graph", None) - self._gfql_rows_edge_aliases = getattr(g, "_gfql_rows_edge_aliases", None) + self._gfql_start_nodes = g._gfql_start_nodes + self._gfql_rows_base_graph = g._gfql_rows_base_graph + self._gfql_rows_edge_aliases = g._gfql_rows_edge_aliases + self._gfql_indexed_bindings_handoff = read_handoff(g) self._nodes = g._nodes self._edges = g._edges self._node = g._node @@ -5170,6 +5206,7 @@ def __init__(self, g: "Plottable") -> None: self._edge = g._edge def bind(self) -> "Plottable": + assert self._g is not None # set by __init__; kept Optional for the protocol return self._g.bind() diff --git a/graphistry/compute/gfql_fast_paths.py b/graphistry/compute/gfql_fast_paths.py index 8475836e85..c5d5882752 100644 --- a/graphistry/compute/gfql_fast_paths.py +++ b/graphistry/compute/gfql_fast_paths.py @@ -2052,6 +2052,7 @@ def _execute_seeded_typed_hop_fast_path( return None ops = list(compiled_query.chain.chain) select_op: Optional[ASTCall] = None + suffix_ops: List[ASTCall] = [] if projection is not None: if len(ops) != 4: return None @@ -2059,9 +2060,16 @@ def _execute_seeded_typed_hop_fast_path( # property-RETURN lowering: [n0, e1, n2, rows(source=alias), select(items)] # (the LDBC IS5 shape: RETURN p.a AS x, p.b). Anything else (ORDER BY / # LIMIT / DISTINCT add further ops; exprs lower differently) falls back. - if len(ops) != 5 or not isinstance(ops[4], ASTCall) or ops[4].function != "select": + if len(ops) < 5 or not isinstance(ops[4], ASTCall) or ops[4].function != "select": return None select_op = ops[4] + suffix_ops = cast(List[ASTCall], ops[5:]) + if any( + not isinstance(op, ASTCall) + or op.function not in ("distinct", "order_by", "skip", "limit") + for op in suffix_ops + ): + return None ops = ops[:4] n0, e1, n2, call = ops if not (isinstance(n0, ASTNode) and isinstance(e1, ASTEdge) @@ -2118,6 +2126,7 @@ def _execute_seeded_typed_hop_fast_path( from graphistry.Engine import is_polars_df from graphistry.compute.chain_fast_paths import ( _seeded_typed_return_dst_pandas_cudf, _seeded_typed_return_dst_polars, + _resident_seed_indexes, ) nodes_frame = base_graph._nodes is_polars = is_polars_df(nodes_frame) @@ -2128,10 +2137,46 @@ def _execute_seeded_typed_hop_fast_path( # the full path CONVERTS the result to the requested engine, so the lean # projection would leak actual-engine frames. Decline; the full path decides. return None + from graphistry.compute.gfql.index import get_index_policy, get_registry + from graphistry.compute.gfql.index.api import _record_indexed_traversal + from graphistry.compute.gfql.index.registry import EDGE_IN_ADJ, EDGE_OUT_ADJ, NODE_ID + index_ctx = _resident_seed_indexes( + base_graph, base_graph._nodes, base_graph._edges, + node, src, dst, direction, + ) + if get_index_policy(base_graph) == "off": + index_reason = "index_policy_off" + elif index_ctx is not None: + index_reason = "served" + else: + registry = get_registry(base_graph) + adjacency_kind = EDGE_OUT_ADJ if direction == "forward" else EDGE_IN_ADJ + index_reason = ( + "index_missing" + if registry.get(NODE_ID) is None or registry.get(adjacency_kind) is None + else "index_stale" + ) helper = _seeded_typed_return_dst_polars if is_polars else _seeded_typed_return_dst_pandas_cudf dst_res = helper(base_graph, n0, n2, e1, src, dst, node, direction) if dst_res is None: + _record_indexed_traversal( + seam="destination_return", + engine=requested_engine, + served=False, + reason="unsupported_shape", + hop_count=1, + public_seed_scan=node not in cast(Dict[str, Any], n0.filter_dict), + ) return None + _record_indexed_traversal( + seam="destination_return", + engine=requested_engine, + served=index_ctx is not None, + reason=index_reason, + hop_count=1, + public_seed_scan=node not in cast(Dict[str, Any], n0.filter_dict), + hop_details=[{"hop": 1}] if index_ctx is not None else None, + ) p_rows, _edges = dst_res if select_items is not None: # Lean property projection (IS5 shape): the deduped destination rows carry @@ -2149,6 +2194,11 @@ def _execute_seeded_typed_hop_fast_path( # dtypes like nullable Int64/StringDtype, categoricals) declines to the # full path rather than risk a silent dtype divergence. import numpy as np + # The upcast above is a PANDAS pivot artifact. cuDF's rows-pivot keeps + # the source dtypes (verified: int64 stays int64, bool stays bool), so + # applying the pandas casts there would diverge from its own canonical + # path. The dtype-class decline guard still applies to both. + is_cudf_rows = "cudf" in type(p_rows).__module__ casts: Dict[str, str] = {} for out_name, prop in select_items: if prop == node: @@ -2159,9 +2209,11 @@ def _execute_seeded_typed_hop_fast_path( if not isinstance(d, np.dtype): return None if d == np.dtype(bool): - casts[out_name] = "object" + if not is_cudf_rows: + casts[out_name] = "object" elif d.kind in "iuf": - casts[out_name] = "float64" + if not is_cudf_rows: + casts[out_name] = "float64" elif d.kind != "O": return None out_frame = p_rows[[prop for _, prop in select_items]].copy() @@ -2173,7 +2225,15 @@ def _execute_seeded_typed_hop_fast_path( # full-path parity: a property RETURN yields an EMPTY edges frame with the # edge schema (never None — res._edges must stay usable). The helper's # edges are the matched hop edges, so take their zero-row head. - out._edges = _edges.head(0) + out._edges = _edges.head(0) if is_polars else _edges.head(0).reset_index(drop=True) + if suffix_ops: + return chain_impl( + out, + suffix_ops, + engine=engine, + policy=policy, + context=context, + ) return out assert projection is not None # narrowed by the gate above # Lean projection: p_rows already IS the RETURN-alias (destination) node set. diff --git a/graphistry/compute/gfql_unified.py b/graphistry/compute/gfql_unified.py index 5ece7cb218..1932c8b0ab 100644 --- a/graphistry/compute/gfql_unified.py +++ b/graphistry/compute/gfql_unified.py @@ -1120,7 +1120,7 @@ def _execute_compiled_query_chain_non_union( and getattr(_first_op, "function", None) == "rows" and getattr(_first_op, "params", {}).get("binding_ops") is not None ): - setattr(dispatch_graph, "_gfql_start_nodes", start_nodes) + dispatch_graph._gfql_start_nodes = start_nodes result = _chain_dispatch(dispatch_graph, compiled_query.chain, engine, policy, context, start_nodes=start_nodes) if compiled_query.empty_result_row is not None: @@ -1848,7 +1848,7 @@ def gfql(self: Plottable, raise dispatch_self = self - setattr(dispatch_self, "_gfql_shortest_path_backend", shortest_path_backend) + dispatch_self._gfql_shortest_path_backend = shortest_path_backend compiled_query = None if where_param and isinstance(query, (dict, ASTLet)): diff --git a/graphistry/tests/compute/gfql/index/test_indexed_bindings.py b/graphistry/tests/compute/gfql/index/test_indexed_bindings.py new file mode 100644 index 0000000000..acfbd28166 --- /dev/null +++ b/graphistry/tests/compute/gfql/index/test_indexed_bindings.py @@ -0,0 +1,839 @@ +"""Generic indexed fixed-hop GFQL differential contract. + +The shapes are reduced from existing LDBC-derived IS1/IS3/IS5/IS7 tests. Product +selection must remain operator/index/cost based; benchmark names live only in test +ids and comments. Every expected result comes from the same-engine canonical path. +""" +from __future__ import annotations + +from typing import Any, Dict, List, Sequence, Tuple, Type + +import numpy as np +import pandas as pd +import pytest + +import graphistry +from graphistry.Engine import Engine +from graphistry.compute.ast import ASTEdge, e_forward, e_undirected, n + + +ENGINES = ["pandas", "polars", "cudf"] +ENGINE_ENUM = { + "pandas": Engine.PANDAS, + "polars": Engine.POLARS, + "cudf": Engine.CUDF, +} +TRACE_KEYS = { + "operation", + "seam", + "engine", + "served", + "reason", + "hop_count", + "public_seed_scan", + "hop_details", + "path", + "decision_reason", +} + + +def _base_frames() -> Tuple[pd.DataFrame, pd.DataFrame]: + nodes = pd.DataFrame( + { + "id": np.arange(1, 13, dtype=np.int64), + "public": np.arange(100, 112, dtype=np.int64), + "kind": [ + "seed", "mid", "mid", "end", "end", "tail", + "tail", "reverse", "seed", "noise", "noise", "noise", + ], + "rank": np.arange(12, dtype=np.int64), + "maybe": [ + 0.0, 1.0, 2.0, None, 4.0, 5.0, + 6.0, 7.0, 8.0, 9.0, 10.0, 11.0, + ], + } + ) + edges = pd.DataFrame( + { + "src": [1, 1, 1, 2, 2, 3, 4, 5, 6, 8, 4, 5, 1, 1, 1, 2, 2], + "dst": [2, 2, 3, 4, 5, 5, 6, 6, 7, 1, 2, 3, 1, 2, 2, 1, 3], + "type": [ + "A", "A", "A", "B", "B", "B", "C", "C", "D", + "REV", "B", "B", "U", "U", "U", "U", "U", + ], + "weight": np.arange(10, 27, dtype=np.int64), + } + ) + return nodes, edges + + +def _native_frames( + engine: str, nodes: pd.DataFrame, edges: pd.DataFrame +) -> Tuple[Any, Any]: + if engine in ("polars", "polars-gpu"): + pl = pytest.importorskip("polars") + return pl.from_pandas(nodes), pl.from_pandas(edges) + if engine == "cudf": + cudf = pytest.importorskip("cudf") + return cudf.from_pandas(nodes), cudf.from_pandas(edges) + return nodes, edges + + +def _graph_from_frames( + engine: str, + nodes: pd.DataFrame, + edges: pd.DataFrame, + *, + node_col: str = "id", + source_col: str = "src", + destination_col: str = "dst", + indexed: bool = True, +) -> Any: + nodes_native, edges_native = _native_frames(engine, nodes, edges) + g = graphistry.nodes(nodes_native, node_col).edges( + edges_native, source_col, destination_col + ) + return g.gfql_index_all(engine=engine) if indexed else g + + +def _graph(engine: str, *, indexed: bool = True) -> Any: + return _graph_from_frames(engine, *_base_frames(), indexed=indexed) + + +def _native_copy(frame: Any) -> Any: + return frame.clone() if "polars" in type(frame).__module__ else frame.copy() + + +def _to_pandas(frame: Any) -> pd.DataFrame: + if "polars" in type(frame).__module__ or "cudf" in type(frame).__module__: + return frame.to_pandas() + return frame + + +def _assert_result_exact(actual: Any, expected: Any, engine: str) -> None: + assert actual._nodes is not None and expected._nodes is not None + if engine in ("polars", "polars-gpu"): + assert "polars" in type(actual._nodes).__module__ + assert actual._nodes.schema == expected._nodes.schema + elif engine == "cudf": + assert "cudf" in type(actual._nodes).__module__ + else: + assert isinstance(actual._nodes, pd.DataFrame) + pd.testing.assert_frame_equal( + _to_pandas(actual._nodes), + _to_pandas(expected._nodes), + check_dtype=True, + check_index_type=True, + ) + assert (actual._edges is None) == (expected._edges is None) + if actual._edges is not None and expected._edges is not None: + pd.testing.assert_frame_equal( + _to_pandas(actual._edges), + _to_pandas(expected._edges), + check_dtype=True, + check_index_type=True, + ) + + +def _run( + g: Any, + query: str, + engine: str, + monkeypatch: pytest.MonkeyPatch, + *, + generic: bool, + index_policy: str = "force", + policy: Any = None, +) -> Any: + if generic: + import graphistry.compute.gfql.index.bindings as indexed_bindings + import graphistry.compute.gfql_unified as gfql_unified + + monkeypatch.setattr( + indexed_bindings, + "try_indexed_connected_bindings_state", + lambda *args, **kwargs: None, + ) + monkeypatch.setattr( + gfql_unified, + "_execute_seeded_typed_hop_fast_path", + lambda *args, **kwargs: None, + ) + return g.gfql( + query, + engine=engine, + index_policy=index_policy, + policy=policy, + ) + + +def _trace_run( + g: Any, + query: str, + engine: str, + *, + index_policy: str = "force", + policy: Any = None, +) -> Tuple[Any, List[Dict[str, Any]]]: + from graphistry.compute.gfql.index import index_trace + + with index_trace() as captured: + out = g.gfql( + query, + engine=engine, + index_policy=index_policy, + policy=policy, + ) + return out, [ + dict(step) + for step in captured + if step.get("operation") == "indexed_traversal" + ] + + +def _assert_decision(step: Dict[str, Any], *, seam: str, served: bool) -> None: + assert TRACE_KEYS <= set(step) + assert step["seam"] == seam + assert step["served"] is served + assert step["path"] == ("index" if served else "scan") + assert step["decision_reason"] == step["reason"] + assert isinstance(step["hop_details"], list) + if not served: + assert step["hop_details"] == [] + + +def _assert_parity( + g: Any, + query: str, + engine: str, + monkeypatch: pytest.MonkeyPatch, + *, + seam: str, + index_policy: str = "force", + expect_served: bool = True, +) -> Tuple[Any, List[Dict[str, Any]]]: + actual, steps = _trace_run( + g, query, engine, index_policy=index_policy + ) + with monkeypatch.context() as m: + expected = _run( + g, + query, + engine, + m, + generic=True, + index_policy=index_policy, + ) + _assert_result_exact(actual, expected, engine) + decisions = [step for step in steps if step.get("seam") == seam] + assert decisions + assert any(step.get("served") is True for step in decisions) is expect_served + for step in decisions: + _assert_decision(step, seam=seam, served=bool(step["served"])) + return actual, decisions + + +DESTINATION_QUERY = ( + "MATCH (source {public:100})-[:A]->(destination) " + "RETURN destination.public AS destination ORDER BY destination" +) +CONNECTED_QUERY = ( + "MATCH (a {public:100})-[:A]->(b)-[:B]->(c) " + "RETURN a.public AS a, b.public AS b, c.public AS c ORDER BY b, c" +) + + +@pytest.mark.parametrize("engine", ENGINES) +@pytest.mark.parametrize( + "case,reason,served", + [ + pytest.param("resident", "served", True), + pytest.param("missing", "index_missing", False), + pytest.param("stale", "index_stale", False), + pytest.param("off", "index_policy_off", False), + ], +) +def test_destination_unique_trace_and_lifecycle( + engine: str, + case: str, + reason: str, + served: bool, + monkeypatch: pytest.MonkeyPatch, +) -> None: + g = _graph(engine, indexed=case != "missing") + if case == "stale": + assert g._edges is not None + g = g.edges(_native_copy(g._edges), "src", "dst") + elif case == "off": + setattr(g, "_gfql_index_policy", "off") + policy = "off" if case == "off" else "force" + actual, steps = _trace_run( + g, DESTINATION_QUERY, engine, index_policy=policy + ) + with monkeypatch.context() as m: + expected = _run( + g, + DESTINATION_QUERY, + engine, + m, + generic=True, + index_policy=policy, + ) + _assert_result_exact(actual, expected, engine) + decisions = [ + step for step in steps if step.get("seam") == "destination_return" + ] + assert len(decisions) == 1 + _assert_decision(decisions[0], seam="destination_return", served=served) + assert decisions[0]["reason"] == reason + assert decisions[0]["hop_count"] == 1 + assert decisions[0]["public_seed_scan"] is True + + +@pytest.mark.parametrize("engine", ENGINES) +@pytest.mark.parametrize( + "case,reason,served", + [ + pytest.param("resident", "served", True), + pytest.param("missing", "index_missing", False), + pytest.param("stale", "index_stale", False), + pytest.param("off", "index_policy_off", False), + ], +) +def test_connected_path_bag_trace_and_lifecycle( + engine: str, + case: str, + reason: str, + served: bool, + monkeypatch: pytest.MonkeyPatch, +) -> None: + g = _graph(engine, indexed=case != "missing") + if case == "stale": + assert g._nodes is not None + g = g.nodes(_native_copy(g._nodes), "id") + elif case == "off": + setattr(g, "_gfql_index_policy", "off") + policy = "off" if case == "off" else "force" + actual, steps = _trace_run(g, CONNECTED_QUERY, engine, index_policy=policy) + with monkeypatch.context() as m: + expected = _run( + g, + CONNECTED_QUERY, + engine, + m, + generic=True, + index_policy=policy, + ) + _assert_result_exact(actual, expected, engine) + decisions = [ + step for step in steps if step.get("seam") == "connected_bindings" + ] + assert len(decisions) == 1 + _assert_decision(decisions[0], seam="connected_bindings", served=served) + assert decisions[0]["reason"] == reason + assert decisions[0]["hop_count"] == 2 + assert decisions[0]["public_seed_scan"] is True + + +STANDARD_DERIVED_POSITIVES = [ + pytest.param( + "MATCH (a {public:100})-[r:A]->(b {kind:'mid'}) " + "RETURN a.public AS a, b.public AS b, r.weight AS w ORDER BY w", + id="is1-directed-projection", + ), + pytest.param( + "MATCH (a {public:100})-[r:U]-(b) " + "RETURN a.public AS a, b.public AS b, r.weight AS w ORDER BY w, b", + id="is3-undirected-multiplicity", + ), + pytest.param(CONNECTED_QUERY, id="is7-connected-two-hop-bag"), + pytest.param( + "MATCH (a {public:100})-[:A]->(b) " + "OPTIONAL MATCH (b)-[:B]->(c) " + "RETURN a.public AS a, b.public AS b, c.public AS c ORDER BY b, c", + id="is7-optional-continuation", + ), + pytest.param( + "MATCH (a {public:100})-[:A]->(b)<-[:B]-(c) " + "RETURN a.public AS a, b.public AS b, c.public AS c ORDER BY b, c", + id="fixed-hop-reverse-composition", + ), + pytest.param( + "MATCH (a {public:100})-[:A]->(b)-[:B]->(c) " + "RETURN DISTINCT b.public AS b, c.public AS c " + "ORDER BY b DESC, c DESC LIMIT 1", + id="canonical-distinct-order-limit-suffix", + ), + pytest.param( + "MATCH (a {public:999999})-[:A]->(b)-[:B]->(c) " + "RETURN a.public AS a, b.public AS b, c.public AS c", + id="official-no-match-stratum", + ), + pytest.param( + "MATCH (a {public:100})-[:A]->(b)-[:B]->(c) " + "RETURN b.public AS b, c.public AS c, c.maybe AS maybe ORDER BY b, c", + id="null-and-dtype-parity", + ), +] + + +@pytest.mark.parametrize("engine", ENGINES) +@pytest.mark.parametrize("query", STANDARD_DERIVED_POSITIVES) +def test_standard_derived_connected_parity( + engine: str, + query: str, + monkeypatch: pytest.MonkeyPatch, +) -> None: + _assert_parity( + _graph(engine), + query, + engine, + monkeypatch, + seam="connected_bindings", + ) + + +def test_pandas_connected_boundary_bypasses_canonical_traversal( + monkeypatch: pytest.MonkeyPatch, +) -> None: + g = _graph("pandas") + with monkeypatch.context() as m: + expected = _run( + g, + CONNECTED_QUERY, + "pandas", + m, + generic=True, + ) + + def unexpected_traversal(*args: Any, **kwargs: Any) -> Any: + raise AssertionError("indexed boundary entered canonical AST traversal") + + monkeypatch.setattr(ASTEdge, "execute", unexpected_traversal) + actual, steps = _trace_run(g, CONNECTED_QUERY, "pandas") + _assert_result_exact(actual, expected, "pandas") + decisions = [ + step for step in steps if step.get("seam") == "connected_bindings" + ] + assert len(decisions) == 1 + _assert_decision(decisions[0], seam="connected_bindings", served=True) + + +@pytest.mark.parametrize("engine", ENGINES) +def test_destination_property_projection_dtype_parity( + engine: str, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Lean property RETURN must match its OWN engine's canonical dtypes. + + The pandas rows-pivot upcasts int->float64 and bool->object; cuDF's keeps the + source dtypes. A single pandas-shaped cast rule silently diverges on cuDF. + """ + nodes, edges = _base_frames() + nodes = nodes.assign(flag=[True, False] * 6) + g = _graph_from_frames(engine, nodes, edges) + query = ( + "MATCH (source {public:100})-[:A]->(destination) " + "RETURN destination.rank AS r, destination.maybe AS m, " + "destination.flag AS f, destination.kind AS k, destination.id AS i " + "ORDER BY i" + ) + _assert_parity(g, query, engine, monkeypatch, seam="destination_return") + + +def test_polars_connected_boundary_bypasses_canonical_traversal( + monkeypatch: pytest.MonkeyPatch, +) -> None: + pytest.importorskip("polars") + from graphistry.compute.gfql.lazy.engine.polars import chain as polars_chain + + g = _graph("polars") + with monkeypatch.context() as m: + expected = _run( + g, + CONNECTED_QUERY, + "polars", + m, + generic=True, + ) + + def unexpected_traversal(*args: Any, **kwargs: Any) -> Any: + raise AssertionError("indexed boundary entered canonical polars traversal") + + monkeypatch.setattr( + polars_chain, "_chain_traversal_polars", unexpected_traversal + ) + actual, steps = _trace_run(g, CONNECTED_QUERY, "polars") + _assert_result_exact(actual, expected, "polars") + decisions = [ + step for step in steps if step.get("seam") == "connected_bindings" + ] + assert len(decisions) == 1 + _assert_decision(decisions[0], seam="connected_bindings", served=True) + + +@pytest.mark.parametrize("engine", ["pandas", "polars"]) +def test_unnamed_middle_rows_call_matches_canonical( + engine: str, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A rows() call with no binding ops reads the TRAVERSAL result, not the + whole graph: an indexed bypass must not engage for an unnamed middle.""" + from graphistry.compute.ast import rows as rows_call + + if engine == "polars": + pytest.importorskip("polars") + g = _graph(engine) + ops = [n({"id": 1}), e_forward(), n(), rows_call()] + with monkeypatch.context() as m: + expected = _run(g, ops, engine, m, generic=True) + actual = g.gfql(ops, engine=engine, index_policy="force") + _assert_result_exact(actual, expected, engine) + + +@pytest.mark.parametrize("seed_kind", ["seed", "noise"]) +def test_pandas_internal_id_plus_constraints_gathers_seed_before_filter( + seed_kind: str, + monkeypatch: pytest.MonkeyPatch, +) -> None: + import graphistry.compute.gfql.index.bindings as indexed_bindings + + g = _graph("pandas") + query = ( + f"MATCH (a {{id:1, kind:'{seed_kind}'}})-[:A]->(b)-[:B]->(c) " + "RETURN a.id AS a, b.id AS b, c.id AS c ORDER BY b, c" + ) + with monkeypatch.context() as m: + expected = _run(g, query, "pandas", m, generic=True) + + filtered_lengths: List[int] = [] + original_filter = indexed_bindings._filter_frame + + def record_filter(frame: Any, *args: Any, **kwargs: Any) -> Any: + filtered_lengths.append(len(frame)) + return original_filter(frame, *args, **kwargs) + + monkeypatch.setattr(indexed_bindings, "_filter_frame", record_filter) + actual, steps = _trace_run(g, query, "pandas") + + _assert_result_exact(actual, expected, "pandas") + decisions = [ + step for step in steps if step.get("seam") == "connected_bindings" + ] + assert len(decisions) == 1 + _assert_decision(decisions[0], seam="connected_bindings", served=True) + assert filtered_lengths + assert filtered_lengths[0] == 1 + + +@pytest.mark.parametrize("engine", ENGINES) +def test_indexed_execution_is_pure( + engine: str, + monkeypatch: pytest.MonkeyPatch, +) -> None: + g = _graph(engine) + assert g._nodes is not None and g._edges is not None + nodes_before = _to_pandas(g._nodes).copy(deep=True) + edges_before = _to_pandas(g._edges).copy(deep=True) + _assert_parity( + g, + CONNECTED_QUERY, + engine, + monkeypatch, + seam="connected_bindings", + ) + pd.testing.assert_frame_equal(_to_pandas(g._nodes), nodes_before) + pd.testing.assert_frame_equal(_to_pandas(g._edges), edges_before) + + +@pytest.mark.parametrize( + "ops,kwargs", + [ + pytest.param( + [ + n({"public": 100}, name="a"), + e_forward(hops=None, min_hops=2, max_hops=2), + n(name="b"), + ], + {}, + id="exact-bounded-varpath", + ), + pytest.param( + [ + n({"public": 100}, name="a"), + e_forward(include_zero_hop_seed=True), + n(name="b"), + ], + {}, + id="zero-hop", + ), + pytest.param( + [n({"public": 100}, name="a"), e_undirected(hops=2), n(name="b")], + {}, + id="multi-hop-undirected", + ), + pytest.param( + [n({"public": 100}, name="a"), e_forward(), n(name="a")], + {}, + id="alias-reentry-cycle", + ), + pytest.param( + [ + n({"public": 100}, name="a", query="rank >= 0"), + e_forward(), + n(name="b"), + ], + {}, + id="node-query", + ), + pytest.param( + [n({"public": 100}, name="a"), e_forward(edge_query="weight > 0"), n(name="b")], + {}, + id="edge-query", + ), + pytest.param( + [n({"public": 100}, name="a"), e_forward(), n(name="b")], + {"start_nodes": pd.DataFrame({"id": [1]})}, + id="carried-seed", + ), + pytest.param( + [n({"public": 100}, name="a"), e_forward(), n(name="b")], + {"alias_prefilters": {"b": {"rank": 1}}}, + id="alias-prefilter", + ), + ], +) +def test_unsupported_shapes_decline_before_work( + 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 + + with index_trace() as captured: + out = indexed_bindings.try_indexed_connected_bindings_state( + _graph("pandas"), + ops, + engine=Engine.PANDAS, + **kwargs, + ) + assert out is None + decisions = [ + dict(step) + for step in captured + if step.get("operation") == "indexed_traversal" + ] + assert len(decisions) == 1 + _assert_decision( + decisions[0], seam="connected_bindings", served=False + ) + assert decisions[0]["reason"] == "unsupported_shape" + + +@pytest.mark.parametrize( + "node_ids,edge_src,edge_dst,seed", + [ + pytest.param(["a", "b"], ["a"], ["b"], "a", id="string-ids"), + pytest.param([1.0, np.nan, 2.0], [1.0], [2.0], 1.0, id="null-float-ids"), + ], +) +def test_unsafe_id_domains_decline_structurally( + node_ids: List[Any], + edge_src: List[Any], + edge_dst: List[Any], + seed: Any, +) -> None: + import graphistry.compute.gfql.index.bindings as indexed_bindings + from graphistry.compute.gfql.index import index_trace + + g = graphistry.nodes(pd.DataFrame({"id": node_ids}), "id").edges( + pd.DataFrame({"src": edge_src, "dst": edge_dst}), "src", "dst" + ).gfql_index_all(engine="pandas") + ops = [n({"id": seed}, name="a"), e_forward(), n(name="b")] + with index_trace() as captured: + out = indexed_bindings.try_indexed_connected_bindings_state( + g, ops, engine=Engine.PANDAS + ) + assert out is None + decisions = [ + dict(step) + for step in captured + if step.get("operation") == "indexed_traversal" + ] + assert len(decisions) == 1 + _assert_decision(decisions[0], seam="connected_bindings", served=False) + assert decisions[0]["reason"] == "unsupported_dtype" + + +def test_shortest_path_never_enters_fixed_hop_helper() -> None: + from graphistry.compute.gfql.index import index_trace + + query = ( + "MATCH (a {id:1}), (b {id:5}), " + "path = shortestPath((a)-[*]-(b)) RETURN length(path) AS hops" + ) + with index_trace() as captured: + _graph("pandas").gfql(query, engine="pandas", index_policy="force") + assert not any( + step.get("operation") == "indexed_traversal" for step in captured + ) + + +@pytest.mark.parametrize("engine", ENGINES) +def test_policy_declines_without_skipping_hooks( + engine: str, + monkeypatch: pytest.MonkeyPatch, +) -> None: + fired: List[str] = [] + + def hook(ctx: Dict[str, Any]) -> None: + fired.append(str(ctx.get("phase", "?"))) + + policy = {"prechain": hook, "postchain": hook} + g = _graph(engine) + actual, steps = _trace_run( + g, CONNECTED_QUERY, engine, policy=policy + ) + actual_fired = list(fired) + fired.clear() + with monkeypatch.context() as m: + expected = _run( + g, + CONNECTED_QUERY, + engine, + m, + generic=True, + policy=policy, + ) + _assert_result_exact(actual, expected, engine) + assert actual_fired == fired + decisions = [ + step for step in steps if step.get("seam") == "connected_bindings" + ] + assert len(decisions) == 1 + _assert_decision(decisions[0], seam="connected_bindings", served=False) + assert decisions[0]["reason"] == "policy_active" + + +@pytest.mark.parametrize("engine", ENGINES) +def test_renamed_and_permuted_shape_remains_generic( + engine: str, + monkeypatch: pytest.MonkeyPatch, +) -> None: + nodes, edges = _base_frames() + nodes = ( + nodes.rename(columns={"id": "vertex", "public": "external"}) + .iloc[::-1] + .reset_index(drop=True) + ) + edges = ( + edges.rename(columns={"src": "source", "dst": "target"}) + .assign( + type=lambda frame: frame["type"].replace({"A": "X", "B": "Y"}) + ) + .iloc[np.roll(np.arange(len(edges)), 5)] + .reset_index(drop=True) + ) + g = _graph_from_frames( + engine, + nodes, + edges, + node_col="vertex", + source_col="source", + destination_col="target", + ) + query = ( + "MATCH (origin {external:100})-[first:X]->(middle)-[:Y]->(target) " + "RETURN origin.external AS origin, middle.external AS middle, " + "target.external AS target, first.weight AS weight " + "ORDER BY middle, target, weight" + ) + _assert_parity( + g, + query, + engine, + monkeypatch, + seam="connected_bindings", + ) + + +@pytest.mark.parametrize("engine", ENGINES) +@pytest.mark.parametrize("error_type", [RuntimeError, MemoryError]) +def test_unexpected_and_memory_errors_propagate( + engine: str, + error_type: Type[BaseException], + monkeypatch: pytest.MonkeyPatch, +) -> None: + import graphistry.compute.gfql.index.bindings as indexed_bindings + + def fail_filter(*args: Any, **kwargs: Any) -> Any: + raise error_type("sentinel") + + monkeypatch.setattr(indexed_bindings, "_filter_frame", fail_filter) + ops = [ + n({"public": 100}, name="source"), + e_forward({"type": "A"}), + n(name="destination"), + ] + with pytest.raises(error_type, match="sentinel"): + indexed_bindings.try_indexed_connected_bindings_state( + _graph(engine), + ops, + engine=ENGINE_ENUM[engine], + ) + + +def test_use_policy_sparse_serves_dense_declines( + monkeypatch: pytest.MonkeyPatch, +) -> None: + n_nodes = 512 + src = np.repeat(np.arange(n_nodes, dtype=np.int64), 2) + edges = pd.DataFrame( + { + "src": src, + "dst": (src + np.tile(np.array([1, 2]), n_nodes)) % n_nodes, + "type": ["X"] * len(src), + } + ) + query = ( + "MATCH (a {kind:'seed'})-[:X]->(b) " + "RETURN a.id AS a, b.id AS b ORDER BY a, b" + ) + for dense, expected_reason in [(False, "served"), (True, "cost_frontier")]: + kinds = ["seed"] * n_nodes if dense else ["seed"] + ["noise"] * (n_nodes - 1) + nodes = pd.DataFrame({"id": np.arange(n_nodes), "kind": kinds}) + g = _graph_from_frames("pandas", nodes, edges) + _, decisions = _assert_parity( + g, + query, + "pandas", + monkeypatch, + seam="connected_bindings", + index_policy="use", + expect_served=not dense, + ) + assert decisions[0]["reason"] == expected_reason + + +def test_explicit_polars_gpu_declines_indexed_helper_and_falls_back( + monkeypatch: pytest.MonkeyPatch, +) -> None: + import graphistry.compute.gfql.index.bindings as indexed_bindings + + g = _graph("polars-gpu") + ops = [ + n({"public": 100}, name="a"), + e_forward({"type": "A"}), + n(name="b"), + ] + assert indexed_bindings.try_indexed_connected_bindings_state( + g, ops, engine=Engine.POLARS_GPU + ) is None + _assert_parity( + g, + CONNECTED_QUERY, + "polars-gpu", + monkeypatch, + seam="connected_bindings", + expect_served=False, + ) diff --git a/graphistry/tests/compute/gfql/test_seeded_typed_hop_fastpath.py b/graphistry/tests/compute/gfql/test_seeded_typed_hop_fastpath.py index 1b880590d3..b57908b597 100644 --- a/graphistry/tests/compute/gfql/test_seeded_typed_hop_fastpath.py +++ b/graphistry/tests/compute/gfql/test_seeded_typed_hop_fastpath.py @@ -671,8 +671,6 @@ def test_is5_shape_engages_and_matches(self, engine): @pytest.mark.parametrize("q,label", [ ("MATCH (m:Message {id:10})-[{type:'HAS_CREATOR'}]->(p:Person) RETURN m.id AS mid, p.id AS pid", "cross-alias"), ("MATCH (m:Message {id:10})-[{type:'HAS_CREATOR'}]->(p:Person) RETURN p, p.age", "mixed whole+prop"), - ("MATCH (m:Message {id:10})-[{type:'HAS_CREATOR'}]->(p:Person) RETURN DISTINCT p.age", "distinct"), - ("MATCH (m:Message {id:10})-[{type:'HAS_CREATOR'}]->(p:Person) RETURN p.age ORDER BY p.age LIMIT 1", "order/limit"), ("MATCH (m:Message {id:10})-[{type:'HAS_CREATOR'}]->(p:Person) RETURN p.nosuch", "absent property"), ]) def test_out_of_shape_declines_with_parity(self, q, label): @@ -690,6 +688,17 @@ def test_out_of_shape_declines_with_parity(self, q, label): _run_diff(g, "pandas", q, fast=False) + @pytest.mark.parametrize("q,label", [ + ("MATCH (m:Message {id:10})-[{type:'HAS_CREATOR'}]->(p:Person) RETURN DISTINCT p.age", "distinct"), + ("MATCH (m:Message {id:10})-[{type:'HAS_CREATOR'}]->(p:Person) RETURN p.age ORDER BY p.age LIMIT 1", "order/limit"), + ]) + def test_canonical_projection_suffix_engages_with_parity(self, q, label): + """DISTINCT / ORDER BY / SKIP / LIMIT after the lean projection are plain + row-frame ops: the fast path now keeps the seeded gather and delegates the + suffix to the canonical chain, so these shapes engage AND stay exact.""" + self._diff(self._rich_graph(), "pandas", q, True) + + class TestSeededProjectionDtypeAndEdgesParity: """Review pins (plans/review-pr-1766/review.md): B1 int-property dtype parity (the full pandas path's rows-pivot upcasts non-id int/float props to float64