diff --git a/CHANGELOG.md b/CHANGELOG.md index 3e7165fbff..04d31f5bb1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,20 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm ## [Development] +### Changed +- **GFQL Cypher parse memoization (perf)**: `parse_cypher` now memoizes its result (LRU over the deterministic lark parse+transform → immutable frozen AST). Repeated identical Cypher queries skip the ~15 ms parse — the dominant per-call cost of small queries (~50% of a Cypher call at 100k rows) — making end-to-end query latency ~1.3–1.7× faster at small/interactive sizes across pandas/polars/cuDF. Safe to share the cached AST: every Cypher AST node is `@dataclass(frozen=True)` and `compile_cypher_query` does not mutate the parsed tree; validation errors still raise and are not cached. + +### Performance +- **GFQL temporal-detection dtype gate (#1650)**: `order_detect_temporal_mode` now short-circuits for numeric/bool/complex columns, which can never hold temporal *text*, instead of running an `astype(str)` + multi-regex `fullmatch` scan on every comparison. Eliminates spurious row-wise stringification in `where_rows`/comparison paths whose output never contains entity-text. Byte-identical results; measured `where_rows` speedups ~3.1× (pandas) and ~4.4–13.3× (cuDF, scaling with row count). Does not address whole-entity `RETURN a` text rendering, which is tracked separately. +- **GFQL generic single-hop fast path (perf, pandas + cuDF)**: a single `MATCH (n)` (node-only) or `MATCH (a {f})-[e]->(b)` (1-hop) — the dominant tabular/crossfilter + basic-graph-query shapes — now skip the forward/backward/combine BFS machinery in the generic engine: node-only returns the filtered node table; 1-hop returns the edges whose endpoints pass the node filters + those endpoint nodes. Same VALUES + node/edge sets as before (345-case adversarial golden: only dtype differs; hop/chain suites; gated to pandas/cuDF — dask/spark keep the full path). **~100× faster on pandas** (node filter 204→2 ms @10M; graph query similarly near-raw); cuDF stays on the resident frame (a couple semi-joins instead of the BFS + ~31 drop_duplicates), capturing the GPU semijoin win. **Minor behavior change:** the 1-hop now PRESERVES node-attribute dtypes (int stays int) instead of the full machinery's spurious int→float merge upcast — making pandas/cuDF consistent with the polars engine. The fast path is automatically skipped when a GFQL `policy` is installed, so the `prechain`/`postchain`/`postload` hooks (which can observe intermediate state and deny execution) always fire on the full path — keeping the optimization observationally transparent. Differential-verified equivalent to the full BFS path across 440 random well-formed graph×query cases plus targeted edge cases: it drops edges to nodes absent from the node table and dedups duplicate node ids (matching the full path's join semantics), and a NaN node id never validates a NaN edge endpoint. +- **GFQL hop/chain redundant-dedup removal (perf)**: dropped the explicit `.unique()` dedup pass that fed only an `.isin()` membership test in the generic traversal — `_filter_edges_by_endpoint`, the undirected combine masks, and the per-hop wavefront filter (`compute/hop.py`, `compute/chain.py`). `isin(s) == isin(s.unique())` by set membership, so this is byte-identical (verified across 345 adversarial graph×query cases: dup/parallel edges, self-loops, isolated/dead-end nodes, cycles, undirected, multi-hop, fixed-point, min/max hops, names, filters, seeds). Each removed `.unique()` is one fewer kernel launch on GPU, where launch latency — not compute — dominates small/mid traversals: cuDF 1-hop chain ~126→103 ms @1M edges (~18% faster), pandas unaffected within noise. + +### Fixed +- **GFQL single-hop fast path ignored `prune_to_endpoints`**: the generic single-hop fast path accepted edges with `prune_to_endpoints=True` (a public `e()/e_forward()/e_reverse()` kwarg) but returned BOTH endpoints, whereas the flag keeps only the arrival side (destinations for forward, sources for reverse). A query like `[n(), e_forward(prune_to_endpoints=True), n()]` silently returned the wrong node/edge set. The fast path now declines this shape and falls through to the full path, which honors the flag. +- **GFQL whole-entity `RETURN a, a.val` emitted a duplicate column**: flattening a whole entity `a` into `a.id, a.val, …` (#1650) shares the `{alias}.{field}` namespace with an explicit property projection, so `RETURN a, a.val` produced two `a.val` columns — a duplicate-named column that breaks column selection and silently drops data on `to_dict`/serialization. The duplicate (always identical data, since dotted backtick aliases are rejected) is now collapsed to a single column, keeping first occurrence. +- **GFQL chain on edges-only graphs (no node binding)**: A chain/Cypher query over a graph with edges but no node-id binding (`g._node is None`) that took the full traversal path (any fast-path-ineligible shape — multi-hop, named/queried/`prune_to_endpoints` edges — or any query with a policy attached) rebuilt the result from the *unbound* input graph, dropping the materialized node-id binding. The endpoint-reconciliation concat then synthesized a spurious `None`-named node column: a corrupt result on older pandas, and a hard `NotImplementedError` (void-block NA fill) on newer pandas (e.g. Python 3.14). The result now carries the materialized node-id binding and a single, correct node column, matching the fast-path output; the original edge binding (e.g. `None`) is still restored. +- **GFQL chain fast path bypassed policy hooks**: the degenerate-shape chain fast path (node-only `MATCH (n)`, single-hop) short-circuited before the `prechain`/`postchain`/`postload` policy hook dispatch, so a policy-bearing query that hit the fast path never fired those hooks (or per-op policy inspection). The fast path is now skipped whenever a policy is attached — `prechain` is a pre-compute gate that must observe/block every load — so policy-bearing queries take the full path; the no-policy path keeps the optimization. + ## [0.56.1 - 2026-05-27] ### Added diff --git a/graphistry/compute/chain.py b/graphistry/compute/chain.py index 837cdb3542..685395411f 100644 --- a/graphistry/compute/chain.py +++ b/graphistry/compute/chain.py @@ -28,11 +28,13 @@ logger = setup_logger(__name__) -def _filter_edges_by_endpoint(edges_df, nodes_df, node_id: str, edge_col: str): +def _filter_edges_by_endpoint( + edges_df: DataFrameT, nodes_df: Optional[DataFrameT], node_id: str, edge_col: str +) -> DataFrameT: if nodes_df is None or not node_id or not edge_col or edge_col not in edges_df.columns: return edges_df - ids = nodes_df[node_id].unique() - return edges_df[edges_df[edge_col].isin(ids)] + # isin() is set-membership, so the dropped .unique() is redundant (byte-identical). + return edges_df[edges_df[edge_col].isin(nodes_df[node_id])] class Chain(ASTSerializable): @@ -219,8 +221,9 @@ def combine_steps( direction = getattr(op, 'direction', 'forward') if isinstance(op, ASTEdge) else 'forward' if direction == 'undirected' and prev_nodes is not None and next_nodes is not None and node_id: - prev_ids = prev_nodes[node_id].unique() - next_ids = next_nodes[node_id].unique() + # isin() dedups internally -> the .unique() pass is redundant + prev_ids = prev_nodes[node_id] + next_ids = next_nodes[node_id] fwd_mask = edges_df[src_col].isin(prev_ids) & edges_df[dst_col].isin(next_ids) rev_mask = edges_df[dst_col].isin(prev_ids) & edges_df[src_col].isin(next_ids) edges_df = edges_df[fwd_mask | rev_mask] @@ -670,6 +673,91 @@ def _chain_otel_attrs( return attrs +def _try_chain_fast_path( + g_in: Plottable, + ops: List[ASTObject], + engine_concrete: Engine, + start_nodes: Optional[DataFrameT] = None, +) -> Optional[Plottable]: + """Degenerate-shape fast path (pandas/cuDF): node-only ``MATCH (n)`` or a plain + single-hop ``MATCH (a)-[e]->(b)`` skip the forward/backward/combine BFS machinery. + Returns the result Plottable, or ``None`` to fall through to the full path. + + Same node/edge sets + VALUES as the full machinery (trackA_golden + hop/chain + suites); the 1-hop additionally preserves int node dtypes (the full path upcasts + int→float via merge). Gated to unnamed/unqueried nodes + a plain single-hop edge; + filtered-undirected and seeded chains fall through. polars/dask/spark also fall + through (own fast path / lazy semantics).""" + from graphistry.compute.filter_by_dict import filter_by_dict + + if engine_concrete not in (Engine.PANDAS, Engine.CUDF): + return None + if start_nodes is not None: + return None # seeded chains use the full path (fast path has no seed) + engine_abs = EngineAbstract(engine_concrete.value) + + def _materialize_fast_path_graph() -> Plottable: + from graphistry.compute.ComputeMixin import _coerce_input_formats # lazy — avoids circular import + g = g_in.materialize_nodes(engine=EngineAbstract(engine_concrete.value)) + return _coerce_input_formats(g, engine_concrete) + + if len(ops) == 1: + n0 = ops[0] + if not (isinstance(n0, ASTNode) and n0._name is None and n0.query is None): + return None + g = _materialize_fast_path_graph() + if g._nodes is None: + return None + nodes = filter_by_dict(g._nodes, n0.filter_dict, engine_abs) if n0.filter_dict else g._nodes + edges = g._edges.iloc[0:0] if g._edges is not None else None + return g.nodes(nodes).edges(edges) if edges is not None else g.nodes(nodes) + + if len(ops) != 3: + return None + n0, e1, n2 = ops + if not (isinstance(n0, ASTNode) and n0._name is None and n0.query is None): + return None + if not (isinstance(n2, ASTNode) and n2._name is None and n2.query is None): + return None + if not (isinstance(e1, ASTEdge) and e1.is_simple_single_hop() + and e1.edge_match is None and e1.source_node_match is None + and e1.destination_node_match is None and e1._name is None + and e1.source_node_query is None and e1.destination_node_query is None + and e1.edge_query is None and not e1.include_zero_hop_seed + and not e1.prune_to_endpoints): # prune keeps only the arrival side -> full path + return None + direction = e1.direction + unconstrained = not n0.filter_dict and not n2.filter_dict + if not unconstrained and direction == "undirected": + return None # filtered-undirected (OR of both directions) -> full path + g = _materialize_fast_path_graph() + if g._nodes is None or g._edges is None: + return None + src, dst, node = g._source, g._destination, g._node + # Keep only edges with BOTH endpoints in the node table (the full path drops + # dangling edges via its joins). dropna so a NaN node id can't validate a NaN + # endpoint — .isin treats NaN as matchable but the BFS joins never match NaN<->NaN. + node_ids = g._nodes[node].dropna() + edges = g._edges[g._edges[src].isin(node_ids) & g._edges[dst].isin(node_ids)] + if not unconstrained: + from_col, to_col = (src, dst) if direction == "forward" else (dst, src) + if n0.filter_dict: + ids = filter_by_dict(g._nodes, n0.filter_dict, engine_abs)[node] + edges = edges[edges[from_col].isin(ids)] + if n2.filter_dict: + ids = filter_by_dict(g._nodes, n2.filter_dict, engine_abs)[node] + edges = edges[edges[to_col].isin(ids)] + concat = df_concat(engine_concrete) + endpoints = concat([ + edges[[src]].rename(columns={src: node}), + edges[[dst]].rename(columns={dst: node}), + ]).drop_duplicates() + nodes = g._nodes[g._nodes[node].isin(endpoints[node])] + # match the full path's merge, which collapses duplicate node-id rows + nodes = nodes.drop_duplicates(subset=[node]) + return g.nodes(nodes).edges(edges) + + @otel_traced("gfql.chain", attrs_fn=_chain_otel_attrs) def chain( self: Plottable, @@ -794,6 +882,13 @@ def _chain_impl( if validate_schema: validate_chain_schema(self, ops, collect_all=False) + # The fast path skips the policy hook dispatch (prechain/postchain below), so only + # take it when no policy is attached — policy-bearing queries must keep hook firing. + if not policy: + _fast = _try_chain_fast_path(self, ops, engine_concrete, start_nodes) + if _fast is not None: + return _fast + if isinstance(ops[0], ASTEdge): logger.debug('adding initial node to ensure initial link has needed reversals') ops = cast(List[ASTObject], [ ASTNode() ]) + ops @@ -1002,7 +1097,14 @@ def _chain_impl( ) if added_edge_index: final_edges_df = final_edges_df.drop(columns=[g._edge]) - g_out = self.nodes(final_nodes_df).edges(final_edges_df, edge=original_edge) + # Rebuild from `self` to restore the ORIGINAL edge binding (`self._edge`, + # often None — `g` carries the internal edge-index binding instead), but + # explicitly carry the materialized node-id binding `g._node`: for an + # edges-only input `self._node is None`, so rebuilding from `self` alone + # drops it, leaving the endpoint-reconciliation concat below to synthesize + # a `None`-named column (corrupt result + a void-block concat crash on + # newer pandas). + g_out = self.nodes(final_nodes_df, g._node).edges(final_edges_df, edge=original_edge) else: g_out = g.nodes(final_nodes_df).edges(final_edges_df) diff --git a/graphistry/compute/gfql/cypher/ast.py b/graphistry/compute/gfql/cypher/ast.py index 0cbf7a61a3..0b9524fe52 100644 --- a/graphistry/compute/gfql/cypher/ast.py +++ b/graphistry/compute/gfql/cypher/ast.py @@ -2,6 +2,8 @@ from typing import Literal, Optional, Tuple, Union +# INVARIANT: keep every node deeply immutable (scalar/tuple fields only) — parse_cypher +# shares results by reference via lru_cache, so a mutable field would poison cache hits. @dataclass(frozen=True) class SourceSpan: line: int diff --git a/graphistry/compute/gfql/cypher/parser.py b/graphistry/compute/gfql/cypher/parser.py index 8d22f1b36d..5f02788051 100644 --- a/graphistry/compute/gfql/cypher/parser.py +++ b/graphistry/compute/gfql/cypher/parser.py @@ -1857,7 +1857,19 @@ def parse_cypher(query: str) -> Union[CypherQuery, CypherUnionQuery, CypherGraph """ if not isinstance(query, str) or query.strip() == "": raise _to_syntax_error("Cypher query must be a non-empty string") + return _parse_cypher_cached(query) + +@lru_cache(maxsize=512) +def _parse_cypher_cached(query: str) -> Union[CypherQuery, CypherUnionQuery, CypherGraphQuery]: + """Cached parse body: pure function of the query text -> immutable frozen AST. + + Memoizing skips the ~15ms lark parse+transform on repeated identical queries + (the dominant per-call cost of small cypher queries). Safe to share the cached + result: every cypher AST node is ``@dataclass(frozen=True)`` and the downstream + ``compile_cypher_query`` does not mutate the parsed tree (verified). Validation + errors raise (and are not cached by ``lru_cache``), preserving error behavior. + """ # Pre-parse detection of known-but-unsupported Cypher forms _check_unsupported_syntax_patterns(query) diff --git a/graphistry/compute/gfql/cypher/result_postprocess.py b/graphistry/compute/gfql/cypher/result_postprocess.py index 263f1e33f8..d8b0103bc0 100644 --- a/graphistry/compute/gfql/cypher/result_postprocess.py +++ b/graphistry/compute/gfql/cypher/result_postprocess.py @@ -7,6 +7,7 @@ from graphistry.Plottable import Plottable from graphistry.compute.typing import DataFrameT, SeriesT +from graphistry.compute.gfql.series_str_compat import is_non_textual_scalar_dtype from .lowering import ResultProjectionColumn, ResultProjectionPlan from graphistry.compute.gfql.row.entity_props import ( @@ -123,6 +124,12 @@ def _project_property_column( if column.source_name is None or column.source_name not in rows_df.columns: raise ValueError(f"projection source column not found: {column.source_name!r}") series = cast(SeriesT, rows_df[column.source_name]) + # Temporal-constructor normalization only applies to STRING values; numeric/bool/ + # complex columns can never hold temporal text, so skip the (otherwise spurious) + # ``astype(str)`` + detection scan and return the column as-is — byte-identical, + # since the scan returns None for these dtypes. Mirrors the #1650/#1651 gate. + if is_non_textual_scalar_dtype(getattr(series, "dtype", None)): + return series if hasattr(series, "astype") and hasattr(cast(SeriesT, series.astype(str)), "str"): normalized = _normalize_temporal_constructor_series( rows_df, diff --git a/graphistry/compute/gfql/expr_parser.py b/graphistry/compute/gfql/expr_parser.py index f3459327d6..0dcd30c55d 100644 --- a/graphistry/compute/gfql/expr_parser.py +++ b/graphistry/compute/gfql/expr_parser.py @@ -21,6 +21,8 @@ DEFAULT_ALLOWED_QUANTIFIERS: FrozenSet[str] = GFQL_ALLOWED_QUANTIFIERS +# INVARIANT: keep every ExprNode deeply immutable (scalar/tuple fields only) — parse_expr +# shares results by reference via lru_cache, so a mutable field would poison cache hits. @dataclass(frozen=True) class Identifier: name: str diff --git a/graphistry/compute/gfql/row/ordering.py b/graphistry/compute/gfql/row/ordering.py index a2aafe4985..20a28d329a 100644 --- a/graphistry/compute/gfql/row/ordering.py +++ b/graphistry/compute/gfql/row/ordering.py @@ -12,6 +12,7 @@ from graphistry.compute.typing import SeriesT from graphistry.compute.gfql.series_str_compat import ( + is_non_textual_scalar_dtype, series_sequence_len, series_str_extract, series_str_fullmatch, @@ -115,9 +116,19 @@ def _actual_string_mask(series: Any, text: Any, null_mask: Any) -> Any: return mask +def _is_non_listable_dtype(series: Any) -> bool: + """True for numeric/bool/complex columns, which can never hold list values or + list-syntax *text*. Lets list/temporal detection skip the spurious + ``astype(str)`` + regex scan on these columns (byte-identical — the scan + returns False/None for them anyway). Mirrors the #1650/#1651 dtype gate.""" + return is_non_textual_scalar_dtype(getattr(series, "dtype", None)) + + def order_detect_list_series(series: Any) -> bool: if not hasattr(series, "isna") or not hasattr(series, "astype"): return False + if _is_non_listable_dtype(series): + return False null_mask = series.isna() non_null = ~null_mask if hasattr(non_null, "any") and not bool(non_null.any()): @@ -143,6 +154,8 @@ def order_detect_stringified_list_series(series: Any) -> bool: """ if not hasattr(series, "isna") or not hasattr(series, "astype"): return False + if _is_non_listable_dtype(series): + return False null_mask = series.isna() non_null = ~null_mask if hasattr(non_null, "any") and not bool(non_null.any()): @@ -214,6 +227,10 @@ def parse_stringified_list_series(series: Any) -> Optional[SeriesT]: def order_detect_temporal_mode(series: Any) -> Optional[str]: if not hasattr(series, "dropna"): return None + # Temporal values are text (Cypher date/time literals/constructors); numeric/bool/ + # complex can't match those regexes, so skip the astype(str)+scan for them (#1650). + if is_non_textual_scalar_dtype(getattr(series, "dtype", None)): + return None non_null = series.dropna() if len(non_null) == 0 or not hasattr(non_null, "astype"): return None diff --git a/graphistry/compute/gfql/row/pipeline.py b/graphistry/compute/gfql/row/pipeline.py index 59eff88c5b..8aa3a7a1ba 100644 --- a/graphistry/compute/gfql/row/pipeline.py +++ b/graphistry/compute/gfql/row/pipeline.py @@ -59,7 +59,7 @@ is_entity_text_scalar, ) from graphistry.compute.gfql.same_path_types import NODE_IDENTITY_COLUMN -from graphistry.compute.gfql.series_str_compat import series_sequence_len, series_str_match +from graphistry.compute.gfql.series_str_compat import is_non_textual_scalar_dtype, series_sequence_len, series_str_match from graphistry.compute.gfql.row.ordering import ( build_list_sort_columns, build_temporal_sort_columns, @@ -1992,6 +1992,10 @@ def _coerce_duration_property(value_text: str) -> Any: def _gfql_series_is_list_like(series: Any) -> bool: if not hasattr(series, "isna") or not hasattr(series, "astype"): return False + # numeric/bool/complex columns can never be list-like — skip the spurious + # astype(str)+regex scan (byte-identical; the scan returns False anyway). + if is_non_textual_scalar_dtype(getattr(series, "dtype", None)): + return False null_mask = series.isna() non_null = ~null_mask if hasattr(non_null, "any") and not bool(non_null.any()): diff --git a/graphistry/compute/gfql/series_str_compat.py b/graphistry/compute/gfql/series_str_compat.py index dc5ffbda26..56eebd58da 100644 --- a/graphistry/compute/gfql/series_str_compat.py +++ b/graphistry/compute/gfql/series_str_compat.py @@ -8,6 +8,24 @@ from graphistry.compute.typing import SeriesT +# Scalar number/bool dtypes — never list-like, never list/temporal text — so the +# detection scans (astype(str)+regex) always return False/None and callers can skip +# them. SSOT for the #1650/#1651 dtype gate; mirrored in +# filter_by_dict._is_numeric_dtype_safe (kept in sync by a test, not imported, to +# avoid an import cycle). +_NON_TEXTUAL_SCALAR_KINDS = frozenset({"i", "u", "f", "b", "c"}) + + +def is_non_textual_scalar_dtype(dtype: Any) -> bool: + """True for numeric/bool/complex dtypes (whose values can never be list-like or + list/temporal text), so list/temporal detection can skip the spurious + ``astype(str)`` + regex scan — byte-identical, since the scan returns + False/None for these dtypes anyway.""" + if dtype is None: + return False + return getattr(dtype, "kind", "O") in _NON_TEXTUAL_SCALAR_KINDS + + def _fill_string_mask_na(mask: Any, series: Any, na: Optional[bool]) -> Any: if na is None or not hasattr(mask, "where") or not hasattr(series, "isna"): return mask diff --git a/graphistry/compute/hop.py b/graphistry/compute/hop.py index 79c7c37bba..fe5a7637d2 100644 --- a/graphistry/compute/hop.py +++ b/graphistry/compute/hop.py @@ -472,7 +472,9 @@ def _build_pairs(src_col: str, dst_col: str) -> DataFrameT: wave_front_iter = wave_front_base[wave_front_base[node_col].isin(allowed_source_series)] first_iter = False - wavefront_ids = wave_front_iter[node_col].unique() + # isin() dedups internally; wavefront_ids feeds only isin -> skip the + # explicit .unique() dedup pass (a kernel launch on GPU). Byte-identical. + wavefront_ids = wave_front_iter[node_col] hop_edges = pairs[pairs[FROM_COL].isin(wavefront_ids)] if allowed_target_intermediate is not None: diff --git a/graphistry/tests/compute/gfql/cypher/test_parser.py b/graphistry/tests/compute/gfql/cypher/test_parser.py index db31baf01a..e8f45bf24b 100644 --- a/graphistry/tests/compute/gfql/cypher/test_parser.py +++ b/graphistry/tests/compute/gfql/cypher/test_parser.py @@ -1173,3 +1173,36 @@ def test_or_where_now_parses_after_earley_swap() -> None: assert parsed.where.expr_tree is not None assert parsed.where.expr_tree.op == "or" assert "OR" in boolean_expr_to_text(parsed.where.expr_tree).upper() + + +def test_parse_cypher_is_memoized_and_safe() -> None: + # parse_cypher memoizes its (deterministic) result: repeated identical + # queries return the SAME cached frozen AST (skips the ~15ms lark parse). + from graphistry.compute.gfql.cypher.parser import _parse_cypher_cached + + q = "MATCH (a)-[e]->(b) WHERE a.val > 50 RETURN a.val" + _parse_cypher_cached.cache_clear() + first = parse_cypher(q) + second = parse_cypher(q) + assert first is second # cache hit returns the identical object + + # Cached AST is immutable, so sharing it across callers is safe. + with pytest.raises(Exception): + first.where = None # type: ignore[misc] # frozen dataclass + + # Distinct query text is a distinct cache entry (no cross-contamination). + other = parse_cypher("MATCH (a) RETURN a.val") + assert other is not first + + # Clearing the cache and re-parsing yields an equal (value-identical) AST. + _parse_cypher_cached.cache_clear() + reparsed = parse_cypher(q) + assert reparsed == first + + +def test_parse_cypher_invalid_input_not_cached() -> None: + # Non-str / empty input must raise (and not poison the cache). + with pytest.raises(GFQLSyntaxError): + parse_cypher("") + with pytest.raises(GFQLSyntaxError): + parse_cypher(" ") diff --git a/graphistry/tests/compute/gfql/cypher/test_result_postprocess.py b/graphistry/tests/compute/gfql/cypher/test_result_postprocess.py index e36fa4ffcb..8d7fb6e61a 100644 --- a/graphistry/tests/compute/gfql/cypher/test_result_postprocess.py +++ b/graphistry/tests/compute/gfql/cypher/test_result_postprocess.py @@ -36,3 +36,33 @@ def test_apply_result_projection_preserves_prefixed_whole_row_metadata() -> None assert out._cypher_entity_projection_meta["node"]["alias"] == "n" assert out._cypher_entity_projection_meta["node"]["id_column"] == "id" assert out._cypher_entity_projection_meta["node"]["ids"].tolist() == ["a"] + + +def test_project_property_column_numeric_passes_through_unchanged() -> None: + # Numeric/bool columns must pass through with dtype + values intact — the + # temporal-constructor scan can never match them, so the gate skips the + # spurious astype(str) and returns the column as-is (byte-identical). + from graphistry.compute.gfql.cypher.result_postprocess import _project_property_column + + df = pd.DataFrame( + { + "x": pd.Series([3, 1, 2], dtype="int64"), + "f": pd.Series([2.0, 1.5, 3.0], dtype="float64"), + "b": pd.Series([True, False, True]), + } + ) + for col, dtype in [("x", "int64"), ("f", "float64"), ("b", "bool")]: + out = _project_property_column(df, column=ResultProjectionColumn(col, "property", col)) + assert str(out.dtype) == dtype, f"{col}: dtype changed to {out.dtype}" + assert out.tolist() == df[col].tolist() + + +def test_numeric_return_order_by_is_numeric_not_lexical() -> None: + # Behavioral guard: a numeric ORDER BY must sort numerically. If the projected + # column were stringified, a lexical sort of [2,10,1] would give [1,10,2]. + nd = pd.DataFrame({"id": [0, 1, 2, 3], "val": [2, 10, 1, 30]}) + ed = pd.DataFrame({"s": [0, 1, 2], "d": [1, 2, 3]}) + g = graphistry.nodes(nd, "id").edges(ed, "s", "d") + out = g.gfql("MATCH (a) RETURN a.val ORDER BY a.val", engine="pandas")._nodes + vals = out[out.columns[0]].tolist() + assert vals == [1, 2, 10, 30] diff --git a/graphistry/tests/compute/gfql/row/test_ordering.py b/graphistry/tests/compute/gfql/row/test_ordering.py index c8c22ef75a..38e222376d 100644 --- a/graphistry/tests/compute/gfql/row/test_ordering.py +++ b/graphistry/tests/compute/gfql/row/test_ordering.py @@ -5,6 +5,7 @@ from graphistry.compute.ast import limit, order_by, rows, select from graphistry.compute.exceptions import GFQLTypeError +from graphistry.compute.gfql.row.ordering import order_detect_temporal_mode from graphistry.tests.test_compute import CGFull @@ -193,3 +194,119 @@ def test_row_pipeline_order_by_multi_key_stringified_list_with_scalar() -> None: assert result["num"].tolist() == [20, 10, 15, 5] leaked = [c for c in result.columns if "__gfql_sort_listparsed" in str(c)] assert leaked == [], f"aux columns leaked: {leaked}" + + +@pytest.mark.parametrize( + "series", + [ + pd.Series([1, 2, 3], dtype="int64"), + pd.Series([1, 2, 3], dtype="uint32"), + pd.Series([1.5, 2.5], dtype="float64"), + pd.Series([True, False], dtype="bool"), + ], + ids=["int64", "uint32", "float64", "bool"], +) +def test_order_detect_temporal_mode_skips_non_text_dtypes(series: pd.Series) -> None: + # Numeric/bool columns can never hold temporal *text*; the detector must + # short-circuit without the astype(str) + multi-regex scan (issue #1650). + assert order_detect_temporal_mode(series) is None + + +def test_order_detect_temporal_mode_still_detects_text_temporals() -> None: + # Gate must not regress detection on object/string columns. + assert order_detect_temporal_mode(pd.Series(["2020-01-01", "2020-02-02"], dtype="object")) == "date" + assert order_detect_temporal_mode(pd.Series(["abc", "def"], dtype="object")) is None + + +def test_where_rows_numeric_filter_returns_correct_rows() -> None: + # End-to-end: a numeric where_rows comparison still filters correctly with the + # temporal-detection gate in place. + nodes_df = pd.DataFrame({"id": [0, 1, 2, 3], "val": [10, 60, 51, 99]}) + edges_df = pd.DataFrame({"s": [0, 1], "d": [2, 3]}) + from graphistry.compute.ast import where_rows + + out = _mk_graph(nodes_df, edges_df).gfql([rows(), where_rows(expr="val > 50")])._nodes + assert sorted(out["val"].tolist()) == [51, 60, 99] + + +@pytest.mark.skipif("TEST_CUDF" not in __import__("os").environ, reason="cuDF lane: set TEST_CUDF=1 (e.g. dgx-spark)") +def test_where_rows_numeric_filter_returns_correct_rows_cudf() -> None: + # cuDF lane for the numeric dtype-gate end-to-end path: the gate is pure + # dtype.kind inspection (engine-agnostic), but compute/gfql/row needs paired + # cuDF coverage per the review conventions. + cudf = pytest.importorskip("cudf") + from graphistry.compute.ast import where_rows + nodes_df = cudf.DataFrame({"id": [0, 1, 2, 3], "val": [10, 60, 51, 99]}) + edges_df = cudf.DataFrame({"s": [0, 1], "d": [2, 3]}) + out = CGFull().nodes(nodes_df, "id").edges(edges_df, "s", "d").gfql( + [rows(), where_rows(expr="val > 50")])._nodes + assert sorted(out["val"].to_pandas().tolist()) == [51, 60, 99] + + +@pytest.mark.parametrize( + "series", + [ + pd.Series([1, 2, 3], dtype="int64"), + pd.Series([1, 2, 3], dtype="uint32"), + pd.Series([1.0, 2.0], dtype="float64"), + pd.Series([True, False], dtype="bool"), + ], + ids=["int64", "uint32", "float64", "bool"], +) +def test_gfql_series_is_list_like_skips_non_listable_dtypes(series: pd.Series) -> None: + # pipeline.py sibling of the ordering gate: numeric/bool columns can never be + # list-like, so the detector short-circuits before the astype(str)+regex scan. + from graphistry.compute.gfql.row.pipeline import RowPipelineMixin + assert RowPipelineMixin._gfql_series_is_list_like(series) is False + + +def test_gfql_series_is_list_like_still_detects_real_lists() -> None: + # Gate must not regress detection of real list/tuple-valued object columns. + # (Stringified lists like "[1, 2]" are intentionally NOT list-like here — that + # case is handled by order_detect_stringified_list_series, not this helper.) + from graphistry.compute.gfql.row.pipeline import RowPipelineMixin + assert RowPipelineMixin._gfql_series_is_list_like(pd.Series([[1, 2], [3, 4]], dtype="object")) is True + assert RowPipelineMixin._gfql_series_is_list_like(pd.Series(["[1, 2]", "[3, 4]"], dtype="object")) is False + assert RowPipelineMixin._gfql_series_is_list_like(pd.Series(["abc", "def"], dtype="object")) is False + + +@pytest.mark.parametrize( + "dtype_str,expected", + [ + ("int64", True), ("uint32", True), ("int8", True), ("float32", True), + ("float64", True), ("bool", True), ("complex128", True), + ("object", False), ("datetime64[ns]", False), ("timedelta64[ns]", False), + (" None: + # Single source of truth for the #1650/#1651 dtype gate (shared by ordering, + # pipeline, and cypher result post-processing). + import numpy as np + from graphistry.compute.gfql.series_str_compat import is_non_textual_scalar_dtype + assert is_non_textual_scalar_dtype(np.dtype(dtype_str)) is expected + + +def test_is_non_textual_scalar_dtype_none() -> None: + from graphistry.compute.gfql.series_str_compat import is_non_textual_scalar_dtype + assert is_non_textual_scalar_dtype(None) is False + + +def test_dtype_gate_kind_set_matches_filter_by_dict_mirror() -> None: + """Drift guard: `filter_by_dict._is_numeric_dtype_safe` keeps a SEPARATE copy of + the dtype-kind set that `series_str_compat._NON_TEXTUAL_SCALAR_KINDS` is the SSOT + for (a cross-layer import would risk an import cycle, so the copy stays — but it + MUST NOT drift). Assert the two agree across every numpy dtype kind, so a future + edit to one set fails here instead of silently desyncing the gate.""" + import numpy as np + from graphistry.compute.gfql.series_str_compat import is_non_textual_scalar_dtype + from graphistry.compute.filter_by_dict import _is_numeric_dtype_safe + # One representative dtype per numpy kind, incl. text/temporal/extension kinds. + samples = [ + np.dtype("int64"), np.dtype("uint8"), np.dtype("float64"), np.dtype("bool"), + np.dtype("complex128"), np.dtype("object"), np.dtype("datetime64[ns]"), + np.dtype("timedelta64[ns]"), np.dtype(" full (non-fast) path + + +def _cudf_or_skip(): + if not (os.environ.get("TEST_CUDF") == "1"): + pytest.skip("cuDF lane: set TEST_CUDF=1 (e.g. on dgx-spark)") + return pytest.importorskip("cudf") + + +def _fast_graph(engine): + nodes = pd.DataFrame({'v': [0, 1, 2, 3, 4], 'attr': [10, 20, 30, 40, 50]}) + edges = pd.DataFrame({'s': [0, 1, 2, 3, 0], 'd': [1, 2, 3, 4, 2], 'w': [5, 6, 7, 8, 9]}) + if engine == "cudf": + cudf = _cudf_or_skip() + nodes = cudf.DataFrame.from_pandas(nodes) + edges = cudf.DataFrame.from_pandas(edges) + return CGFull().nodes(nodes, 'v').edges(edges, 's', 'd') + + +def _setsig(r): + """Engine-agnostic (node-id set, edge (s,d) set) — values, not dtypes.""" + def topd(df): + return df.to_pandas() if df is not None and "cudf" in type(df).__module__ else df + nn = topd(r._nodes) + ee = topd(r._edges) + nodes = sorted(nn['v'].tolist()) if nn is not None else [] + edges = sorted(map(tuple, ee[['s', 'd']].itertuples(index=False, name=None))) \ + if ee is not None and len(ee) else [] + return nodes, edges + + +# shapes that ARE accelerated by the fast path +_FAST_SHAPES = [ + ("node_only", lambda: [n()]), + ("node_filter", lambda: [n({'attr': 20})]), + ("node_pred", lambda: [n({'attr': is_in([10, 30])})]), + ("hop_fwd", lambda: [n(), e_forward(hops=1), n()]), + ("hop_rev", lambda: [n(), e_reverse(hops=1), n()]), + ("hop_undirected_unconstrained", lambda: [n(), e_undirected(hops=1), n()]), + ("hop_fwd_src_filter", lambda: [n({'attr': 10}), e_forward(hops=1), n()]), + ("hop_fwd_dst_filter", lambda: [n(), e_forward(hops=1), n({'attr': 40})]), + ("hop_fwd_both_filter", lambda: [n({'attr': 10}), e_forward(hops=1), n({'attr': 30})]), + ("hop_rev_dst_filter", lambda: [n(), e_reverse(hops=1), n({'attr': 10})]), +] + +# shapes that BYPASS the fast path (still must be correct via the full path) +_BYPASS_SHAPES = [ + ("hops_2", lambda: [n(), e_forward(hops=2), n()]), + ("filtered_undirected", lambda: [n({'attr': 10}), e_undirected(hops=1), n({'attr': 30})]), + ("edge_match", lambda: [n(), e_forward(hops=1, edge_match={'w': 5}), n()]), + ("named_node", lambda: [n(name='x'), e_forward(hops=1), n()]), + # prune_to_endpoints: fast path returns both endpoints; full path keeps only the + # arrival side. Must bypass the fast path (regression guard for the prune gate). + ("prune_endpoints_fwd", lambda: [n(), e_forward(hops=1, prune_to_endpoints=True), n()]), + ("prune_endpoints_rev", lambda: [n(), e_reverse(hops=1, prune_to_endpoints=True), n()]), +] + + +@pytest.mark.parametrize("engine", ["pandas", "cudf"]) +@pytest.mark.parametrize("label,build", _FAST_SHAPES + _BYPASS_SHAPES, + ids=[s[0] for s in _FAST_SHAPES + _BYPASS_SHAPES]) +def test_fast_path_differential_parity_vs_full_path(engine, label, build): + """Fast path output == full (policy-forced BFS) path output, by node/edge SET, + for every accelerated shape AND every bypass shape, on pandas and cuDF. + + For FAST shapes `g.gfql(ops)` exercises the fast path and the policy-forced call + the full BFS, so this is a true differential. For BYPASS shapes both calls take + the full path (the point being they MUST decline the fast path); the decline is + asserted directly below so the bypass cases are not merely full-vs-full.""" + from graphistry.compute.chain import _try_chain_fast_path + from graphistry.Engine import Engine + g = _fast_graph(engine) + fast = g.gfql(build()) + full = g.gfql(build(), policy=_FAST_NOOP_POLICY) + assert _setsig(fast) == _setsig(full), f"{engine}/{label}: fast != full" + # Bypass shapes must genuinely decline the fast path (not vacuously full==full). + if engine == "pandas" and any(label == s[0] for s in _BYPASS_SHAPES): + eng = Engine.PANDAS + assert _try_chain_fast_path(g, build(), eng, None) is None, \ + f"{label}: bypass shape must decline the fast path" + + +@pytest.mark.parametrize("engine", ["pandas", "cudf"]) +def test_fast_path_preserves_int_node_dtypes(engine): + """Documented behavior change: the 1-hop fast path PRESERVES node-attribute + dtypes (int stays int) where the full BFS path upcasts int->float via merge. + Lock both sides so neither silently regresses.""" + g = _fast_graph(engine) + q = [n(), e_forward(hops=1), n()] + fast = g.gfql(q) + full = g.gfql(q, policy=_FAST_NOOP_POLICY) + + def kind(df, col): + pdf = df.to_pandas() if "cudf" in type(df).__module__ else df + return pdf[col].dtype.kind + + # The feature's promise: the 1-hop fast path keeps int node attrs as int. + assert kind(fast._nodes, 'attr') == 'i', "fast path must keep int node attrs as int" + # The full BFS path today upcasts int->float (a known merge wart this fast path + # sidesteps). Assert only that it stays numeric, so a future full-path dtype fix + # does not break this test; the fast-path promise above is what we hard-lock. + assert kind(full._nodes, 'attr') in ('i', 'f') + # node-only never traverses a merge, so it stays int regardless of path. + assert kind(g.gfql([n()])._nodes, 'attr') == 'i' + + +def test_fast_path_gating_returns_none_for_ineligible(): + """Unit-level gate: _try_chain_fast_path must DECLINE (return None) for every + shape/condition it does not cover, so those queries reach the correct full + path. Eligible shapes must be accepted (non-None).""" + from graphistry.Engine import Engine + g = _fast_graph("pandas") + seed = pd.DataFrame({'v': [0]}) + + eligible = [ + [n()], + [n({'attr': 20})], + [n(), e_forward(hops=1), n()], + [n(), e_reverse(hops=1), n()], + ] + for ops in eligible: + assert _try_chain_fast_path(g, ops, Engine.PANDAS, None) is not None, f"should accept {ops}" + + ineligible = [ + ("hops_2", [n(), e_forward(hops=2), n()], None, Engine.PANDAS), + ("filtered_undirected", [n({'attr': 10}), e_undirected(hops=1), n({'attr': 30})], None, Engine.PANDAS), + ("edge_match", [n(), e_forward(hops=1, edge_match={'w': 5}), n()], None, Engine.PANDAS), + ("named_node", [n(name='x'), e_forward(hops=1), n()], None, Engine.PANDAS), + ("node_query", [n(query='attr > 5'), e_forward(hops=1), n()], None, Engine.PANDAS), + ("prune_endpoints", [n(), e_forward(hops=1, prune_to_endpoints=True), n()], None, Engine.PANDAS), + ("seeded", [n()], seed, Engine.PANDAS), + ("non_eager_engine", [n()], None, Engine.DASK), + ("two_ops", [n(), e_forward(hops=1)], None, Engine.PANDAS), + ] + for label, ops, sn, eng in ineligible: + assert _try_chain_fast_path(g, ops, eng, sn) is None, f"should decline {label}" + + +def test_chain_otel_span_attrs_mapped_correctly(monkeypatch): + """Regression: the `gfql.chain` otel decorator must wrap `chain()`, not the + `_try_chain_fast_path` helper defined just above it. If it drifts onto the + fast path, `_chain_otel_attrs` receives the fast path's positional args + (g_in, ops, engine_concrete, start_nodes) so `gfql.validate_schema` gets bound + to start_nodes (a DataFrame/None) and `chain()` itself emits no span. + Enable otel + detail, capture spans, and assert correct attr mapping.""" + import importlib + import graphistry.compute.chain as chain_mod + from contextlib import contextmanager + # `import graphistry.otel` binds to a shadowing client attr, so resolve the + # real module via importlib. otel_enabled/otel_span are looked up in the otel + # module (inside otel_traced's wrapper); otel_detail_enabled is looked up in + # chain.py (inside _chain_otel_attrs). Patch each in its own namespace. + otel_mod = importlib.import_module("graphistry.otel") + + captured = [] + monkeypatch.setattr(otel_mod, "otel_enabled", lambda: True) + monkeypatch.setattr(chain_mod, "otel_detail_enabled", lambda: True) + + @contextmanager + def _fake_span(name, attrs=None): + captured.append((name, attrs or {})) + yield None + + monkeypatch.setattr(otel_mod, "otel_span", _fake_span) + + g = CGFull().nodes(pd.DataFrame({'v': [0, 1, 2]}), 'v').edges( + pd.DataFrame({'s': [0, 1], 'd': [1, 2]}), 's', 'd') + g.gfql([n()]) # fast-path-eligible shape + + chain_spans = [a for (nm, a) in captured if nm == "gfql.chain"] + assert chain_spans, "chain() must emit a gfql.chain span" + attrs = chain_spans[0] + assert attrs.get("gfql.chain_len") == 1 + # The bug bound validate_schema to start_nodes (None / a DataFrame); the + # correct mapping is the bool default. + assert isinstance(attrs.get("gfql.validate_schema"), bool), \ + f"validate_schema attr must be a bool, got {type(attrs.get('gfql.validate_schema'))}" + + +@pytest.mark.parametrize("engine", ["pandas", "cudf"]) +def test_fast_path_drops_edges_to_absent_nodes(engine): + """The 1-hop fast path must drop edges whose endpoints are not in the node + table (the full BFS path does, via its edge<->node joins). A node table that + omits an edge endpoint must not yield dangling edges — nor a non-empty result + where the full path is empty.""" + nodes = pd.DataFrame({'v': [0, 1], 'attr': [1, 2]}) + edges = pd.DataFrame({'s': [0, 1], 'd': [1, 99]}) # 99 absent from nodes + if engine == "cudf": + cudf = _cudf_or_skip() + nodes, edges = cudf.DataFrame.from_pandas(nodes), cudf.DataFrame.from_pandas(edges) + g = CGFull().nodes(nodes, 'v').edges(edges, 's', 'd') + for q in ([n(), e_forward(hops=1), n()], + [n(), e_reverse(hops=1), n()], + [n(), e_undirected(hops=1), n()], + [n({'attr': 2}), e_forward(hops=1), n()]): + assert _setsig(g.gfql(q)) == _setsig(g.gfql(q, policy=_FAST_NOOP_POLICY)), \ + f"dangling-edge divergence for {q}" + + +@pytest.mark.parametrize("engine", ["pandas", "cudf"]) +def test_fast_path_drops_nan_endpoint_edges(engine): + """A NaN node id must not validate a NaN edge endpoint. pandas/cuDF `.isin` + treat NaN as matchable (NaN.isin([NaN]) is True), but the full BFS path's joins + never match NaN<->NaN, so it drops NaN-endpoint edges. The fast path's `.dropna()` + on the node-id column must keep it consistent. Regression guard for the NaN fix.""" + import numpy as np + nodes = pd.DataFrame({'v': [0.0, 1.0, np.nan], 'attr': [1, 2, 3]}) # NaN node id present + edges = pd.DataFrame({'s': [0.0, 1.0], 'd': [1.0, np.nan]}) # NaN destination endpoint + if engine == "cudf": + cudf = _cudf_or_skip() + nodes, edges = cudf.DataFrame.from_pandas(nodes), cudf.DataFrame.from_pandas(edges) + g = CGFull().nodes(nodes, 'v').edges(edges, 's', 'd') + for q in ([n(), e_forward(hops=1), n()], [n(), e_reverse(hops=1), n()]): + assert _setsig(g.gfql(q)) == _setsig(g.gfql(q, policy=_FAST_NOOP_POLICY)), \ + f"NaN-endpoint divergence for {q}" + + +@pytest.mark.parametrize("engine", ["pandas", "cudf"]) +def test_fast_path_dedups_duplicate_node_ids_on_hop(engine): + """A malformed node table with duplicate ids must not make the 1-hop fast path + diverge from the full path (which collapses dup rows via its merge).""" + nodes = pd.DataFrame({'v': [0, 0, 1, 2], 'attr': [1, 1, 2, 3]}) + edges = pd.DataFrame({'s': [0, 1], 'd': [1, 2]}) + if engine == "cudf": + cudf = _cudf_or_skip() + nodes, edges = cudf.DataFrame.from_pandas(nodes), cudf.DataFrame.from_pandas(edges) + g = CGFull().nodes(nodes, 'v').edges(edges, 's', 'd') + q = [n(), e_forward(hops=1), n()] + assert _setsig(g.gfql(q)) == _setsig(g.gfql(q, policy=_FAST_NOOP_POLICY)) diff --git a/graphistry/tests/test_policy_hooks.py b/graphistry/tests/test_policy_hooks.py index 7dc0808ae7..72db1e47cd 100644 --- a/graphistry/tests/test_policy_hooks.py +++ b/graphistry/tests/test_policy_hooks.py @@ -63,6 +63,34 @@ def postload_policy(context: PolicyContext) -> None: g.gfql([n()], policy={'postload': postload_policy}) assert hook_called['postload'], "Postload hook should have been called" + @pytest.mark.parametrize("ops_label,ops", [ + ("node_only", [n()]), # 1-op chain fast-path shape + ("single_hop", [n(), e(), n()]), # 3-op chain fast-path shape + ]) + def test_fast_path_shapes_still_invoke_postload(self, ops_label, ops): + """Regression (#1650): the degenerate-shape chain fast path must not bypass + policy hooks. Node-only and single-hop chains are exactly the shapes the fast + path short-circuits; with a policy attached they must take the full path so + postload still fires. See chain.py `_try_chain_fast_path` gating. + + Uses an edges-only graph (no node binding) on purpose: the full single-hop + path must keep the synthesized node-id binding through endpoint reconciliation + and return a valid result, not a corrupt `None`-named column (which also + crashes the concat on newer pandas). See chain.py g_out rebuild.""" + called = {'postload': False} + + def postload_policy(context: PolicyContext) -> None: + called['postload'] = True + + df = pd.DataFrame({'s': ['a', 'b', 'c'], 'd': ['b', 'c', 'd']}) + g = graphistry.edges(df, 's', 'd') + + out = g.gfql(ops, policy={'postload': postload_policy}) + assert called['postload'], f"postload must fire for {ops_label} fast-path shape" + # result must carry a valid node-id binding + no spurious None-named column + assert out._node is not None, f"{ops_label}: node-id binding lost" + assert None not in list(out._nodes.columns), f"{ops_label}: corrupt None column" + def test_precall_hook_called(self): """Test that precall hook is called for call operations.""" from graphistry.compute.ast import call