diff --git a/CHANGELOG.md b/CHANGELOG.md index 95b0306882..1a1ad99e9c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -39,6 +39,7 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm ### Added - **GFQL native Polars bindings-row tables (`rows(binding_ops)`) — traversal Cypher on polars (#1709)**: the Cypher multi-alias lowering's `rows(binding_ops=...)` op (one row per matched path) now runs natively on `engine='polars'` for **fixed-length connected patterns** — unblocking traversal-shaped Cypher that previously NIE'd: multi-alias property projections (`MATCH (a)-[e]->(b) RETURN a.x, e.w, b.y`), top-k in-degree (`RETURN b.id, count(a) ... ORDER BY ... LIMIT`, graph-benchmark q1/q2), and fixed multi-hop counts (`MATCH (a)-->(b)-->(c) RETURN count(*)`, q8/q9), with forward/reverse/undirected edges, node/edge filters, and edge-alias payload columns — **plus bounded directed variable-length segments** (`-[*i..k]->`, typed `-[:TYPE*i..k]->`, exactly-k; iterative pair joins with Cypher path multiplicity and zero-hop rows), covering the q3 shape (`MATCH (a:Person)-[:FOLLOWS*1..2]->(b:Person) ... RETURN avg(b.age)`). This rung covers the q1–q4 and q8–q9 binding-table shapes exercised by its parity tests; connected q5–q7 planning/execution is layered in follow-on PRs. Also adds native `with_(extend=True)` (emitted by the bindings-path aggregate lowering) and an honest decline for `group_by(key_prefixes=...)` (whole-row bindings grouping — was a latent silent-wrong-key trap). **Honestly deferred** (`NotImplementedError`, no pandas bridge): unbounded `[*]`, undirected/aliased variable-length, shortestPath scalar bindings, node/edge `query=`/endpoint-match params, cartesian (`MATCH (a),(b)`) mode, seeded re-entry contexts. Differential parity vs the pandas oracle (+20 tests incl. path multiplicity and undirected self-loops; conformance corpus extended). The Polars-GPU engine uses the same lazy target-collected plan; exact-head DGX validation is required before making a GPU performance claim. +- **GFQL Cypher connected-pattern (q5–q7) planning + execution on all engines**: completes the `rows(binding_ops)` follow-on promised above — multi-alias *connected* `MATCH` patterns that filter and join across shared endpoints (`MATCH (a)-[]->(b)<-[]-(c) WHERE … RETURN …`, star/two-star shapes) now plan and execute natively on pandas, cuDF, `engine='polars'`, and `engine='polars-gpu'` (previously an honest `NotImplementedError` on polars). The connected-join planner pushes eligible single-alias endpoint predicates into per-frame `filter_dict` prefilters when the column dtype makes the pushdown exact, and keeps them as a post-join `where_rows` residual otherwise (nullable/decimal/object/predicate/`AllOf`/`toLower` shapes), so results stay byte-identical to the full-join semantics on every engine. The two-star executor adds grouped-count and bindings-row fast paths with a per-execution result cache (no cross-call staleness under in-place graph mutation) and openCypher-correct ordering — NULL sorts as the largest value (`ORDER BY … ASC` puts nulls last), and `count(a)`/`count(DISTINCT a)`/`count(*)` follow Cypher multiplicity. Differential parity vs the hand-derived openCypher oracle across all four engines (connected-join conformance corpus). ### Fixed - **GFQL seeded indexes now engage for small Polars/cuDF frontiers**: NaN-free native Polars frames retain object identity during normalization, preserving resident CSR validity, while a tunable absolute floor prevents the fractional cost gate from routing tiny seeded hops over low-cardinality slices to a full scan. Scan/index result parity is unchanged; the default floor is a routing heuristic whose engine/data crossover remains subject to measurement. @@ -46,7 +47,8 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm - **GFQL Cypher `=~` / scalar-fn cross-engine hardening (review wave, dgx-verified)**: (1) composed `=~` (`WHERE … =~ … OR …`, `RETURN`-expression position) now works on `engine='cudf'` — the series evaluator used raw `.str.fullmatch`, which cuDF lacks, and the resulting `AttributeError` was masked as "unsupported predicate op" (it now routes through the `Fullmatch` predicate's engine workarounds, and honest `NotImplementedError` declines pass through instead of being re-labeled); (2) the cuDF fullmatch emulation anchors alternations as a whole (`^(ab|cd)$` — bare `^ab|cd$` silently matched `'abXXX'`); (3) the cuDF case-insensitive `(?i)` lowercase-folding workaround now declines the fold-unsound shapes — uppercase escape classes (`.lower()` turns `\D` into `\d`, silently inverting the predicate), case-crossing character ranges (`(?i)[A-z]` silently narrowed; `[X-b]` folded to an invalid range), and non-ASCII patterns — while lowercase escapes (`\d`, `\.`) keep folding as before; lookaround, backreferences, and named-group refs decline up front (libcudf rejects them at kernel-compile time); (4) polars `Match`/`Fullmatch` lowering applies the same Rust-regex guard as `Contains` (lookaround/backrefs decline instead of a non-NIE `ComputeError` at collect); (5) `toLower`/`toUpper` on a non-string column decline like neo4j's type error instead of broadcasting the stringified Series repr (pandas/cuDF) or raising a non-NIE `SchemaError` (polars-gpu); (6) polars `floor`/`ceil`/`round` cast to `Float64` so integer columns return Float like neo4j and the pandas engine; (7) invalid regex patterns on the composed path raise a clear "invalid regex pattern" error instead of "unsupported predicate op". - **GFQL Cypher `round()` hardening (review wave, dgx-verified)**: (1) the `polars` extra's floor is now `polars>=1.29` — `Expr.round(mode=)` shipped in py-1.29.0 (pola-rs/polars#22248), not 1.5 as previously pinned, so 1.5–1.28 installs crashed with a raw `TypeError` on `round(x, p>0)` under `engine='polars'`; (2) `round(x, p>308)` is the identity on both engines (a float64 has no digits there) instead of pandas raising through an unclear decline while polars returned identity — parity restored, `10.0**p` overflow guarded; (3) polars `round(x, p>0)` normalizes `-0.0` to `+0.0` like the pandas kernel (`round(-0.04, 1)` was `0.0` on pandas vs `-0.0` on polars — invisible to value equality, pinned by a sign-bit test); (4) documented the precision>0 decimal-string deviation vs neo4j (`round(2.675, 2)` = `2.67` binary-double here vs `2.68` BigDecimal there) and added deterministic tie/hazard matrix cases so ties actually reach cuDF/polars-gpu (the fixture's normal-distribution floats never tied). - **GFQL `materialize_nodes()` on an edges-only graph under `engine='polars'`**: crashed with `AttributeError: 'Series' object has no attribute 'drop_duplicates'` — the node-id derivation used pandas-only `.drop_duplicates()` / `.reset_index()` / `.rename(columns=)` on a polars Series/DataFrame. Now polars-aware (`.unique(maintain_order=True)`, `.select().rename({...})`), matching the pandas oracle. Surfaced by any `let()` DAG on an edges-only graph under Polars (the DAG pre-materializes nodes). pandas/cuDF paths unchanged. -- **CI: `compute_networkx` HITS test was scipy-version flaky**: `test_compute_networkx_hits_outputs_hubs_and_authorities` used a pure directed 3-cycle (`a→b→c→a`), whose adjacency is a permutation matrix (all singular values equal). networkx HITS computes scores via scipy `svds(k=1)`, which then returns an arbitrary vector from that fully-degenerate singular space; when its components sum to ~0, HITS's `h /= h.sum()` normalization blows up to `inf`/`nan`. Which vector `svds` returns is scipy/LAPACK-version dependent, so the test passed on the pinned scipy but failed on the `nx-upper-scipy` CI profile's newer scipy — a latent flake, not a code regression. Switched to a graph with a well-defined hub/authority structure (`a→b`, `a→c`, `b→c`), whose dominant singular vector is unique (Perron-Frobenius) and therefore version-stable. Test-only change.- **GFQL `materialize_nodes()` on an edges-only graph under `engine='polars'`**: crashed with `AttributeError: 'Series' object has no attribute 'drop_duplicates'` — the node-id derivation used pandas-only `.drop_duplicates()` / `.reset_index()` / `.rename(columns=)` on a polars Series/DataFrame. Now polars-aware (`.unique(maintain_order=True)`, `.select().rename({...})`), matching the pandas oracle. Surfaced by any `let()` DAG on an edges-only graph under Polars (the DAG pre-materializes nodes). pandas/cuDF paths unchanged.- **GFQL `engine='polars-gpu'` LazyFrame-input coercion now collects on the GPU executor**: `Engine.df_to_engine(lazyframe, POLARS_GPU)` materialized a `polars.LazyFrame` input with a bare `.collect()` on the CPU default executor — ignoring `gpu_executor()` and not distinguishing `POLARS` from `POLARS_GPU`. It now routes through the target-aware lazy collect, so a LazyFrame coerced under `polars-gpu` collects on the cudf-polars GPU executor (and `polars` honors `cpu_streaming()`). No change for already-materialized frame inputs (the common path). +- **CI: `compute_networkx` HITS test was scipy-version flaky**: `test_compute_networkx_hits_outputs_hubs_and_authorities` used a pure directed 3-cycle (`a→b→c→a`), whose adjacency is a permutation matrix (all singular values equal). networkx HITS computes scores via scipy `svds(k=1)`, which then returns an arbitrary vector from that fully-degenerate singular space; when its components sum to ~0, HITS's `h /= h.sum()` normalization blows up to `inf`/`nan`. Which vector `svds` returns is scipy/LAPACK-version dependent, so the test passed on the pinned scipy but failed on the `nx-upper-scipy` CI profile's newer scipy — a latent flake, not a code regression. Switched to a graph with a well-defined hub/authority structure (`a→b`, `a→c`, `b→c`), whose dominant singular vector is unique (Perron-Frobenius) and therefore version-stable. Test-only change. +- **GFQL `engine='polars-gpu'` LazyFrame-input coercion now collects on the GPU executor**: `Engine.df_to_engine(lazyframe, POLARS_GPU)` materialized a `polars.LazyFrame` input with a bare `.collect()` on the CPU default executor — ignoring `gpu_executor()` and not distinguishing `POLARS` from `POLARS_GPU`. It now routes through the target-aware lazy collect, so a LazyFrame coerced under `polars-gpu` collects on the cudf-polars GPU executor (and `polars` honors `cpu_streaming()`). No change for already-materialized frame inputs (the common path). - **GFQL Polars engine `contains` honors `regex=`/`flags=`**: the native polars `filter_by_dict` lowering of the `Contains` predicate always used `str.contains(..., literal=False)`, so a **literal** request (`contains(pat, regex=False)`) was still regex-interpreted — a pattern with a metacharacter over-matched (e.g. `contains('a.c', regex=False)` matched `'abc'`), diverging from pandas/cuDF. It also dropped `flags=`. Now `regex=False` lowers to a literal match (`literal=True`; case-insensitive literal folds both sides, matching pandas' result), and regex mode maps `case=`/`flags=` (IGNORECASE/MULTILINE/DOTALL/VERBOSE) to a Rust-regex inline flag prefix (`(?ims…)`). +1 engine-parametrized differential-parity test. - **GFQL adjacency-index cost gate is engine-aware (never slower than scan)**: the seeded-hop planner falls back to a full scan once the frontier covers too large a fraction of the source keys (past that, scanning all edges once beats many index probes). That crossover fraction is *engine-dependent* — a vectorized-scan engine (polars/cuDF/GPU) has a far faster scan, so its crossover is much smaller than pandas'. The gate previously used a single `0.5·n_keys` threshold, so on polars a **resident index under `index_policy='use'` ran ~2× slower than the plain scan** for mid-size frontiers (measured ~frac 0.02–0.5; correct result, just slow). The gate now uses a per-engine crossover (pandas ~0.5, vectorized engines ~0.02; GPU provisional pending large-graph measurement), so a resident index never loses to the un-indexed path on any engine. Small-frontier wins (the point of the index) are unchanged. +1 engine-parametrized regression test. - **GFQL `ne()` / `<>` on NULL now follows openCypher/SQL 3-valued logic (pandas)**: `n({"col": ne(x)})` and cypher `WHERE n.col <> x` over a NULL/NA cell used to KEEP the null row on the pandas engine (`NaN != x` → True), diverging from cuDF and the polars engine (both drop it) — and even from pandas' own `WHERE NOT n.col = x` path. Per openCypher/SQL three-valued logic, `null <> x` is `null` (an unknown value cannot be proven unequal to `x`), so a null cell is **not** a match and the row is excluded — consistent with `eq`/`gt`/`lt`/`IN` (which already dropped nulls). Fixed the `NE` predicate to mask out nulls; this corrects both the `filter_dict` predicate path and the single-entity cypher `<>` WHERE path on pandas. cuDF/polars/polars-gpu were already conformant. Verified across all four engines (`ne`, `<>`, `NOT =`, `NOT IN` all drop the null). Note: this is a behavior change for `ne()` on nullable columns under the default pandas engine. (Broader openCypher null-semantics alignment + docs tracked in #1664.) diff --git a/bin/ci_cypher_surface_guard_baseline.json b/bin/ci_cypher_surface_guard_baseline.json index 72fc38aa20..ef5aa58e40 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": 9237 + "lowering_py_max_lines": 9244 } diff --git a/graphistry/compute/dataframe/join.py b/graphistry/compute/dataframe/join.py index 6f7081fa94..dc8fc97aa2 100644 --- a/graphistry/compute/dataframe/join.py +++ b/graphistry/compute/dataframe/join.py @@ -3,7 +3,7 @@ import operator from typing import Any, Dict, List, Optional, Sequence, Tuple, cast -from graphistry.Engine import Engine +from graphistry.Engine import Engine, POLARS_ENGINES from graphistry.compute.typing import DataFrameT, DomainT @@ -20,9 +20,15 @@ def joined_hidden_scalar_columns(frame: DataFrameT) -> DataFrameT: if suffix.startswith("__cypher_reentry_") or suffix.startswith("__gfql_hidden_"): hidden_suffixes.setdefault(suffix, []).append(column) out = frame + is_polars = "polars" in type(frame).__module__ for suffix, columns in hidden_suffixes.items(): if suffix in out.columns: continue + if is_polars: + import polars as pl + expr = pl.coalesce([pl.col(column) for column in columns]).alias(suffix) + out = out.with_columns(expr) + continue series = out[columns[0]] for column in columns[1:]: if hasattr(series, "combine_first"): @@ -44,6 +50,13 @@ def joined_alias_columns(frame: DataFrameT) -> DataFrameT: elif suffix == "id" and alias not in alias_candidates: alias_candidates[alias] = column out = frame + is_polars = "polars" in type(frame).__module__ + if is_polars and alias_candidates: + import polars as pl + return cast(DataFrameT, out.with_columns([ + pl.col(source_column).alias(alias) + for alias, source_column in alias_candidates.items() + ])) for alias, source_column in alias_candidates.items(): out = out.assign(**{alias: out[source_column]}) return out @@ -65,6 +78,8 @@ def connected_inner_join_rows( join_cols_list = list(join_cols) keep_cols_list = list(keep_cols) rhs = cast(DataFrameT, pattern_rows[keep_cols_list]) + if engine in POLARS_ENGINES: + return cast(DataFrameT, joined_rows.join(rhs, on=join_cols_list, how="inner")) if engine != Engine.CUDF: return cast(DataFrameT, joined_rows.merge(rhs, on=join_cols_list, how="inner")) diff --git a/graphistry/compute/gfql/cypher/lowering.py b/graphistry/compute/gfql/cypher/lowering.py index 7f1b56fbd2..a47aa415a5 100644 --- a/graphistry/compute/gfql/cypher/lowering.py +++ b/graphistry/compute/gfql/cypher/lowering.py @@ -8209,6 +8209,13 @@ def _connected_join_required_property_aliases( return None required.update(alias for alias in non_aggregate_aliases if alias in node_aliases) required.update(alias for alias in prop_aliases if alias in node_aliases) + # A bare node alias used *inside* an aggregate (e.g. ``count(p)`` / + # ``count(DISTINCT p)``) is rewritten by _connected_join_alias_identity_expr to + # ``p.__gfql_node_id__``, so its identity column must still be attached even though + # it is neither a non-aggregate use nor a property access. Attaching by alias here + # is a no-op when the alias is already required via a property access (e.g. + # ``count(p.age)``); it only adds the otherwise-missing bare-aggregate case. + required.update(alias for alias in _aggregate_aliases if alias in node_aliases) return required diff --git a/graphistry/compute/gfql_unified.py b/graphistry/compute/gfql_unified.py index 179c14c98c..3aaf151bd1 100644 --- a/graphistry/compute/gfql_unified.py +++ b/graphistry/compute/gfql_unified.py @@ -2,8 +2,9 @@ # ruff: noqa: E501 from dataclasses import replace +import pandas as pd from types import MappingProxyType -from typing import Any, Dict, List, Literal, Mapping, Optional, Sequence, Tuple, Union, cast +from typing import Any, Dict, List, Literal, Mapping, Optional, Sequence, Set, Tuple, Union, cast from graphistry.Plottable import Plottable from graphistry.Engine import Engine, EngineAbstract, POLARS_ENGINES, df_concat, df_cons, df_to_engine, df_unique, resolve_engine from graphistry.util import setup_logger @@ -22,6 +23,7 @@ expand_policy ) from graphistry.compute.gfql.same_path_types import ( + NODE_IDENTITY_COLUMN, WhereComparison, normalize_where_entries, parse_where_json, @@ -65,6 +67,7 @@ joined_alias_columns as _joined_alias_columns, joined_hidden_scalar_columns as _joined_hidden_scalar_columns, ) +from graphistry.compute.filter_by_dict import filter_by_dict from graphistry.compute.gfql.ir.compilation import PhysicalPlan, PlanContext from graphistry.compute.gfql.ir.logical_plan import LogicalPlan from graphistry.compute.gfql.physical_planner import PhysicalPlanner @@ -472,6 +475,1115 @@ def _optional_arm_start_nodes( return _chain_dispatch(joined_plottable, plan.post_join_chain, engine, policy, context) + +def _is_connected_fast_single_hop(edge_op: ASTEdge) -> bool: + return ( + getattr(edge_op, "direction", None) == "forward" + and getattr(edge_op, "hops", None) in (None, 1) + and getattr(edge_op, "min_hops", None) is None + and getattr(edge_op, "max_hops", None) is None + and getattr(edge_op, "output_min_hops", None) is None + and getattr(edge_op, "output_max_hops", None) is None + and getattr(edge_op, "label_node_hops", None) is None + and getattr(edge_op, "label_edge_hops", None) is None + and not bool(getattr(edge_op, "label_seeds", False)) + and not bool(getattr(edge_op, "to_fixed_point", False)) + and getattr(edge_op, "source_node_match", None) is None + and getattr(edge_op, "destination_node_match", None) is None + and getattr(edge_op, "source_node_query", None) is None + and getattr(edge_op, "destination_node_query", None) is None + and getattr(edge_op, "edge_query", None) is None + and getattr(edge_op, "_name", None) is None + and not bool(getattr(edge_op, "include_zero_hop_seed", False)) + ) + + + + +_CACHE_MISSING = object() + + +def _connected_join_filter_value_cache_key(value: Any) -> Optional[Tuple[str, str]]: + if isinstance(value, bool): + return "bool", "true" if value else "false" + if isinstance(value, str): + return "str", value + if isinstance(value, int) and not isinstance(value, bool): + return "int", str(value) + if isinstance(value, float): + return "float", repr(value) + if value is None: + return "none", "" + + predicate_value = getattr(value, "val", _CACHE_MISSING) + if predicate_value is not _CACHE_MISSING: + scalar_key = _connected_join_filter_value_cache_key(predicate_value) + if scalar_key is None: + return None + return f"{type(value).__module__}.{type(value).__qualname__}", f"{scalar_key[0]}:{scalar_key[1]}" + + regex_pattern = getattr(value, "pat", _CACHE_MISSING) + regex_case = getattr(value, "case", _CACHE_MISSING) + regex_flags = getattr(value, "flags", _CACHE_MISSING) + regex_na = getattr(value, "na", _CACHE_MISSING) + if ( + regex_pattern is not _CACHE_MISSING + and regex_case is not _CACHE_MISSING + and regex_flags is not _CACHE_MISSING + and regex_na is not _CACHE_MISSING + ): + pattern_key = _connected_join_filter_value_cache_key(regex_pattern) + case_key = _connected_join_filter_value_cache_key(regex_case) + flags_key = _connected_join_filter_value_cache_key(regex_flags) + na_key = _connected_join_filter_value_cache_key(regex_na) + if pattern_key is None or case_key is None or flags_key is None or na_key is None: + return None + return ( + f"{type(value).__module__}.{type(value).__qualname__}", + f"pat={pattern_key[0]}:{pattern_key[1]}|case={case_key[0]}:{case_key[1]}|flags={flags_key[0]}:{flags_key[1]}|na={na_key[0]}:{na_key[1]}", + ) + + predicates = getattr(value, "predicates", _CACHE_MISSING) + if predicates is not _CACHE_MISSING and isinstance(predicates, (list, tuple)): + child_keys = [] + for predicate in predicates: + child_key = _connected_join_filter_value_cache_key(predicate) + if child_key is None: + return None + child_keys.append(f"{child_key[0]}:{child_key[1]}") + return f"{type(value).__module__}.{type(value).__qualname__}", "|".join(child_keys) + + return None + + +def _connected_join_simple_filter_cache_key(filter_dict: Optional[dict]) -> Optional[Tuple[Tuple[str, str, str], ...]]: + if not filter_dict: + return () + items: List[Tuple[str, str, str]] = [] + for key, value in filter_dict.items(): + if not isinstance(key, str): + return None + value_key = _connected_join_filter_value_cache_key(value) + if value_key is None: + return None + items.append((key, value_key[0], value_key[1])) + return tuple(sorted(items)) + + +def _connected_join_cached_node_filter( + base_graph: Plottable, + nodes_obj: DataFrameT, + node_match: Optional[dict], + *, + engine: Engine, + cache_store: Optional[Dict[str, Any]] = None, +) -> DataFrameT: + cache_key = _connected_join_simple_filter_cache_key(node_match) + if cache_key is None: + nodes = df_to_engine(nodes_obj, engine) + if engine in POLARS_ENGINES: + from graphistry.compute.gfql.lazy.engine.polars.predicates import filter_by_dict_polars + return cast(DataFrameT, filter_by_dict_polars(nodes, node_match)) + return filter_by_dict(nodes, node_match, engine=EngineAbstract(engine.value)) + + cache_attr = "_gfql_connected_join_node_filter_cache" + # Per-execution cache only (threaded via cache_store); NEVER setattr onto the caller's + # Plottable -- that leaked results across gfql() calls keyed by id(), returning stale + # answers after an in-place edge/node mutation (BLOCKER 1). None => no caching. + cache = cache_store.setdefault(cache_attr, {}) if cache_store is not None else None + full_key = (id(nodes_obj), engine.value, cache_key) + if cache is not None and full_key in cache: + return cast(DataFrameT, cache[full_key]) + + nodes = df_to_engine(nodes_obj, engine) + if engine in POLARS_ENGINES: + from graphistry.compute.gfql.lazy.engine.polars.predicates import filter_by_dict_polars + filtered = cast(DataFrameT, filter_by_dict_polars(nodes, node_match)) + else: + filtered = filter_by_dict(nodes, node_match, engine=EngineAbstract(engine.value)) + if cache is not None: + cache[full_key] = filtered + return cast(DataFrameT, filtered) + + +def _connected_join_cached_node_ids( + base_graph: Plottable, + nodes_obj: DataFrameT, + node_matches: Sequence[Optional[dict]], + *, + node_col: str, + engine: Engine, + cache_store: Optional[Dict[str, Any]] = None, +) -> Optional[DataFrameT]: + if engine not in POLARS_ENGINES: + return None + + match_keys: List[Tuple[Tuple[str, str, str], ...]] = [] + for node_match in node_matches: + match_key = _connected_join_simple_filter_cache_key(node_match) + if match_key is None: + return None + match_keys.append(match_key) + + cache_attr = "_gfql_connected_join_node_ids_cache" + # Per-execution cache only (threaded via cache_store); NEVER setattr onto the caller's + # Plottable -- that leaked results across gfql() calls keyed by id(), returning stale + # answers after an in-place edge/node mutation (BLOCKER 1). None => no caching. + cache = cache_store.setdefault(cache_attr, {}) if cache_store is not None else None + full_key = (id(nodes_obj), engine.value, node_col, tuple(match_keys)) + if cache is not None and full_key in cache: + return cast(DataFrameT, cache[full_key]) + + from graphistry.compute.gfql.lazy.engine.polars.predicates import filter_by_dict_polars + + nodes = df_to_engine(nodes_obj, engine) + filtered = nodes + for node_match in node_matches: + filtered = cast(DataFrameT, filter_by_dict_polars(filtered, node_match)) + ids = cast(DataFrameT, filtered.select(node_col).unique()) + if cache is not None: + cache[full_key] = ids + return ids + + +def _connected_join_cached_edge_filter( + base_graph: Plottable, + edges_obj: DataFrameT, + edge_match: Optional[dict], + *, + engine: Engine, + cache_store: Optional[Dict[str, Any]] = None, +) -> DataFrameT: + cache_key = _connected_join_simple_filter_cache_key(edge_match) + if cache_key is None: + edges = df_to_engine(edges_obj, engine) + if engine in POLARS_ENGINES: + from graphistry.compute.gfql.lazy.engine.polars.predicates import filter_by_dict_polars + return cast(DataFrameT, filter_by_dict_polars(edges, edge_match)) + return filter_by_dict(edges, edge_match, engine=EngineAbstract(engine.value)) + + cache_attr = "_gfql_connected_join_edge_filter_cache" + # Per-execution cache only (threaded via cache_store); NEVER setattr onto the caller's + # Plottable -- that leaked results across gfql() calls keyed by id(), returning stale + # answers after an in-place edge/node mutation (BLOCKER 1). None => no caching. + cache = cache_store.setdefault(cache_attr, {}) if cache_store is not None else None + full_key = (id(edges_obj), engine.value, cache_key) + if cache is not None and full_key in cache: + return cast(DataFrameT, cache[full_key]) + + edges = df_to_engine(edges_obj, engine) + if engine in POLARS_ENGINES: + from graphistry.compute.gfql.lazy.engine.polars.predicates import filter_by_dict_polars + filtered = cast(DataFrameT, filter_by_dict_polars(edges, edge_match)) + else: + filtered = filter_by_dict(edges, edge_match, engine=EngineAbstract(engine.value)) + if cache is not None: + cache[full_key] = filtered + return cast(DataFrameT, filtered) + + +def _connected_join_cached_singleton_dst_source_counts( # pragma: no cover + base_graph: Plottable, + edges_obj: DataFrameT, + edge_domain: DataFrameT, + edge_match: Optional[dict], + singleton_dst: Any, + *, + src_col: str, + dst_col: str, + engine: Engine, + cache_store: Optional[Dict[str, Any]] = None, +) -> Optional[DataFrameT]: + edge_key = _connected_join_simple_filter_cache_key(edge_match) + dst_key = _connected_join_filter_value_cache_key(singleton_dst) + if edge_key is None or dst_key is None: + return None + + cache_attr = "_gfql_connected_join_singleton_dst_source_counts_cache" + # Per-execution cache only (threaded via cache_store); NEVER setattr onto the caller's + # Plottable -- that leaked results across gfql() calls keyed by id(), returning stale + # answers after an in-place edge/node mutation (BLOCKER 1). None => no caching. + cache = cache_store.setdefault(cache_attr, {}) if cache_store is not None else None + full_key = (id(edges_obj), engine.value, src_col, dst_col, edge_key, dst_key) + if cache is not None and full_key in cache: + return cast(DataFrameT, cache[full_key]) + + if engine in POLARS_ENGINES: + import polars as pl + counts = cast( + DataFrameT, + edge_domain + .filter(pl.col(dst_col) == singleton_dst) + .group_by(src_col) + .len("__left_count__"), + ) + else: + filtered = edge_domain[edge_domain[dst_col] == singleton_dst] + counts = cast(DataFrameT, filtered.groupby(src_col, sort=False).size().reset_index(name="__left_count__")) + + if cache is not None: + cache[full_key] = counts + return counts + + +def _connected_join_cached_first_arm_shared_counts( + base_graph: Plottable, + nodes_obj: DataFrameT, + edges_obj: DataFrameT, + first_edges: DataFrameT, + shared_ids: DataFrameT, + first_edge_match: Optional[dict], + first_start_match: Optional[dict], + second_start_match: Optional[dict], + singleton_dst: Any, + *, + shared_alias: str, + src_col: str, + dst_col: str, + engine: Engine, + cache_store: Optional[Dict[str, Any]] = None, +) -> Optional[DataFrameT]: + if engine not in POLARS_ENGINES: + return None + + edge_key = _connected_join_simple_filter_cache_key(first_edge_match) + first_start_key = _connected_join_simple_filter_cache_key(first_start_match) + second_start_key = _connected_join_simple_filter_cache_key(second_start_match) + dst_key = _connected_join_filter_value_cache_key(singleton_dst) + if edge_key is None or first_start_key is None or second_start_key is None or dst_key is None: + return None + + cache_attr = "_gfql_connected_join_first_arm_shared_counts_cache" + # Per-execution cache only (threaded via cache_store); NEVER setattr onto the caller's + # Plottable -- that leaked results across gfql() calls keyed by id(), returning stale + # answers after an in-place edge/node mutation (BLOCKER 1). None => no caching. + cache = cache_store.setdefault(cache_attr, {}) if cache_store is not None else None + full_key = ( + id(nodes_obj), + id(edges_obj), + engine.value, + shared_alias, + src_col, + dst_col, + edge_key, + first_start_key, + second_start_key, + dst_key, + ) + if cache is not None and full_key in cache: + return cast(DataFrameT, cache[full_key]) + + import polars as pl + + counts = ( + first_edges + .filter(pl.col(dst_col) == singleton_dst) + .join(shared_ids, left_on=src_col, right_on=shared_alias, how="semi") + .group_by(src_col) + .len("__left_count__") + .rename({src_col: shared_alias}) + ) + if cache is not None: + cache[full_key] = counts + return cast(DataFrameT, counts) + + +def _connected_join_cached_second_arm_group_rows( + base_graph: Plottable, + nodes_obj: DataFrameT, + edges_obj: DataFrameT, + second_edges: DataFrameT, + shared_ids: DataFrameT, + second_leaf_ids: DataFrameT, + second_leaf_nodes: DataFrameT, + first_start_match: Optional[dict], + second_start_match: Optional[dict], + second_leaf_match: Optional[dict], + second_edge_match: Optional[dict], + second_leaf_singleton: Any, + group_prop_refs: List[Tuple[str, str]], + output_group_keys: List[str], + *, + node_col: str, + src_col: str, + dst_col: str, + engine: Engine, + cache_store: Optional[Dict[str, Any]] = None, +) -> Optional[DataFrameT]: + if engine not in POLARS_ENGINES: + return None + + first_start_key = _connected_join_simple_filter_cache_key(first_start_match) + second_start_key = _connected_join_simple_filter_cache_key(second_start_match) + second_leaf_key = _connected_join_simple_filter_cache_key(second_leaf_match) + second_edge_key = _connected_join_simple_filter_cache_key(second_edge_match) + if first_start_key is None or second_start_key is None or second_leaf_key is None or second_edge_key is None: + return None + + prop_key = tuple((str(out_col), str(prop)) for out_col, prop in group_prop_refs) + output_key = tuple(str(key) for key in output_group_keys) + cache_attr = "_gfql_connected_join_second_arm_group_rows_cache" + # Per-execution cache only (threaded via cache_store); NEVER setattr onto the caller's + # Plottable -- that leaked results across gfql() calls keyed by id(), returning stale + # answers after an in-place edge/node mutation (BLOCKER 1). None => no caching. + cache = cache_store.setdefault(cache_attr, {}) if cache_store is not None else None + full_key = ( + id(nodes_obj), + id(edges_obj), + engine.value, + node_col, + src_col, + dst_col, + first_start_key, + second_start_key, + second_leaf_key, + second_edge_key, + prop_key, + output_key, + ) + if cache is not None and full_key in cache: + return cast(DataFrameT, cache[full_key]) + + import polars as pl + + if second_leaf_singleton is _CACHE_MISSING: + right_base = ( + second_edges + .join(shared_ids, left_on=src_col, right_on=node_col, how="semi") + .join(second_leaf_ids, left_on=dst_col, right_on=node_col, how="semi") + ) + else: + right_base = ( + second_edges + .filter(pl.col(dst_col) == second_leaf_singleton) + .join(shared_ids, left_on=src_col, right_on=node_col, how="semi") + ) + if group_prop_refs: + lookup_key = "__gfql_fast_second_leaf_id__" + lookup_exprs = [pl.col(node_col).alias(lookup_key)] + [pl.col(prop).alias(out_col) for out_col, prop in group_prop_refs] + # Dedup by node identity (matches pandas drop_duplicates(subset=[node_col])); a + # duplicate node row would otherwise multiply the join and over-count. + second_lookup = second_leaf_nodes.select(lookup_exprs).unique(subset=[lookup_key]) + right_base = right_base.join(second_lookup, left_on=dst_col, right_on=lookup_key, how="inner") + right_rows = right_base.select([pl.col(src_col)] + [pl.col(key) for key in output_group_keys]) + if cache is not None: + cache[full_key] = right_rows + return cast(DataFrameT, right_rows) + + +def _property_ref(expr: Any, valid_aliases: Sequence[str]) -> Optional[Tuple[str, str]]: + if not isinstance(expr, str) or "." not in expr: + return None + alias, prop = expr.split(".", 1) + if alias not in valid_aliases or not prop: + return None + return alias, prop + + +def _connected_join_post_property_columns(plan: ConnectedMatchJoinPlan, alias: str) -> List[str]: + prefix = f"{alias}." + out: List[str] = [] + + def walk(value: Any) -> None: + if isinstance(value, str): + if value.startswith(prefix): + prop = value[len(prefix):] + if prop and prop not in out: + out.append(prop) + return + if isinstance(value, Mapping): + for item in value.values(): + walk(item) + return + if isinstance(value, (list, tuple)): + for item in value: + walk(item) + + for op in plan.post_join_chain.chain: + if isinstance(op, ASTCall): + walk(op.params) + return out + + +def _connected_join_two_star_split_residuals( + plan: ConnectedMatchJoinPlan, + alias_targets: Mapping[str, ASTObject], + materialized_aliases: Set[str], +) -> Optional[Tuple[Dict[str, List[str]], "Chain"]]: + """Split leading post-join ``where_rows`` residuals by single node alias. + + #1729's connected-join lowering emits row predicates it cannot push into ``filter_dict`` + (e.g. ``toLower(i.interest) = toLower('fine dining')``) as leading ``where_rows`` ops in + ``post_join_chain``. The structural fast paths cannot apply a residual to the aggregated + frame, but a residual that references exactly ONE node alias the fast path materializes can + be applied to that alias's node set before counting. Returns ``(alias -> [expr, ...], + remaining_chain)`` when every leading residual is so attributable, or ``None`` when any + leading residual spans >1 alias, references a non-materialized alias, or is not a bare + ``expr`` residual -- the caller then declines to the slow path. + """ + from graphistry.compute.gfql.cypher.lowering import _expr_match_aliases + + ops = list(plan.post_join_chain.chain) + residuals: Dict[str, List[str]] = {} + consumed = 0 + for op in ops: + if not (isinstance(op, ASTCall) and op.function == "where_rows"): + break + params = op.params or {} + expr = params.get("expr") + if params.get("filter_dict") or not isinstance(expr, str): + return None + try: + aliases = _expr_match_aliases( + expr, alias_targets=alias_targets, params=None, field="where", line=0, column=0 + ) + except Exception: + return None + attributable = {alias for alias in aliases if alias in materialized_aliases} + if len(aliases) != 1 or len(attributable) != 1: + return None + residuals.setdefault(next(iter(attributable)), []).append(expr) + consumed += 1 + rest = Chain(ops[consumed:], where=plan.post_join_chain.where) + return residuals, rest + + +def _connected_join_apply_node_residuals( + base_graph: Plottable, + node_frame: DataFrameT, + alias: str, + exprs: Sequence[str], + node_col: str, + *, + engine: Engine, +) -> DataFrameT: + """Filter a fast-path node frame by single-alias post-join residual expressions. + + Reuses the row pipeline's ``where_rows`` evaluator (identical semantics to the slow path, + so toLower/etc. behave exactly as they would post-join) by aliasing the node columns to + ``alias.col`` and dispatching a where_rows chain, then renaming back. ``validate_schema`` is + disabled because the residual references flat ``alias.col`` columns rather than a bound + graph element. + """ + is_polars = "polars" in type(node_frame).__module__ + if is_polars: + aliased = node_frame.rename({col: f"{alias}.{col}" for col in node_frame.columns}) + else: + aliased = node_frame.rename(columns={col: f"{alias}.{col}" for col in node_frame.columns}) + from graphistry.compute.chain import chain as _chain_fn + + aliased_graph = base_graph.nodes(aliased, f"{alias}.{node_col}") + filtered_graph = _chain_fn( + aliased_graph, + [ASTCall("where_rows", {"expr": expr}) for expr in exprs], + engine=EngineAbstract(engine.value), + validate_schema=False, + ) + filtered = cast(DataFrameT, filtered_graph._nodes) + if is_polars: + return cast(DataFrameT, filtered.rename({f"{alias}.{col}": col for col in node_frame.columns})) + return cast(DataFrameT, filtered.rename(columns={f"{alias}.{col}": col for col in node_frame.columns})) + + +def _connected_join_filter_node_frames_by_residuals( + base_graph: Plottable, + residual_map: Mapping[str, Sequence[str]], + frames: Mapping[str, DataFrameT], + node_col: str, + *, + engine: Engine, +) -> Dict[str, DataFrameT]: + """Apply each alias's post-join residuals to its materialized node frame.""" + out: Dict[str, DataFrameT] = dict(frames) + for alias, exprs in residual_map.items(): + if alias in out and exprs: + out[alias] = _connected_join_apply_node_residuals( + base_graph, out[alias], alias, exprs, node_col, engine=engine + ) + return out + + +def _connected_join_two_star_fast_grouped_count( + base_graph: Plottable, + plan: ConnectedMatchJoinPlan, + *, + engine: Engine, + cache_store: Optional[Dict[str, Any]] = None, +) -> Optional[DataFrameT]: + # Per-execution cache scope: a fresh store when none is threaded in, so intra-query reuse + # is preserved without ever persisting caches on the caller's Plottable (BLOCKER 1). + if cache_store is None: + cache_store = {} + if len(plan.pattern_chains) != 2 or len(plan.pattern_shared_node_aliases) != 1: + return None + if len(plan.pattern_shared_node_aliases[0]) != 1: + return None + shared_alias = plan.pattern_shared_node_aliases[0][0] + + parsed = [] + for pattern_chain in plan.pattern_chains: + if pattern_chain.where: + return None + ops = list(pattern_chain.chain) + if len(ops) != 3: + return None + start_op, edge_op, end_op = ops + if not isinstance(start_op, ASTNode) or not isinstance(edge_op, ASTEdge) or not isinstance(end_op, ASTNode): + return None + if getattr(start_op, "query", None) is not None or getattr(end_op, "query", None) is not None: + return None + if not _is_connected_fast_single_hop(edge_op): + return None + start_alias = getattr(start_op, "_name", None) + end_alias = getattr(end_op, "_name", None) + if start_alias != shared_alias or not isinstance(end_alias, str): + return None + parsed.append((start_op, edge_op, end_op, end_alias)) + + first_start, first_edge, first_end, first_end_alias = parsed[0] + second_start, second_edge, second_end, second_end_alias = parsed[1] + if first_end_alias == second_end_alias: + return None + + alias_targets: Dict[str, ASTObject] = { + shared_alias: first_start, + first_end_alias: first_end, + second_end_alias: second_end, + } + materialized_aliases = {shared_alias, first_end_alias, second_end_alias} + split = _connected_join_two_star_split_residuals(plan, alias_targets, materialized_aliases) + if split is None: + return None + residual_map, rest_chain = split + + post_ops = [op for op in rest_chain.chain if isinstance(op, ASTCall)] + if len(post_ops) not in (2, 3, 4): + return None + if post_ops[0].function != "with_" or post_ops[1].function != "group_by": + return None + suffix = post_ops[2:] + order_call: Optional[ASTCall] = None + limit_call: Optional[ASTCall] = None + select_call: Optional[ASTCall] = None + for op in suffix: + if op.function == "order_by" and order_call is None and limit_call is None and select_call is None: + order_call = op + elif op.function == "limit" and limit_call is None and select_call is None: + limit_call = op + elif op.function == "select" and select_call is None: + select_call = op + else: + return None + + with_items_raw = post_ops[0].params.get("items") + group_keys_raw = post_ops[1].params.get("keys") + aggs_raw = post_ops[1].params.get("aggregations") + if not isinstance(with_items_raw, list) or not isinstance(group_keys_raw, list) or not isinstance(aggs_raw, list): + return None + if len(aggs_raw) != 1: + return None + agg = aggs_raw[0] + if not isinstance(agg, (tuple, list)) or len(agg) != 3 or str(agg[1]).lower() != "count": + return None + agg_alias = str(agg[0]) + agg_input = agg[2] + + with_items: Dict[str, Any] = {} + for item in with_items_raw: + if not isinstance(item, (tuple, list)) or len(item) != 2 or not isinstance(item[0], str): + return None + with_items[item[0]] = item[1] + # count(p) over the shared alias now lowers `p` to its identity column + # `p.__gfql_node_id__` (see _connected_join_alias_identity_expr); accept either the + # bare shared alias or that identity form -- the fast count is shared-node multiplicity + # regardless, so the identity rewrite does not change the computation. + shared_identity = f"{shared_alias}.{NODE_IDENTITY_COLUMN}" + if not isinstance(agg_input, str) or with_items.get(agg_input) not in (shared_alias, shared_identity): + return None + + group_keys = [str(key) for key in group_keys_raw] + group_prop_refs: List[Tuple[str, str]] = [] + output_group_keys: List[str] = [] + for key in group_keys: + expr = with_items.get(key) + if key == "__cypher_group__" and expr == 1: + continue + prop_ref = _property_ref(expr, (second_end_alias,)) + if prop_ref is None: + return None + output_group_keys.append(key) + group_prop_refs.append((key, prop_ref[1])) + + order_keys: List[Tuple[str, bool]] = [] + if order_call is not None: + raw_order = order_call.params.get("keys") + if not isinstance(raw_order, list): + return None + available = set(output_group_keys) | {agg_alias} + for item in raw_order: + if not isinstance(item, (tuple, list)) or len(item) != 2 or not isinstance(item[0], str): + return None + if item[0] not in available: + return None + direction = str(item[1]).lower() + if direction not in {"asc", "ascending", "desc", "descending"}: + return None + order_keys.append((item[0], direction in {"desc", "descending"})) + + limit_value: Optional[int] = None + if limit_call is not None: + raw_limit = limit_call.params.get("value") + if not isinstance(raw_limit, int) or raw_limit < 0: + return None + limit_value = raw_limit + + select_items: Optional[List[Tuple[str, str]]] = None + if select_call is not None: + raw_select = select_call.params.get("items") + if not isinstance(raw_select, list): + return None + select_items = [] + available = set(output_group_keys) | {agg_alias} + for item in raw_select: + if not isinstance(item, (tuple, list)) or len(item) != 2 or not isinstance(item[0], str) or not isinstance(item[1], str): + return None + if item[0] not in available: + return None + select_items.append((item[0], item[1])) + + nodes_obj = getattr(base_graph, "_nodes", None) + edges_obj = getattr(base_graph, "_edges", None) + node_col = getattr(base_graph, "_node", None) + src_col = getattr(base_graph, "_source", None) + dst_col = getattr(base_graph, "_destination", None) + if nodes_obj is None or edges_obj is None or node_col is None or src_col is None or dst_col is None: + return None + node_col = str(node_col) + src_col = str(src_col) + dst_col = str(dst_col) + if node_col not in nodes_obj.columns or src_col not in edges_obj.columns or dst_col not in edges_obj.columns: + return None + + nodes_source = cast(DataFrameT, nodes_obj) + nodes = df_to_engine(nodes_source, engine) + edges = cast(DataFrameT, edges_obj) + + if engine in POLARS_ENGINES: + import polars as pl + from graphistry.compute.gfql.lazy.engine.polars.predicates import filter_by_dict_polars + + # Residual node filters (e.g. toLower(...)) must be applied to the materialized node + # frames, so residual queries use the direct (non-cached) path -- the id/count caches + # re-derive node sets from filter_dict alone and would drop the residual. + if isinstance(nodes_obj, (pl.DataFrame, pl.LazyFrame)) and not residual_map: + shared_ids = _connected_join_cached_node_ids( + base_graph, + nodes_source, + [cast(Optional[dict], first_start.filter_dict), cast(Optional[dict], second_start.filter_dict)], + node_col=node_col, + engine=engine, + cache_store=cache_store, + ) + first_leaf_ids = _connected_join_cached_node_ids( + base_graph, + nodes_source, + [cast(Optional[dict], first_end.filter_dict)], + node_col=node_col, + engine=engine, + cache_store=cache_store, + ) + second_leaf_ids = _connected_join_cached_node_ids( + base_graph, + nodes_source, + [cast(Optional[dict], second_end.filter_dict)], + node_col=node_col, + engine=engine, + cache_store=cache_store, + ) + first_leaf_nodes = _connected_join_cached_node_filter(base_graph, nodes_source, cast(Optional[dict], first_end.filter_dict), engine=engine, cache_store=cache_store) + second_leaf_nodes = _connected_join_cached_node_filter(base_graph, nodes_source, cast(Optional[dict], second_end.filter_dict), engine=engine, cache_store=cache_store) + if shared_ids is None or first_leaf_ids is None or second_leaf_ids is None: + return None + else: + shared_nodes = filter_by_dict_polars(nodes, cast(Optional[dict], first_start.filter_dict)) + shared_nodes = filter_by_dict_polars(shared_nodes, cast(Optional[dict], second_start.filter_dict)) + first_leaf_nodes = filter_by_dict_polars(nodes, cast(Optional[dict], first_end.filter_dict)) + second_leaf_nodes = filter_by_dict_polars(nodes, cast(Optional[dict], second_end.filter_dict)) + if residual_map: + _frames = _connected_join_filter_node_frames_by_residuals( + base_graph, + residual_map, + {shared_alias: shared_nodes, first_end_alias: first_leaf_nodes, second_end_alias: second_leaf_nodes}, + node_col, + engine=engine, + ) + shared_nodes, first_leaf_nodes, second_leaf_nodes = _frames[shared_alias], _frames[first_end_alias], _frames[second_end_alias] + shared_ids = shared_nodes.select(node_col).unique() + first_leaf_ids = first_leaf_nodes.select(node_col).unique() + second_leaf_ids = second_leaf_nodes.select(node_col).unique() + first_edges = _connected_join_cached_edge_filter(base_graph, edges, cast(Optional[dict], first_edge.edge_match), engine=engine, cache_store=cache_store) + second_edges = _connected_join_cached_edge_filter(base_graph, edges, cast(Optional[dict], second_edge.edge_match), engine=engine, cache_store=cache_store) + + for _, prop in group_prop_refs: + if prop not in second_leaf_nodes.columns: + return None + + def singleton_id(ids: Any) -> Any: + if not isinstance(ids, pl.DataFrame) or len(ids) != 1: + return _CACHE_MISSING + return ids.get_column(node_col)[0] + + first_leaf_singleton = singleton_id(first_leaf_ids) + second_leaf_singleton = singleton_id(second_leaf_ids) + + cached_left_counts: Optional[DataFrameT] = None + if first_leaf_singleton is not _CACHE_MISSING and not residual_map: + shared_ids_for_left = shared_ids.rename({node_col: shared_alias}) if node_col != shared_alias else shared_ids + cached_left_counts = _connected_join_cached_first_arm_shared_counts( + base_graph, + nodes_source, + edges, + first_edges, + shared_ids_for_left, + cast(Optional[dict], first_edge.edge_match), + cast(Optional[dict], first_start.filter_dict), + cast(Optional[dict], second_start.filter_dict), + first_leaf_singleton, + shared_alias=shared_alias, + src_col=src_col, + dst_col=dst_col, + engine=engine, + cache_store=cache_store, + ) + # Unreachable under the current lowering: _connected_join_cached_first_arm_shared_counts + # only returns None when a filter/edge/dst cache key is uncacheable, but every value + # that reaches this cached polars path is cacheable (scalar equality pushes as scalars; + # comparisons/toLower/ranges lower to residuals routed off the cached path). Kept as a + # defensive fallback; excluded from coverage since no legitimate query reaches it. + if cached_left_counts is None: # pragma: no cover + cached_left_counts = _connected_join_cached_singleton_dst_source_counts( + base_graph, + edges, + first_edges, + cast(Optional[dict], first_edge.edge_match), + first_leaf_singleton, + src_col=src_col, + dst_col=dst_col, + engine=engine, + cache_store=cache_store, + ) + if cached_left_counts is not None: + if shared_alias in cached_left_counts.columns: + left_counts = cached_left_counts + else: + left_counts = ( + cached_left_counts + .join(shared_ids, left_on=src_col, right_on=node_col, how="semi") + .rename({src_col: shared_alias}) + ) + elif first_leaf_singleton is _CACHE_MISSING: + left_edges = ( + first_edges + .join(shared_ids, left_on=src_col, right_on=node_col, how="semi") + .join(first_leaf_ids, left_on=dst_col, right_on=node_col, how="semi") + ) + left_counts = ( + left_edges + .group_by(src_col) + .len("__left_count__") + .rename({src_col: shared_alias}) + ) + else: + left_edges = ( + first_edges + .filter(pl.col(dst_col) == first_leaf_singleton) + .join(shared_ids, left_on=src_col, right_on=node_col, how="semi") + ) + left_counts = ( + left_edges + .group_by(src_col) + .len("__left_count__") + .rename({src_col: shared_alias}) + ) + cached_right_rows = None if residual_map else _connected_join_cached_second_arm_group_rows( + base_graph, + nodes_source, + edges, + second_edges, + shared_ids, + second_leaf_ids, + second_leaf_nodes, + cast(Optional[dict], first_start.filter_dict), + cast(Optional[dict], second_start.filter_dict), + cast(Optional[dict], second_end.filter_dict), + cast(Optional[dict], second_edge.edge_match), + second_leaf_singleton, + group_prop_refs, + output_group_keys, + node_col=node_col, + src_col=src_col, + dst_col=dst_col, + engine=engine, + cache_store=cache_store, + ) + if cached_right_rows is not None: + right_rows = cached_right_rows if src_col == shared_alias else cached_right_rows.rename({src_col: shared_alias}) + else: + if second_leaf_singleton is _CACHE_MISSING: + right_base = ( + second_edges + .join(shared_ids, left_on=src_col, right_on=node_col, how="semi") + .join(second_leaf_ids, left_on=dst_col, right_on=node_col, how="semi") + ) + else: + right_base = ( + second_edges + .filter(pl.col(dst_col) == second_leaf_singleton) + .join(shared_ids, left_on=src_col, right_on=node_col, how="semi") + ) + if group_prop_refs: + lookup_key = "__gfql_fast_second_leaf_id__" + lookup_exprs = [pl.col(node_col).alias(lookup_key)] + [pl.col(prop).alias(out_col) for out_col, prop in group_prop_refs] + # Dedup by node identity to match the pandas drop_duplicates(subset=[node_col]); + # a duplicate node row would otherwise multiply the join and over-count. + second_lookup = second_leaf_nodes.select(lookup_exprs).unique(subset=[lookup_key]) + right_base = right_base.join(second_lookup, left_on=dst_col, right_on=lookup_key, how="inner") + right_rows = right_base.select([pl.col(src_col).alias(shared_alias)] + [pl.col(key) for key in output_group_keys]) + if not output_group_keys and len(left_counts) > 0 and bool(left_counts.select((pl.col("__left_count__") == 1).all()).item()): + matched_rows = right_rows.join(left_counts.select(shared_alias), on=shared_alias, how="semi") + out_df = matched_rows.select(pl.len().cast(pl.Int64).alias(agg_alias)) + else: + joined = right_rows.join(left_counts, on=shared_alias, how="inner") + if len(joined) == 0: + return cast(DataFrameT, joined.select([])) + if output_group_keys: + out_df = joined.group_by(output_group_keys, maintain_order=True).agg(pl.col("__left_count__").sum().cast(pl.Int64).alias(agg_alias)) + else: + out_df = joined.select(pl.col("__left_count__").sum().cast(pl.Int64).alias(agg_alias)) + if order_keys: + # openCypher orders NULL as the largest value: ASC -> nulls last, DESC -> nulls + # first. Polars defaults to nulls-first, which flips WHICH ROW an ORDER BY ... LIMIT + # returns, so pin nulls_last per key. + out_df = out_df.sort( + [key for key, _ in order_keys], + descending=[desc for _, desc in order_keys], + nulls_last=[not desc for _, desc in order_keys], + ) + if limit_value is not None: + out_df = out_df.head(limit_value) + if select_items is not None: + out_df = out_df.select([pl.col(src).alias(dst) for src, dst in select_items]) + return cast(DataFrameT, out_df) + + filter_engine = EngineAbstract(engine.value) + shared_nodes = filter_by_dict(nodes, cast(Optional[dict], first_start.filter_dict), engine=filter_engine) + shared_nodes = filter_by_dict(shared_nodes, cast(Optional[dict], second_start.filter_dict), engine=filter_engine) + first_leaf_nodes = filter_by_dict(nodes, cast(Optional[dict], first_end.filter_dict), engine=filter_engine) + second_leaf_nodes = filter_by_dict(nodes, cast(Optional[dict], second_end.filter_dict), engine=filter_engine) + if residual_map: + _frames = _connected_join_filter_node_frames_by_residuals( + base_graph, + residual_map, + {shared_alias: shared_nodes, first_end_alias: first_leaf_nodes, second_end_alias: second_leaf_nodes}, + node_col, + engine=engine, + ) + shared_nodes, first_leaf_nodes, second_leaf_nodes = _frames[shared_alias], _frames[first_end_alias], _frames[second_end_alias] + first_edges = _connected_join_cached_edge_filter(base_graph, edges, cast(Optional[dict], first_edge.edge_match), engine=engine, cache_store=cache_store) + second_edges = _connected_join_cached_edge_filter(base_graph, edges, cast(Optional[dict], second_edge.edge_match), engine=engine, cache_store=cache_store) + + shared_ids = shared_nodes[node_col].drop_duplicates() + first_leaf_ids = first_leaf_nodes[node_col].drop_duplicates() + second_leaf_ids = second_leaf_nodes[node_col].drop_duplicates() + for _, prop in group_prop_refs: + if prop not in second_leaf_nodes.columns: + return None + + left_edges = first_edges[first_edges[src_col].isin(shared_ids) & first_edges[dst_col].isin(first_leaf_ids)] + left_counts = left_edges.groupby(src_col, sort=False).size().reset_index(name="__left_count__").rename(columns={src_col: shared_alias}) + right_base = second_edges[second_edges[src_col].isin(shared_ids) & second_edges[dst_col].isin(second_leaf_ids)] + if group_prop_refs: + lookup_key = "__gfql_fast_second_leaf_id__" + prop_cols = [] + for _, prop in group_prop_refs: + if prop not in prop_cols: + prop_cols.append(prop) + second_lookup = second_leaf_nodes[[node_col] + prop_cols].drop_duplicates(subset=[node_col]).rename(columns={node_col: lookup_key}) + for out_col, prop in group_prop_refs: + second_lookup[out_col] = second_lookup[prop] + right_base = right_base.merge(second_lookup[[lookup_key] + output_group_keys], left_on=dst_col, right_on=lookup_key, how="inner") + right_rows = right_base[[src_col] + output_group_keys].rename(columns={src_col: shared_alias}) + joined = right_rows.merge(left_counts, on=shared_alias, how="inner") + if len(joined) == 0: + return df_cons(engine)() + if output_group_keys: + out_df = joined.groupby(output_group_keys, sort=False, dropna=False)["__left_count__"].sum().reset_index(name=agg_alias) + else: + out_df = pd.DataFrame({agg_alias: [int(joined["__left_count__"].sum())]}) + if order_keys: + out_df = cast(DataFrameT, out_df.sort_values(by=[key for key, _ in order_keys], ascending=[not desc for _, desc in order_keys])) + if limit_value is not None: + out_df = cast(DataFrameT, out_df.head(limit_value)) + if select_items is not None: + out_df = out_df[[src for src, _ in select_items]].rename(columns={src: dst for src, dst in select_items}) + return df_to_engine(out_df.reset_index(drop=True), engine) + + +def _connected_join_two_star_fast_rows( + base_graph: Plottable, + plan: ConnectedMatchJoinPlan, + *, + engine: Engine, + cache_store: Optional[Dict[str, Any]] = None, +) -> Optional[DataFrameT]: + """Fast rows for T1 two-star connected joins. + + Supported shape: ``(p)-[r1]->(i), (p)-[r2]->(c)`` where both arms are fixed + forward one-hop patterns, share the same start node alias, have no residual + per-arm row filters, and downstream projections need at most properties from + the second leaf alias. The returned frame preserves row multiplicity, then + the normal row pipeline handles RETURN/GROUP/ORDER/LIMIT. + """ + # Per-execution cache scope (see _connected_join_two_star_fast_grouped_count); never + # persisted on the caller's Plottable. + if cache_store is None: + cache_store = {} + if len(plan.pattern_chains) != 2 or len(plan.pattern_shared_node_aliases) != 1: + return None + if len(plan.pattern_shared_node_aliases[0]) != 1: + return None + shared_alias = plan.pattern_shared_node_aliases[0][0] + + parsed = [] + for pattern_chain in plan.pattern_chains: + if pattern_chain.where: + return None + ops = list(pattern_chain.chain) + if len(ops) != 3: + return None + start_op, edge_op, end_op = ops + if not isinstance(start_op, ASTNode) or not isinstance(edge_op, ASTEdge) or not isinstance(end_op, ASTNode): + return None + if getattr(start_op, "query", None) is not None or getattr(end_op, "query", None) is not None: + return None + if not _is_connected_fast_single_hop(edge_op): + return None + start_alias = getattr(start_op, "_name", None) + end_alias = getattr(end_op, "_name", None) + if start_alias != shared_alias or not isinstance(end_alias, str): + return None + parsed.append((start_op, edge_op, end_op, end_alias)) + + first_start, first_edge, first_end, first_end_alias = parsed[0] + second_start, second_edge, second_end, second_end_alias = parsed[1] + if first_end_alias == second_end_alias: + return None + + attach_aliases = plan.pattern_attach_prop_aliases + if not attach_aliases or len(attach_aliases) != 2: + return None + normalized_attach_aliases: List[Tuple[str, ...]] = [] + for aliases in attach_aliases: + if aliases is None: + return None + normalized_attach_aliases.append(aliases) + needed_attach_aliases = {alias for aliases in normalized_attach_aliases for alias in aliases} + if any(alias != second_end_alias for alias in needed_attach_aliases): + return None + second_props_needed = _connected_join_post_property_columns(plan, second_end_alias) + # A bare aggregate over the second leaf (count(c) / count(DISTINCT c)) lowers `c` to its + # identity column c.__gfql_node_id__, which is NOT a materializable node property here -- + # fast_rows would drop it and post_join's count(c.__gfql_node_id__) would dereference a + # missing column. Decline to the (correct) slow path in that case. + if NODE_IDENTITY_COLUMN in second_props_needed: + return None + if second_end_alias in needed_attach_aliases and not second_props_needed: + return None + + nodes_obj = getattr(base_graph, "_nodes", None) + edges_obj = getattr(base_graph, "_edges", None) + node_col = getattr(base_graph, "_node", None) + src_col = getattr(base_graph, "_source", None) + dst_col = getattr(base_graph, "_destination", None) + if nodes_obj is None or edges_obj is None or node_col is None or src_col is None or dst_col is None: + return None + node_col = str(node_col) + src_col = str(src_col) + dst_col = str(dst_col) + if node_col not in nodes_obj.columns or src_col not in edges_obj.columns or dst_col not in edges_obj.columns: + return None + + nodes = df_to_engine(cast(DataFrameT, nodes_obj), engine) + edges = cast(DataFrameT, edges_obj) + + if engine in POLARS_ENGINES: + import polars as pl + from graphistry.compute.gfql.lazy.engine.polars.predicates import filter_by_dict_polars + + shared_nodes = filter_by_dict_polars(nodes, cast(Optional[dict], first_start.filter_dict)) + shared_nodes = filter_by_dict_polars(shared_nodes, cast(Optional[dict], second_start.filter_dict)) + first_leaf_nodes = filter_by_dict_polars(nodes, cast(Optional[dict], first_end.filter_dict)) + second_leaf_nodes = filter_by_dict_polars(nodes, cast(Optional[dict], second_end.filter_dict)) + first_edges = _connected_join_cached_edge_filter(base_graph, edges, cast(Optional[dict], first_edge.edge_match), engine=engine, cache_store=cache_store) + second_edges = _connected_join_cached_edge_filter(base_graph, edges, cast(Optional[dict], second_edge.edge_match), engine=engine, cache_store=cache_store) + + shared_ids = shared_nodes.select(node_col).unique() + first_leaf_ids = first_leaf_nodes.select(node_col).unique() + second_leaf_ids = second_leaf_nodes.select(node_col).unique() + second_props = [col for col in second_props_needed if col in second_leaf_nodes.columns] + + left_rows = ( + first_edges + .join(shared_ids, left_on=src_col, right_on=node_col, how="semi") + .join(first_leaf_ids, left_on=dst_col, right_on=node_col, how="semi") + .select(pl.col(src_col).alias(shared_alias)) + ) + right_base = ( + second_edges + .join(shared_ids, left_on=src_col, right_on=node_col, how="semi") + .join(second_leaf_ids, left_on=dst_col, right_on=node_col, how="semi") + ) + if second_props: + second_lookup_key = "__gfql_fast_second_leaf_id__" + # Dedup by node identity (matches pandas drop_duplicates(subset=[node_col])); a + # duplicate node row would otherwise multiply the join and inflate row multiplicity. + second_lookup = second_leaf_nodes.select([node_col] + second_props).unique(subset=[node_col]).rename({node_col: second_lookup_key}) + second_lookup = second_lookup.rename({col: f"{second_end_alias}.{col}" for col in second_props}) + right_base = right_base.join(second_lookup, left_on=dst_col, right_on=second_lookup_key, how="inner") + right_select = [pl.col(src_col).alias(shared_alias)] + [pl.col(f"{second_end_alias}.{col}") for col in second_props] + right_rows = right_base.select(right_select) + return cast(DataFrameT, left_rows.join(right_rows, on=shared_alias, how="inner")) + + filter_engine = EngineAbstract(engine.value) + shared_nodes = filter_by_dict(nodes, cast(Optional[dict], first_start.filter_dict), engine=filter_engine) + shared_nodes = filter_by_dict(shared_nodes, cast(Optional[dict], second_start.filter_dict), engine=filter_engine) + first_leaf_nodes = filter_by_dict(nodes, cast(Optional[dict], first_end.filter_dict), engine=filter_engine) + second_leaf_nodes = filter_by_dict(nodes, cast(Optional[dict], second_end.filter_dict), engine=filter_engine) + first_edges = _connected_join_cached_edge_filter(base_graph, edges, cast(Optional[dict], first_edge.edge_match), engine=engine, cache_store=cache_store) + second_edges = _connected_join_cached_edge_filter(base_graph, edges, cast(Optional[dict], second_edge.edge_match), engine=engine, cache_store=cache_store) + + shared_ids = shared_nodes[node_col].drop_duplicates() + first_leaf_ids = first_leaf_nodes[node_col].drop_duplicates() + second_leaf_ids = second_leaf_nodes[node_col].drop_duplicates() + second_props = [col for col in second_props_needed if col in second_leaf_nodes.columns] + + left_edges = first_edges[first_edges[src_col].isin(shared_ids) & first_edges[dst_col].isin(first_leaf_ids)] + left_rows = left_edges[[src_col]].rename(columns={src_col: shared_alias}) + right_base = second_edges[second_edges[src_col].isin(shared_ids) & second_edges[dst_col].isin(second_leaf_ids)] + if second_props: + second_lookup_key = "__gfql_fast_second_leaf_id__" + second_lookup_cols = [node_col] + second_props + second_lookup = second_leaf_nodes[second_lookup_cols].drop_duplicates(subset=[node_col]).rename( + columns={ + node_col: second_lookup_key, + **{col: f"{second_end_alias}.{col}" for col in second_props}, + } + ) + right_base = right_base.merge(second_lookup, left_on=dst_col, right_on=second_lookup_key, how="inner") + right_cols = [src_col] + [f"{second_end_alias}.{col}" for col in second_props] + right_rows = right_base[right_cols].rename(columns={src_col: shared_alias}) + return cast(DataFrameT, left_rows.merge(right_rows, on=shared_alias, how="inner")) + def _apply_connected_match_join( base_graph: Plottable, plan: ConnectedMatchJoinPlan, @@ -480,17 +1592,47 @@ def _apply_connected_match_join( policy: Optional[PolicyDict], context: ExecutionContext, ) -> Plottable: - from graphistry.compute.ast import ASTCall, serialize_binding_ops + from graphistry.compute.ast import ASTCall, ASTNode as _ASTNode, serialize_binding_ops requested_engine = resolve_engine(cast(Any, engine), base_graph) dispatch_engine: Union[EngineAbstract, str] = engine df_ctor = df_cons(requested_engine) node_col = getattr(base_graph, "_node", "id") + # One cache scope per connected-join execution: shared across the fast paths but never + # persisted on the caller's Plottable, so a second gfql() after an in-place mutation + # recomputes instead of returning a stale cached answer (BLOCKER 1). + cache_store: Dict[str, Any] = {} + + fast_grouped_count = _connected_join_two_star_fast_grouped_count(base_graph, plan, engine=requested_engine, cache_store=cache_store) + if fast_grouped_count is not None: + out = base_graph.bind() + out._nodes = fast_grouped_count + out._edges = df_ctor() + return out + + fast_rows = _connected_join_two_star_fast_rows(base_graph, plan, engine=requested_engine, cache_store=cache_store) + if fast_rows is not None: + if len(fast_rows) == 0: + out = base_graph.bind() + out._nodes = df_ctor() + out._edges = df_ctor() + return out + fast_rows = _joined_hidden_scalar_columns(fast_rows) + fast_rows = _joined_alias_columns(fast_rows) + joined_plottable = base_graph.bind() + joined_plottable._nodes = fast_rows + joined_plottable._edges = df_ctor() + return _chain_dispatch(joined_plottable, plan.post_join_chain, dispatch_engine, policy, context) + joined_rows: Optional[DataFrameT] = None + pattern_attach_prop_aliases = plan.pattern_attach_prop_aliases or tuple(None for _ in plan.pattern_chains) for idx, pattern_chain in enumerate(plan.pattern_chains): + rows_params: Dict[str, Any] = {"binding_ops": serialize_binding_ops(pattern_chain.chain)} + if idx < len(pattern_attach_prop_aliases) and pattern_attach_prop_aliases[idx] is not None: + rows_params["attach_prop_aliases"] = list(cast(Tuple[str, ...], pattern_attach_prop_aliases[idx])) with_rows = Chain( - list(pattern_chain.chain) + [ASTCall("rows", {"binding_ops": serialize_binding_ops(pattern_chain.chain)})], + list(pattern_chain.chain) + [ASTCall("rows", rows_params)], where=pattern_chain.where, ) pattern_result = _chain_dispatch(base_graph, with_rows, dispatch_engine, policy, context) @@ -502,17 +1644,33 @@ def _apply_connected_match_join( return out # The rows op now emits the full binding schema even at 0 rows (#25), so an emptied # pattern carries its columns and flows through post_join_chain -- which is where the - # aggregate RETURN lives. Short-circuiting here dropped that column. - pattern_rows = cast(DataFrameT, pattern_rows[_binding_join_columns(pattern_rows)]) + # aggregate RETURN lives; short-circuiting here dropped that column. Beyond the join + # columns, rung-3 execution also keeps the bare node-alias columns (e.g. `count(i)` + # needs the `i` binding column downstream in post_join_chain). + node_aliases = [ + cast(str, getattr(op, "_name")) + for op in pattern_chain.chain + if isinstance(op, _ASTNode) and isinstance(getattr(op, "_name", None), str) + ] + keep_binding_columns = _binding_join_columns(pattern_rows) + [ + alias for alias in node_aliases if alias in pattern_rows.columns + ] + pattern_rows = cast(DataFrameT, pattern_rows[keep_binding_columns]) if joined_rows is None: joined_rows = pattern_rows continue shared_aliases = plan.pattern_shared_node_aliases[idx - 1] join_cols = [ - f"{alias}.{node_col}" + alias for alias in shared_aliases - if f"{alias}.{node_col}" in joined_rows.columns and f"{alias}.{node_col}" in pattern_rows.columns + if alias in joined_rows.columns and alias in pattern_rows.columns ] + if not join_cols: + join_cols = [ + f"{alias}.{node_col}" + for alias in shared_aliases + if f"{alias}.{node_col}" in joined_rows.columns and f"{alias}.{node_col}" in pattern_rows.columns + ] if not join_cols: raise GFQLValidationError( ErrorCode.E108, diff --git a/graphistry/tests/compute/gfql/coverage_baselines/ci-pandas-py3.12.json b/graphistry/tests/compute/gfql/coverage_baselines/ci-pandas-py3.12.json index 64d089086f..a47db2309c 100644 --- a/graphistry/tests/compute/gfql/coverage_baselines/ci-pandas-py3.12.json +++ b/graphistry/tests/compute/gfql/coverage_baselines/ci-pandas-py3.12.json @@ -63,6 +63,6 @@ "graphistry/compute/gfql/temporal/truncation.py": 76.92, "graphistry/compute/gfql/temporal/values.py": 88.0, "graphistry/compute/gfql/temporal_text.py": 61.11, - "graphistry/compute/gfql_unified.py": 78.0 + "graphistry/compute/gfql_unified.py": 67.5 } } diff --git a/graphistry/tests/compute/gfql/cypher/test_lowering.py b/graphistry/tests/compute/gfql/cypher/test_lowering.py index 6a11c5cadb..d060710516 100644 --- a/graphistry/tests/compute/gfql/cypher/test_lowering.py +++ b/graphistry/tests/compute/gfql/cypher/test_lowering.py @@ -7,7 +7,10 @@ import graphistry from typing import Any, Callable, Dict, List, Optional, Tuple, cast +from graphistry.Engine import Engine + from graphistry.compute.ast import ASTCall, ASTNode, ASTEdge, ASTEdgeForward, ASTEdgeReverse, ASTEdgeUndirected +from graphistry.compute.chain import Chain from graphistry.compute.exceptions import ErrorCode, GFQLSyntaxError, GFQLTypeError, GFQLValidationError from graphistry.compute.predicates.is_in import IsIn from graphistry.compute.gfql.same_path_types import NODE_IDENTITY_COLUMN, col, compare @@ -39,7 +42,15 @@ from graphistry.plugins.networkx.policy import NETWORKX_SCIPY_EXTRA_REQUIREMENTS, NETWORKX_VERSION_SPEC, SCIPY_VERSION_SPEC from graphistry.compute.gfql.cypher.ast import ExpressionText, OrderByClause, OrderItem, ReturnClause, ReturnItem, SourceSpan from graphistry.compute.gfql.cypher.lowering import CompiledCypherExecutionExtras, CompiledCypherGraphQuery, compile_cypher_query -from graphistry.compute.gfql.cypher.lowering import _logical_plan_route_for_query +from graphistry.compute.gfql.cypher.lowering import ConnectedMatchJoinPlan, _logical_plan_route_for_query +from graphistry.compute.gfql_unified import ( + _connected_join_cached_edge_filter, + _connected_join_cached_singleton_dst_source_counts, + _connected_join_post_property_columns, + _connected_join_two_star_fast_grouped_count, + _connected_join_two_star_fast_rows, + _connected_join_two_star_split_residuals, +) from graphistry.compute.gfql.frontends.cypher.binder import FrontendBinder from graphistry.compute.gfql.ir.bound_ir import BoundIR, BoundQueryPart, SemanticTable from graphistry.compute.gfql.ir.compilation import PlanContext @@ -14936,7 +14947,6 @@ def test_issue_1273_multi_source_grouped_aggregate(agg: str, expected: list) -> assert result._nodes.to_dict(orient="records") == expected - def test_issue_1712_connected_comma_pattern_where_intersects() -> None: """#1712: a connected comma-pattern sharing a node alias with a WHERE on a leaf alias must intersect both patterns (the WHERE was silently dropped on the @@ -15964,7 +15974,9 @@ def test_t1_connected_comma_grouped_projection_attaches_only_city_props() -> Non plan = _compiled_connected_join_plan(query) assert "where_rows" in _post_join_functions(query) - assert plan.pattern_attach_prop_aliases == (("i",), ("c",)) + # `p` is attached because count(p) lowers `p` to its `p.__gfql_node_id__` identity + # (connected-plan #1729); the toLower(i.interest) residual attaches `i`. + assert plan.pattern_attach_prop_aliases == (("i", "p"), ("c", "p")) def test_t1_connected_comma_pushes_q5_literal_filters_and_retains_lower_residual() -> None: @@ -16006,7 +16018,8 @@ def test_t1_connected_comma_pushes_reversed_single_alias_filters_before_join() - plan = _compiled_connected_join_plan(query) assert "where_rows" in _post_join_functions(query) - assert plan.pattern_attach_prop_aliases == (("i",), ()) + # count(p) attaches `p` (identity rewrite, #1729); toLower(i.interest) attaches `i`. + assert plan.pattern_attach_prop_aliases == (("i", "p"), ("p",)) filters_by_alias = _compiled_connected_join_filters(query) assert not any("interest" in entry.get("i", {}) for entry in filters_by_alias) @@ -16028,7 +16041,8 @@ def test_t1_connected_comma_retains_lower_property_plain_lowercase_literal() -> plan = _compiled_connected_join_plan(query) assert "where_rows" in _post_join_functions(query) - assert plan.pattern_attach_prop_aliases == (("i",), ()) + # count(p) attaches `p` (identity rewrite, #1729); toLower(i.interest) attaches `i`. + assert plan.pattern_attach_prop_aliases == (("i", "p"), ("p",)) def test_t1_connected_comma_retains_reversed_uppercase_plain_literal_residual() -> None: @@ -16045,7 +16059,8 @@ def test_t1_connected_comma_retains_reversed_uppercase_plain_literal_residual() plan = _compiled_connected_join_plan(query) assert "where_rows" in _post_join_functions(query) - assert plan.pattern_attach_prop_aliases == (("i",), ()) + # count(p) attaches `p` (identity rewrite, #1729); toLower(i.interest) attaches `i`. + assert plan.pattern_attach_prop_aliases == (("i", "p"), ("p",)) @pytest.mark.parametrize( @@ -16094,7 +16109,8 @@ def test_t1_connected_comma_retains_unicode_lower_equality_residual(where_expr: plan = _compiled_connected_join_plan(query) assert "where_rows" in _post_join_functions(query) - assert plan.pattern_attach_prop_aliases == (("i",), ()) + # count(p) attaches `p` (identity rewrite, #1729); toLower(i.interest) attaches `i`. + assert plan.pattern_attach_prop_aliases == (("i", "p"), ("p",)) def test_t1_connected_comma_pushes_q7_range_filters_before_join() -> None: @@ -16112,7 +16128,8 @@ def test_t1_connected_comma_pushes_q7_range_filters_before_join() -> None: plan = _compiled_connected_join_plan(query) assert "where_rows" in _post_join_functions(query) - assert plan.pattern_attach_prop_aliases == (("i",), ("c",)) + # count(p) attaches `p` (identity rewrite, #1729); toLower(i.interest) attaches `i`. + assert plan.pattern_attach_prop_aliases == (("i", "p"), ("c", "p")) filters_by_alias = _compiled_connected_join_filters(query) assert any("age" in entry.get("p", {}) for entry in filters_by_alias) @@ -16120,6 +16137,608 @@ def test_t1_connected_comma_pushes_q7_range_filters_before_join() -> None: assert any(entry.get("c", {}).get("country") == "France" for entry in filters_by_alias) +@pytest.mark.parametrize("engine", ["pandas", "polars"]) +def test_t1_connected_comma_q5_global_count_runs_on_pandas_and_polars(engine: str) -> None: + if engine == "polars": + pytest.importorskip("polars") + query = ( + "MATCH (p {node_type:'Person'})-[{rel:'HAS_INTEREST'}]->(i {node_type:'Interest'}), " + "(p)-[{rel:'LIVES_IN'}]->(c {node_type:'City'}) " + "WHERE toLower(i.interest) = toLower('fine dining') " + "AND toLower(p.gender) = toLower('male') " + "AND c.city = 'London' AND c.country = 'United Kingdom' " + "RETURN count(p) AS numPersons" + ) + + result = _mk_graph_benchmark_t1_shape_graph().gfql(query, engine=engine) + assert _to_pandas_df(result._nodes).to_dict(orient="records") == [{"numPersons": 1}] + + +@pytest.mark.parametrize("engine", ["pandas", "polars"]) +def test_t1_connected_comma_q6_q7_grouped_count_runs_on_pandas_and_polars(engine: str) -> None: + if engine == "polars": + pytest.importorskip("polars") + query = ( + "MATCH (p {node_type:'Person'})-[{rel:'HAS_INTEREST'}]->(i {node_type:'Interest'}), " + "(p)-[{rel:'LIVES_IN'}]->(c {node_type:'City'}) " + "WHERE toLower(i.interest) = toLower('fine dining') " + "AND p.age >= 23 AND p.age <= 30 " + "AND c.country = 'United Kingdom' " + "RETURN count(p) AS numPersons, c.state AS state, c.country AS country " + "ORDER BY state" + ) + + result = _mk_graph_benchmark_t1_shape_graph().gfql(query, engine=engine) + assert _to_pandas_df(result._nodes).to_dict(orient="records") == [ + {"numPersons": 2, "state": "England", "country": "United Kingdom"} + ] + + +@pytest.mark.parametrize("engine", ["pandas", "polars"]) +def test_t1_connected_comma_two_star_fast_path_preserves_multiplicity(engine: str) -> None: + if engine == "polars": + pytest.importorskip("polars") + nodes = pd.DataFrame({ + "id": [0, 10, 20], + "node_type": ["Person", "City", "Interest"], + "gender": ["male", None, None], + "city": [None, "London", None], + "country": [None, "United Kingdom", None], + "state": [None, "England", None], + "interest": [None, None, "Fine Dining"], + }) + edges = pd.DataFrame({ + "s": [0, 0, 0, 0], + "d": [20, 20, 10, 10], + "rel": ["HAS_INTEREST", "HAS_INTEREST", "LIVES_IN", "LIVES_IN"], + }) + query = ( + "MATCH (p {node_type:'Person'})-[{rel:'HAS_INTEREST'}]->(i {node_type:'Interest'}), " + "(p)-[{rel:'LIVES_IN'}]->(c {node_type:'City'}) " + "WHERE toLower(i.interest) = toLower('fine dining') " + "AND toLower(p.gender) = toLower('male') " + "AND c.city = 'London' AND c.country = 'United Kingdom' " + "RETURN count(p) AS numPersons" + ) + + result = _mk_graph(nodes, edges).gfql(query, engine=engine) + assert _to_pandas_df(result._nodes).to_dict(orient="records") == [{"numPersons": 4}] + + +@pytest.mark.parametrize("engine", ["pandas", "polars"]) +def test_t1_connected_comma_two_star_fast_path_groups_second_leaf_props(engine: str) -> None: + if engine == "polars": + pytest.importorskip("polars") + nodes = pd.DataFrame({ + "id": [0, 1, 10, 11, 20], + "node_type": ["Person", "Person", "City", "City", "Interest"], + "gender": ["female", "female", None, None, None], + "city": [None, None, "London", "Bristol", None], + "country": [None, None, "United Kingdom", "United Kingdom", None], + "state": [None, None, "England", "England", None], + "interest": [None, None, None, None, "tennis"], + }) + edges = pd.DataFrame({ + "s": [0, 1, 0, 1], + "d": [20, 20, 10, 11], + "rel": ["HAS_INTEREST", "HAS_INTEREST", "LIVES_IN", "LIVES_IN"], + }) + query = ( + "MATCH (p {node_type:'Person'})-[{rel:'HAS_INTEREST'}]->(i {node_type:'Interest'}), " + "(p)-[{rel:'LIVES_IN'}]->(c {node_type:'City'}) " + "WHERE toLower(i.interest) = toLower('tennis') AND toLower(p.gender) = toLower('female') " + "RETURN count(p) AS numPersons, c.city AS city, c.country AS country " + "ORDER BY city" + ) + + result = _mk_graph(nodes, edges).gfql(query, engine=engine) + assert _to_pandas_df(result._nodes).to_dict(orient="records") == [ + {"numPersons": 1, "city": "Bristol", "country": "United Kingdom"}, + {"numPersons": 1, "city": "London", "country": "United Kingdom"}, + ] + + +@pytest.mark.parametrize("engine", ["pandas", "polars"]) +def test_t6_connected_comma_two_star_direct_count_preserves_multiplicity(engine: str) -> None: + if engine == "polars": + pytest.importorskip("polars") + nodes = pd.DataFrame({ + "id": [0, 10, 20], + "node_type": ["Person", "City", "Interest"], + "gender": ["male", None, None], + "city": [None, "London", None], + "country": [None, "United Kingdom", None], + "interest": [None, None, "Fine Dining"], + }) + edges = pd.DataFrame({ + "s": [0, 0, 0, 0], + "d": [20, 20, 10, 10], + "rel": ["HAS_INTEREST", "HAS_INTEREST", "LIVES_IN", "LIVES_IN"], + }) + query = ( + "MATCH (p {node_type:'Person'})-[{rel:'HAS_INTEREST'}]->(i {node_type:'Interest'}), " + "(p)-[{rel:'LIVES_IN'}]->(c {node_type:'City'}) " + "WHERE toLower(i.interest) = toLower('fine dining') " + "AND toLower(p.gender) = toLower('male') " + "AND c.city = 'London' AND c.country = 'United Kingdom' " + "RETURN count(p) AS numPersons" + ) + graph = _mk_graph(nodes, edges) + plan = _compiled_connected_join_plan(query) + + direct = _connected_join_two_star_fast_grouped_count(graph, plan, engine=Engine(engine)) + assert direct is not None + assert _to_pandas_df(direct).to_dict(orient="records") == [{"numPersons": 4}] + + result = graph.gfql(query, engine=engine) + assert _to_pandas_df(result._nodes).to_dict(orient="records") == [{"numPersons": 4}] + + +@pytest.mark.parametrize("engine", ["pandas", "polars"]) +def test_t6_connected_comma_two_star_direct_grouped_count_orders_and_limits(engine: str) -> None: + if engine == "polars": + pytest.importorskip("polars") + nodes = pd.DataFrame({ + "id": [0, 1, 10, 11, 20], + "node_type": ["Person", "Person", "City", "City", "Interest"], + "gender": ["female", "female", None, None, None], + "city": [None, None, "London", "Bristol", None], + "country": [None, None, "United Kingdom", "United Kingdom", None], + "state": [None, None, "England", "England", None], + "interest": [None, None, None, None, "tennis"], + }) + edges = pd.DataFrame({ + "s": [0, 0, 1, 0, 1], + "d": [20, 20, 20, 10, 11], + "rel": ["HAS_INTEREST", "HAS_INTEREST", "HAS_INTEREST", "LIVES_IN", "LIVES_IN"], + }) + query = ( + "MATCH (p {node_type:'Person'})-[{rel:'HAS_INTEREST'}]->(i {node_type:'Interest'}), " + "(p)-[{rel:'LIVES_IN'}]->(c {node_type:'City'}) " + "WHERE toLower(i.interest) = toLower('tennis') AND toLower(p.gender) = toLower('female') " + "RETURN count(p) AS numPersons, c.city AS city, c.country AS country " + "ORDER BY numPersons DESC, city LIMIT 1" + ) + graph = _mk_graph(nodes, edges) + plan = _compiled_connected_join_plan(query) + + direct = _connected_join_two_star_fast_grouped_count(graph, plan, engine=Engine(engine)) + assert direct is not None + assert _to_pandas_df(direct).to_dict(orient="records") == [ + {"city": "London", "country": "United Kingdom", "numPersons": 2} + ] + + result = graph.gfql(query, engine=engine) + assert _to_pandas_df(result._nodes).to_dict(orient="records") == [ + {"city": "London", "country": "United Kingdom", "numPersons": 2} + ] + + +def test_t9_connected_comma_two_star_direct_polars_native_reuses_node_filter_cache() -> None: + pl = pytest.importorskip("polars") + nodes = pd.DataFrame({ + "id": [0, 1, 10, 11, 20], + "node_type": ["Person", "Person", "City", "City", "Interest"], + "gender": ["female", "female", None, None, None], + "city": [None, None, "London", "Bristol", None], + "country": [None, None, "United Kingdom", "United Kingdom", None], + "interest": [None, None, None, None, "tennis"], + }) + edges = pd.DataFrame({ + "s": [0, 0, 1, 0, 1], + "d": [20, 20, 20, 10, 11], + "rel": ["HAS_INTEREST", "HAS_INTEREST", "HAS_INTEREST", "LIVES_IN", "LIVES_IN"], + }) + query = ( + "MATCH (p {node_type:'Person'})-[{rel:'HAS_INTEREST'}]->(i {node_type:'Interest'}), " + "(p)-[{rel:'LIVES_IN'}]->(c {node_type:'City'}) " + # Plain-equality filters push into filter_dict so this exercises the polars-native + # CACHED path (the point of this test). Under connected-plan #1729, toLower(...) lowers + # to a post-join where_rows residual that deliberately bypasses the id/count caches + # (they key on filter_dict alone), so a residual query would not populate them. + "WHERE i.interest = 'tennis' AND p.gender = 'female' " + "RETURN count(p) AS numPersons, c.city AS city, c.country AS country " + "ORDER BY numPersons DESC, city LIMIT 1" + ) + graph = cast( + _CypherTestGraph, + _CypherTestGraph().nodes(pl.from_pandas(nodes), "id").edges(pl.from_pandas(edges), "s", "d"), + ) + plan = _compiled_connected_join_plan(query) + + # The polars-native cached path populates a PER-EXECUTION cache_store (never the Plottable, + # see BLOCKER 1). Threading one store across two calls exercises intra-store reuse: the + # cached frames are returned by identity, not recomputed. + cache_store: dict = {} + expected = [{"city": "London", "country": "United Kingdom", "numPersons": 2}] + + direct = _connected_join_two_star_fast_grouped_count(graph, plan, engine=Engine.POLARS, cache_store=cache_store) + assert direct is not None + assert _to_pandas_df(direct).to_dict(orient="records") == expected + + node_filter_cache = cache_store["_gfql_connected_join_node_filter_cache"] + node_ids_cache = cache_store["_gfql_connected_join_node_ids_cache"] + first_arm_cache = cache_store["_gfql_connected_join_first_arm_shared_counts_cache"] + second_arm_cache = cache_store["_gfql_connected_join_second_arm_group_rows_cache"] + assert len(node_filter_cache) == 2 + assert len(node_ids_cache) == 3 + assert len(first_arm_cache) == 1 + assert len(second_arm_cache) == 1 + snapshot = { + name: {key: id(value) for key, value in sub.items()} + for name, sub in cache_store.items() + } + + direct_again = _connected_join_two_star_fast_grouped_count(graph, plan, engine=Engine.POLARS, cache_store=cache_store) + assert direct_again is not None + assert _to_pandas_df(direct_again).to_dict(orient="records") == expected + # Same store -> cached frames reused by identity, sizes unchanged. + assert { + name: {key: id(value) for key, value in sub.items()} + for name, sub in cache_store.items() + } == snapshot + + # BLOCKER 1: caches must NEVER be persisted on the caller's Plottable (that leaked stale + # results across gfql() calls after in-place mutation). + assert not [attr for attr in vars(graph) if attr.startswith("_gfql_connected_join")] + + # A fresh execution (no shared store) recomputes from scratch: correct + still no leak. + fresh = graph.gfql(query, engine="polars") + assert _to_pandas_df(fresh._nodes).to_dict(orient="records") == expected + assert not [attr for attr in vars(graph) if attr.startswith("_gfql_connected_join")] + + +@pytest.mark.parametrize("engine", ["pandas", "polars"]) +def test_t1_connected_comma_two_star_direct_grouped_count_applies_single_alias_residual(engine: str) -> None: + # Connected-plan #1729 lowers toLower(...) equality to a post-join where_rows RESIDUAL + # (not a Fullmatch filter_dict push). The structural fast grouped-count must still ENGAGE + # and apply that single-alias residual to the first-leaf node set before counting, else it + # would over-count. Hand-derived Cypher truth: only i0 ("Fine Dining") matches + # toLower(...)='fine dining'; p0 and p1 each reach i0 and live in c0 -> count(p)=2. Without + # the residual, p0's second interest (i1) would inflate the London group to 3. + if engine == "polars": + pytest.importorskip("polars") + nodes = pd.DataFrame({ + "id": ["p0", "p1", "i0", "i1", "c0"], + "node_type": ["Person", "Person", "Interest", "Interest", "City"], + "interest": [None, None, "Fine Dining", "photography", None], + "city": [None, None, None, None, "London"], + }) + edges = pd.DataFrame({ + "s": ["p0", "p0", "p1", "p0", "p1"], + "d": ["i0", "i1", "i0", "c0", "c0"], + "rel": ["HAS_INTEREST", "HAS_INTEREST", "HAS_INTEREST", "LIVES_IN", "LIVES_IN"], + }) + query = ( + "MATCH (p {node_type:'Person'})-[{rel:'HAS_INTEREST'}]->(i {node_type:'Interest'}), " + "(p)-[{rel:'LIVES_IN'}]->(c {node_type:'City'}) " + "WHERE toLower(i.interest) = toLower('fine dining') " + "RETURN count(p) AS n, c.city AS city" + ) + graph = _mk_graph(nodes, edges) + plan = _compiled_connected_join_plan(query) + assert "where_rows" in _post_join_functions(query) + + # The fast grouped-count engages despite the residual (does not silently decline). + direct = _connected_join_two_star_fast_grouped_count(graph, plan, engine=Engine(engine)) + assert direct is not None + assert _to_pandas_df(direct).to_dict(orient="records") == [{"n": 2, "city": "London"}] + + result = graph.gfql(query, engine=engine) + assert _to_pandas_df(result._nodes).to_dict(orient="records") == [{"n": 2, "city": "London"}] + + +def test_t1_connected_comma_two_star_multi_alias_residual_declines_to_slow_path() -> None: + # A post-join residual spanning MORE THAN ONE materialized alias cannot be applied to a + # single node set, so _connected_join_two_star_split_residuals returns None and the fast + # path declines to the (correct) slow path. Verify the split helper declines AND the query + # still answers correctly. + nodes = pd.DataFrame({ + "id": ["p0", "p1", "i0", "i1", "c0", "c1"], + "node_type": ["Person", "Person", "Interest", "Interest", "City", "City"], + "interest": [None, None, "London", "Paris", None, None], + "city": [None, None, None, None, "London", "Berlin"], + }) + edges = pd.DataFrame({ + "s": ["p0", "p1", "p0", "p1"], + "d": ["i0", "i1", "c0", "c1"], + "rel": ["HAS_INTEREST", "HAS_INTEREST", "LIVES_IN", "LIVES_IN"], + }) + graph = _mk_graph(nodes, edges) + # Cross-alias residual: interest string equals the city string (spans i and c). + query = ( + "MATCH (p {node_type:'Person'})-[{rel:'HAS_INTEREST'}]->(i {node_type:'Interest'}), " + "(p)-[{rel:'LIVES_IN'}]->(c {node_type:'City'}) " + "WHERE i.interest = c.city RETURN count(p) AS n" + ) + plan = _compiled_connected_join_plan(query) + materialized = set() + alias_targets = {} + for pattern in plan.pattern_chains: + for op in pattern.chain: + name = getattr(op, "_name", None) + if isinstance(op, ASTNode) and isinstance(name, str): + materialized.add(name) + alias_targets[name] = op + assert _connected_join_two_star_split_residuals(plan, alias_targets, materialized) is None + + # Hand-derived truth: only p0 has interest 'London' AND lives in city 'London' -> n=1. + assert _to_pandas_df(graph.gfql(query, engine="pandas")._nodes).to_dict(orient="records") == [{"n": 1}] + + +def _mk_two_star_coverage_graph(engine: str) -> Any: + # p0 age30 SF, p1 age45 SF, p2 age28 NY; all like i0 (Fine Dining). p1 excluded by age<=40. + nodes = pd.DataFrame({ + "id": ["p0", "p1", "p2", "i0", "c0", "c1"], + "node_type": ["Person", "Person", "Person", "Interest", "City", "City"], + "age": [30, 45, 28, None, None, None], + "score": [1.5, 2.5, 1.5, None, None, None], + "active": [True, True, False, None, None, None], + "interest": [None, None, None, "Fine Dining", None, None], + "city": [None, None, None, None, "SF", "NY"], + }) + edges = pd.DataFrame({ + "s": ["p0", "p1", "p2", "p0", "p1", "p2"], + "d": ["i0", "i0", "i0", "c0", "c0", "c1"], + "rel": ["HAS_INTEREST", "HAS_INTEREST", "HAS_INTEREST", "LIVES_IN", "LIVES_IN", "LIVES_IN"], + }) + if engine == "polars": + pl = pytest.importorskip("polars") + nodes, edges = pl.from_pandas(nodes), pl.from_pandas(edges) + return graphistry.nodes(nodes, "id").edges(edges, "s", "d") + + +@pytest.mark.parametrize("engine", ["pandas", "polars"]) +def test_t1_connected_comma_two_star_range_filter_grouped_count(engine: str) -> None: + # Range filter lowers to an AllOf(GE,LE) predicate pushed into the node filter_dict, which + # exercises the polars cached fast path's filter-value cache key (composite + scalar-val + # predicate branches). Hand-derived truth: p0 (age30,SF) and p2 (age28,NY) pass 25<=age<=40; + # p1 (age45) excluded -> SF:1, NY:1. + if engine == "polars": + pytest.importorskip("polars") + graph = _mk_two_star_coverage_graph(engine) + query = ( + "MATCH (p {node_type:'Person'})-[{rel:'HAS_INTEREST'}]->(i {node_type:'Interest'}), " + "(p)-[{rel:'LIVES_IN'}]->(c {node_type:'City'}) " + "WHERE p.age >= 25 AND p.age <= 40 " + "RETURN count(p) AS n, c.city AS city ORDER BY city" + ) + got = _to_pandas_df(graph.gfql(query, engine=engine)._nodes).to_dict(orient="records") + assert got == [{"n": 1, "city": "NY"}, {"n": 1, "city": "SF"}] + + +@pytest.mark.parametrize("engine", ["pandas", "polars"]) +def test_t1_connected_comma_two_star_bool_and_float_filters_grouped_count(engine: str) -> None: + # Inline bool + float property filters push scalar bool/float values into the node + # filter_dict, exercising those filter-value cache-key branches on the polars cached path. + # Hand-derived truth: active=true AND score=1.5 -> only p0 (SF). (p2 is score1.5 but + # active=false; p1 is active but score2.5.) + if engine == "polars": + pytest.importorskip("polars") + graph = _mk_two_star_coverage_graph(engine) + query = ( + "MATCH (p {node_type:'Person', active: true, score: 1.5})-[{rel:'HAS_INTEREST'}]->(i {node_type:'Interest'}), " + "(p)-[{rel:'LIVES_IN'}]->(c {node_type:'City'}) " + "RETURN count(p) AS n, c.city AS city ORDER BY city" + ) + got = _to_pandas_df(graph.gfql(query, engine=engine)._nodes).to_dict(orient="records") + assert got == [{"n": 1, "city": "SF"}] + + +@pytest.mark.parametrize("engine", ["pandas", "polars"]) +def test_t1_connected_comma_two_star_singleton_leaf_grouped_count(engine: str) -> None: + # A single Interest node (i0) makes the first-leaf id set a singleton, exercising the + # singleton dst->source count optimization / cached first-arm path. Hand-derived truth: + # all three persons like the one interest; grouped by city -> SF:2 (p0,p1), NY:1 (p2). + if engine == "polars": + pytest.importorskip("polars") + graph = _mk_two_star_coverage_graph(engine) + query = ( + "MATCH (p {node_type:'Person'})-[{rel:'HAS_INTEREST'}]->(i {node_type:'Interest'}), " + "(p)-[{rel:'LIVES_IN'}]->(c {node_type:'City'}) " + "RETURN count(p) AS n, c.city AS city ORDER BY n DESC, city" + ) + got = _to_pandas_df(graph.gfql(query, engine=engine)._nodes).to_dict(orient="records") + assert got == [{"n": 2, "city": "SF"}, {"n": 1, "city": "NY"}] + + +@pytest.mark.parametrize("engine", ["pandas", "polars"]) +def test_t1_connected_comma_two_star_count_star_uses_fast_rows(engine: str) -> None: + # count(*) references no alias identity, so fast_rows materializes the joined rows and + # post_join counts them. Hand-derived truth: one (p,i,c) row per person (each likes i0 and + # lives in one city) -> 3 rows -> count(*) = 3. + if engine == "polars": + pytest.importorskip("polars") + graph = _mk_two_star_coverage_graph(engine) + direct = _connected_join_two_star_fast_rows( + graph, _compiled_connected_join_plan( + "MATCH (p {node_type:'Person'})-[{rel:'HAS_INTEREST'}]->(i {node_type:'Interest'}), " + "(p)-[{rel:'LIVES_IN'}]->(c {node_type:'City'}) RETURN count(*) AS n" + ), engine=Engine(engine), + ) + assert direct is not None # fast_rows engages for count(*) + query = ( + "MATCH (p {node_type:'Person'})-[{rel:'HAS_INTEREST'}]->(i {node_type:'Interest'}), " + "(p)-[{rel:'LIVES_IN'}]->(c {node_type:'City'}) RETURN count(*) AS n" + ) + assert _to_pandas_df(graph.gfql(query, engine=engine)._nodes).to_dict(orient="records") == [{"n": 3}] + + +@pytest.mark.parametrize("engine", ["pandas", "polars"]) +def test_t1_connected_comma_two_star_unsupported_aggregate_shapes_fall_back(engine: str) -> None: + # Shapes the fast grouped-count does not serve (two aggregates; a non-count aggregate) must + # DECLINE to the slow path and still answer correctly. Verifies the decline guards + fallback. + if engine == "polars": + pytest.importorskip("polars") + graph = _mk_two_star_coverage_graph(engine) + base = ( + "MATCH (p {node_type:'Person'})-[{rel:'HAS_INTEREST'}]->(i {node_type:'Interest'}), " + "(p)-[{rel:'LIVES_IN'}]->(c {node_type:'City'}) " + ) + two_agg = base + "RETURN count(p) AS n, count(DISTINCT c) AS m" + plan = _compiled_connected_join_plan(two_agg) + assert _connected_join_two_star_fast_grouped_count(graph, plan, engine=Engine(engine)) is None + if engine == "pandas": + # count(p)=3 persons; count(DISTINCT c)=2 (SF, NY). + assert _to_pandas_df(graph.gfql(two_agg, engine=engine)._nodes).to_dict(orient="records") == [{"n": 3, "m": 2}] + # Non-count aggregate (sum) grouped by city -> SF 30+45=75, NY 28. + got = _to_pandas_df(graph.gfql(base + "RETURN sum(p.age) AS s, c.city AS city ORDER BY city", engine=engine)._nodes).to_dict(orient="records") + assert got == [{"s": 28.0, "city": "NY"}, {"s": 75.0, "city": "SF"}] + + +@pytest.mark.parametrize("engine", ["pandas", "polars"]) +def test_t1_connected_comma_two_star_grouped_count_order_asc_limit(engine: str) -> None: + # Exercises the fast grouped-count ORDER BY (asc) + LIMIT branch. SF:2, NY:1; ORDER BY city + # ASC LIMIT 1 -> NY. + if engine == "polars": + pytest.importorskip("polars") + graph = _mk_two_star_coverage_graph(engine) + query = ( + "MATCH (p {node_type:'Person'})-[{rel:'HAS_INTEREST'}]->(i {node_type:'Interest'}), " + "(p)-[{rel:'LIVES_IN'}]->(c {node_type:'City'}) " + "RETURN count(p) AS n, c.city AS city ORDER BY city ASC LIMIT 1" + ) + assert _to_pandas_df(graph.gfql(query, engine=engine)._nodes).to_dict(orient="records") == [{"n": 1, "city": "NY"}] + + +def test_t1_connected_comma_two_star_count_second_leaf_alias_no_crash() -> None: + # Regression: count(c)/count(DISTINCT c) lower `c` to c.__gfql_node_id__ (identity), which + # fast_rows cannot materialize as a node property -- it must DECLINE to the slow path rather + # than emit a frame that makes post_join's count(c.__gfql_node_id__) dereference a missing + # column (previously a GFQLTypeError crash on pandas). Slow-path truth: 3 (p,i,c) rows + # (p0->c0, p1->c0, p2->c1) -> count(c)=3, count(DISTINCT c)=2 (c0, c1). + graph = _mk_two_star_coverage_graph("pandas") + base = ( + "MATCH (p {node_type:'Person'})-[{rel:'HAS_INTEREST'}]->(i {node_type:'Interest'}), " + "(p)-[{rel:'LIVES_IN'}]->(c {node_type:'City'}) " + ) + # fast_rows declines (identity of the second leaf is not a materializable property). + assert _connected_join_two_star_fast_rows( + graph, _compiled_connected_join_plan(base + "RETURN count(c) AS n"), engine=Engine.PANDAS + ) is None + assert _to_pandas_df(graph.gfql(base + "RETURN count(c) AS n", engine="pandas")._nodes).to_dict(orient="records") == [{"n": 3}] + assert _to_pandas_df(graph.gfql(base + "RETURN count(DISTINCT c) AS n", engine="pandas")._nodes).to_dict(orient="records") == [{"n": 2}] + + +def test_t1_connected_comma_two_star_no_stale_cache_after_inplace_edge_mutation() -> None: + # BLOCKER 1: the fast-path caches were setattr'd onto the caller's Plottable and keyed by + # id(), so a second gfql() after an in-place edge mutation returned a STALE wrong answer. + # The per-execution cache_store fix must recompute on the second call (and never leak a + # cache attribute onto the Plottable). + nodes = pd.DataFrame({ + "id": ["p0", "p1", "i0", "c0", "c1"], + "node_type": ["Person", "Person", "Interest", "City", "City"], + "city": [None, None, None, "SF", "NY"], + }) + edges = pd.DataFrame({ + "s": ["p0", "p1", "p0", "p1"], + "d": ["i0", "i0", "c0", "c1"], + "rel": ["HAS_INTEREST", "HAS_INTEREST", "LIVES_IN", "LIVES_IN"], + }) + g = graphistry.nodes(nodes, "id").edges(edges, "s", "d") + query = ( + "MATCH (p {node_type:'Person'})-[{rel:'HAS_INTEREST'}]->(i {node_type:'Interest'}), " + "(p)-[{rel:'LIVES_IN'}]->(c {node_type:'City'}) " + "RETURN count(p) AS n, c.city AS city" + ) + + def rows(res: Any) -> Any: + return sorted(_to_pandas_df(res._nodes).to_dict(orient="records"), key=lambda r: str(r["city"])) + + assert rows(g.gfql(query, engine="pandas")) == [{"n": 1, "city": "NY"}, {"n": 1, "city": "SF"}] + assert not [attr for attr in vars(g) if attr.startswith("_gfql_connected_join")] + + # Drop the p1->c1 (NY) LIVES_IN edge in place on the SAME edge object g holds by reference. + ny_idx = edges.index[(edges["s"] == "p1") & (edges["d"] == "c1")] + edges.drop(index=ny_idx, inplace=True) + assert g._edges is edges # the fix must not depend on this being false + + # Truth after the mutation: only SF remains. + assert rows(g.gfql(query, engine="pandas")) == [{"n": 1, "city": "SF"}] + assert not [attr for attr in vars(g) if attr.startswith("_gfql_connected_join")] + + +def test_t1_connected_comma_two_star_polars_grouped_count_orders_nulls_as_largest() -> None: + pl = pytest.importorskip("polars") + # BLOCKER 3: the polars fast grouped-count sort omitted nulls_last, and polars defaults to + # nulls-first. openCypher orders NULL as the largest value, so ASC ... LIMIT 1 must return + # the smallest NON-null group key (NY), not the NULL-city group. + nodes = pl.DataFrame({ + "id": ["p0", "p1", "p2", "i0", "c0", "c1", "c2"], + "node_type": ["Person", "Person", "Person", "Interest", "City", "City", "City"], + "city": [None, None, None, None, "NY", "SF", None], + }) + edges = pl.DataFrame({ + "s": ["p0", "p1", "p2", "p0", "p1", "p2"], + "d": ["i0", "i0", "i0", "c0", "c1", "c2"], + "rel": ["HAS_INTEREST", "HAS_INTEREST", "HAS_INTEREST", "LIVES_IN", "LIVES_IN", "LIVES_IN"], + }) + g = graphistry.nodes(nodes, "id").edges(edges, "s", "d") + query = ( + "MATCH (p {node_type:'Person'})-[{rel:'HAS_INTEREST'}]->(i {node_type:'Interest'}), " + "(p)-[{rel:'LIVES_IN'}]->(c {node_type:'City'}) " + "RETURN count(p) AS n, c.city AS city ORDER BY city ASC LIMIT 1" + ) + assert _to_pandas_df(g.gfql(query, engine="polars")._nodes).to_dict(orient="records") == [{"n": 1, "city": "NY"}] + + +def test_t1_connected_comma_two_star_polars_grouped_count_dedups_duplicate_node_rows() -> None: + pl = pytest.importorskip("polars") + # IMPORTANT: the polars second-leaf lookup lacked .unique(subset=[node_col]) (the pandas + # branch drop_duplicates it), so a duplicate node row multiplied the join and over-counted. + nodes = pl.DataFrame({ + "id": ["p0", "i0", "c0", "c0"], # c0 duplicated + "node_type": ["Person", "Interest", "City", "City"], + "city": [None, None, "SF", "SF"], + }) + edges = pl.DataFrame({ + "s": ["p0", "p0"], + "d": ["i0", "c0"], + "rel": ["HAS_INTEREST", "LIVES_IN"], + }) + g = graphistry.nodes(nodes, "id").edges(edges, "s", "d") + query = ( + "MATCH (p {node_type:'Person'})-[{rel:'HAS_INTEREST'}]->(i {node_type:'Interest'}), " + "(p)-[{rel:'LIVES_IN'}]->(c {node_type:'City'}) " + "RETURN count(p) AS n, c.city AS city" + ) + # Truth: one (p0,i0,c0) match -> SF n=1 (the duplicate c0 node row must not inflate it). + assert _to_pandas_df(g.gfql(query, engine="polars")._nodes).to_dict(orient="records") == [{"n": 1, "city": "SF"}] + + +def test_t1_connected_comma_two_star_fast_path_rejects_first_leaf_props() -> None: + query = ( + "MATCH (p {node_type:'Person'})-[{rel:'HAS_INTEREST'}]->(i {node_type:'Interest'}), " + "(p)-[{rel:'LIVES_IN'}]->(c {node_type:'City'}) " + "WHERE toLower(i.interest) = toLower('fine dining') " + "RETURN i.interest AS interest, count(p) AS numPersons" + ) + + plan = _compiled_connected_join_plan(query) + assert _connected_join_two_star_fast_rows(_mk_graph_benchmark_t1_shape_graph(), plan, engine=Engine.PANDAS) is None + + +def test_t1_connected_comma_edge_filter_cache_reuses_simple_edge_match() -> None: + graph = _mk_graph_benchmark_t1_shape_graph() + edges = graph._edges + assert edges is not None + + # Reuse is scoped to a per-execution cache_store (never the Plottable, see BLOCKER 1): + # two calls sharing the store return the identical cached frame. + cache_store: dict = {} + first = _connected_join_cached_edge_filter(graph, edges, {"rel": "HAS_INTEREST"}, engine=Engine.PANDAS, cache_store=cache_store) + second = _connected_join_cached_edge_filter(graph, edges, {"rel": "HAS_INTEREST"}, engine=Engine.PANDAS, cache_store=cache_store) + + assert first is second + assert set(first["rel"].unique()) == {"HAS_INTEREST"} + # No cache attribute is persisted on the Plottable. + assert not [attr for attr in vars(graph) if attr.startswith("_gfql_connected_join")] + + # Without a shared store, calls do not leak or cross-contaminate (fresh compute each time). + isolated = _connected_join_cached_edge_filter(graph, edges, {"rel": "HAS_INTEREST"}, engine=Engine.PANDAS) + assert set(isolated["rel"].unique()) == {"HAS_INTEREST"} + assert not [attr for attr in vars(graph) if attr.startswith("_gfql_connected_join")] + def test_issue_1413_ic3_entity_membership_positive_same_city_friend_only() -> None: graph = _mk_ic3_cross_country_shape_graph() @@ -18704,6 +19323,31 @@ def test_count_table_frame_op_error_and_empty_paths() -> None: out3 = frame_ops.count_table(ctx3, table="nodes", alias="c") assert out3._nodes.to_dict(orient="records") == [{"c": 0}] + +def test_connected_join_post_property_columns_walks_nested_params() -> None: + plan = ConnectedMatchJoinPlan( + pattern_chains=(), + pattern_shared_node_aliases=(), + post_join_chain=Chain( + [ + ASTCall( + "select", + { + "items": [ + ("city", "a.city"), + ("nested", {"left": ["a.country", "b.age", "a.city"]}), + ] + }, + ), + ], + validate=False, + ), + ) + + assert _connected_join_post_property_columns(plan, "a") == ["city", "country"] + assert _connected_join_post_property_columns(plan, "b") == ["age"] + + def test_string_cypher_parenthesized_and_preserves_missing_property_as_null() -> None: # A parenthesized AND must stay on the row-filter path, not the filter_dict path: # the row engine treats an absent property as null (Cypher semantics), whereas