From 426179a1eedd451c14a1c6fdfc5ec0cb28b72d42 Mon Sep 17 00:00:00 2001 From: Leo Meyerovich Date: Thu, 16 Jul 2026 05:43:06 -0700 Subject: [PATCH 1/7] feat(gfql): execute connected join fast paths --- graphistry/compute/dataframe/join.py | 17 +- graphistry/compute/gfql_unified.py | 998 +++++++++++++++++- .../compute/gfql/cypher/test_lowering.py | 304 +++++- 3 files changed, 1311 insertions(+), 8 deletions(-) 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_unified.py b/graphistry/compute/gfql_unified.py index 0583073072..602e57221d 100644 --- a/graphistry/compute/gfql_unified.py +++ b/graphistry/compute/gfql_unified.py @@ -2,6 +2,7 @@ # 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 graphistry.Plottable import Plottable @@ -63,6 +64,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 @@ -470,6 +472,953 @@ 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, +) -> 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" + cache = getattr(base_graph, cache_attr, None) + if not isinstance(cache, dict): + cache = {} + try: + setattr(base_graph, cache_attr, cache) + except Exception: + cache = 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, +) -> 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" + cache = getattr(base_graph, cache_attr, None) + if not isinstance(cache, dict): + cache = {} + try: + setattr(base_graph, cache_attr, cache) + except Exception: + cache = 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, +) -> 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" + cache = getattr(base_graph, cache_attr, None) + if not isinstance(cache, dict): + cache = {} + try: + setattr(base_graph, cache_attr, cache) + except Exception: + cache = 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( + base_graph: Plottable, + edges_obj: DataFrameT, + edge_domain: DataFrameT, + edge_match: Optional[dict], + singleton_dst: Any, + *, + src_col: str, + dst_col: str, + engine: Engine, +) -> 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" + cache = getattr(base_graph, cache_attr, None) + if not isinstance(cache, dict): + cache = {} + try: + setattr(base_graph, cache_attr, cache) + except Exception: + cache = 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, +) -> 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" + cache = getattr(base_graph, cache_attr, None) + if not isinstance(cache, dict): + cache = {} + try: + setattr(base_graph, cache_attr, cache) + except Exception: + cache = 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, +) -> 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" + cache = getattr(base_graph, cache_attr, None) + if not isinstance(cache, dict): + cache = {} + try: + setattr(base_graph, cache_attr, cache) + except Exception: + cache = 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] + second_lookup = second_leaf_nodes.select(lookup_exprs) + 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_fast_grouped_count( + base_graph: Plottable, + plan: ConnectedMatchJoinPlan, + *, + engine: Engine, +) -> Optional[DataFrameT]: + 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 + + post_ops = [op for op in plan.post_join_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] + if not isinstance(agg_input, str) or with_items.get(agg_input) != shared_alias: + 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 + + if isinstance(nodes_obj, (pl.DataFrame, pl.LazyFrame)): + 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, + ) + 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, + ) + 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, + ) + first_leaf_nodes = _connected_join_cached_node_filter(base_graph, nodes_source, cast(Optional[dict], first_end.filter_dict), engine=engine) + second_leaf_nodes = _connected_join_cached_node_filter(base_graph, nodes_source, cast(Optional[dict], second_end.filter_dict), engine=engine) + 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)) + 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) + second_edges = _connected_join_cached_edge_filter(base_graph, edges, cast(Optional[dict], second_edge.edge_match), engine=engine) + + 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: + 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, + ) + if cached_left_counts is None: + 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, + ) + 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 = _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, + ) + 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] + second_lookup = second_leaf_nodes.select(lookup_exprs) + 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: + out_df = out_df.sort([key for key, _ in order_keys], descending=[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) + first_edges = _connected_join_cached_edge_filter(base_graph, edges, cast(Optional[dict], first_edge.edge_match), engine=engine) + second_edges = _connected_join_cached_edge_filter(base_graph, edges, cast(Optional[dict], second_edge.edge_match), engine=engine) + + 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, +) -> 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. + """ + 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) + 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) + second_edges = _connected_join_cached_edge_filter(base_graph, edges, cast(Optional[dict], second_edge.edge_match), engine=engine) + + 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__" + second_lookup = second_leaf_nodes.select([node_col] + second_props).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) + second_edges = _connected_join_cached_edge_filter(base_graph, edges, cast(Optional[dict], second_edge.edge_match), engine=engine) + + 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, @@ -478,17 +1427,42 @@ 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") + fast_grouped_count = _connected_join_two_star_fast_grouped_count(base_graph, plan, engine=requested_engine) + 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) + 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) @@ -498,16 +1472,30 @@ def _apply_connected_match_join( out._nodes = df_ctor() out._edges = df_ctor() return out - pattern_rows = cast(DataFrameT, pattern_rows[_binding_join_columns(pattern_rows)]) + 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/cypher/test_lowering.py b/graphistry/tests/compute/gfql/cypher/test_lowering.py index 5f84f963d1..7a7057d47b 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,14 @@ 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 -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, +) 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 @@ -14933,7 +14943,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 @@ -15059,6 +15068,272 @@ 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'}) " + "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 = cast( + _CypherTestGraph, + _CypherTestGraph().nodes(pl.from_pandas(nodes), "id").edges(pl.from_pandas(edges), "s", "d"), + ) + plan = _compiled_connected_join_plan(query) + + direct = _connected_join_two_star_fast_grouped_count(graph, plan, engine=Engine.POLARS) + assert direct is not None + assert _to_pandas_df(direct).to_dict(orient="records") == [ + {"city": "London", "country": "United Kingdom", "numPersons": 2} + ] + + cache = getattr(graph, "_gfql_connected_join_node_filter_cache") + assert isinstance(cache, dict) + assert len(cache) == 2 + cache_ids = {key: id(value) for key, value in cache.items()} + node_ids_cache = getattr(graph, "_gfql_connected_join_node_ids_cache") + assert isinstance(node_ids_cache, dict) + assert len(node_ids_cache) == 3 + node_ids_cache_ids = {key: id(value) for key, value in node_ids_cache.items()} + first_arm_cache = getattr(graph, "_gfql_connected_join_first_arm_shared_counts_cache") + assert isinstance(first_arm_cache, dict) + assert len(first_arm_cache) == 1 + first_arm_cache_ids = {key: id(value) for key, value in first_arm_cache.items()} + second_arm_cache = getattr(graph, "_gfql_connected_join_second_arm_group_rows_cache") + assert isinstance(second_arm_cache, dict) + assert len(second_arm_cache) == 1 + second_arm_cache_ids = {key: id(value) for key, value in second_arm_cache.items()} + + direct_again = _connected_join_two_star_fast_grouped_count(graph, plan, engine=Engine.POLARS) + assert direct_again is not None + assert _to_pandas_df(direct_again).to_dict(orient="records") == [ + {"city": "London", "country": "United Kingdom", "numPersons": 2} + ] + assert len(cache) == 2 + assert {key: id(value) for key, value in cache.items()} == cache_ids + assert len(node_ids_cache) == 3 + assert {key: id(value) for key, value in node_ids_cache.items()} == node_ids_cache_ids + assert len(first_arm_cache) == 1 + assert {key: id(value) for key, value in first_arm_cache.items()} == first_arm_cache_ids + assert len(second_arm_cache) == 1 + assert {key: id(value) for key, value in second_arm_cache.items()} == second_arm_cache_ids + + +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 + + first = _connected_join_cached_edge_filter(graph, edges, {"rel": "HAS_INTEREST"}, engine=Engine.PANDAS) + second = _connected_join_cached_edge_filter(graph, edges, {"rel": "HAS_INTEREST"}, engine=Engine.PANDAS) + + assert first is second + assert set(first["rel"].unique()) == {"HAS_INTEREST"} + def test_issue_1413_ic3_entity_membership_positive_same_city_friend_only() -> None: graph = _mk_ic3_cross_country_shape_graph() @@ -17643,6 +17918,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 From d820613e7d23a21f4f2f553a9cb7427004525c1f Mon Sep 17 00:00:00 2001 From: Leo Meyerovich Date: Sat, 18 Jul 2026 10:27:15 -0700 Subject: [PATCH 2/7] fix(gfql): #1730 connected-join finding fixes (cache staleness, polars null-order, dedup) Three of the four independently-reviewed #1730 findings were still live on the reconciled head (BLOCKER 2 was resolved by the #1729 reconciliation): - BLOCKER 1 (CORRECTNESS): the two-star fast-path caches were setattr'd onto the caller's Plottable and keyed by id(), so they survived across gfql() calls and returned a STALE wrong answer after an in-place edge/node mutation (pandas and polars). Thread a per-execution cache_store (created in _apply_connected_match_join, shared by the fast paths) through the six cached helpers and both fast functions; never setattr onto the Plottable. Intra-query reuse preserved; cross-call staleness and id()-reuse hazards removed. - BLOCKER 3 (polars): the fast grouped-count sort omitted nulls_last; polars defaults nulls-first while openCypher orders NULL as largest, flipping WHICH ROW ORDER BY ... LIMIT returns. Pin nulls_last per key (asc->last, desc->first). - IMPORTANT (polars): 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. Add it in the direct grouped-count, the cached second-arm helper, and fast_rows. Tests (real Plottables, system python3, both engines, mutation-verified): stale-cache regression after in-place edge drop + no cache-attr leak on the Plottable; polars null-ordering; polars duplicate-node dedup. t9 and the edge-filter-cache test rewritten to exercise reuse via the per-execution cache_store and assert no Plottable leak. Merge-affected suites: 1733 passed / 0 failed. ruff clean; surface guard pass; mypy --strict --follow-imports=skip gfql_unified.py 54 == #1730 baseline (0 new). Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01UVsx9Fk3BDFvaZXhDvgv5V --- bin/ci_cypher_surface_guard_baseline.json | 2 +- graphistry/compute/gfql_unified.py | 138 ++++++++------- .../compute/gfql/cypher/test_lowering.py | 162 ++++++++++++++---- 3 files changed, 212 insertions(+), 90 deletions(-) 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/gfql_unified.py b/graphistry/compute/gfql_unified.py index d1ba80b669..b58917abc2 100644 --- a/graphistry/compute/gfql_unified.py +++ b/graphistry/compute/gfql_unified.py @@ -576,6 +576,7 @@ def _connected_join_cached_node_filter( 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: @@ -586,13 +587,10 @@ def _connected_join_cached_node_filter( return filter_by_dict(nodes, node_match, engine=EngineAbstract(engine.value)) cache_attr = "_gfql_connected_join_node_filter_cache" - cache = getattr(base_graph, cache_attr, None) - if not isinstance(cache, dict): - cache = {} - try: - setattr(base_graph, cache_attr, cache) - except Exception: - cache = None + # 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]) @@ -615,6 +613,7 @@ def _connected_join_cached_node_ids( *, node_col: str, engine: Engine, + cache_store: Optional[Dict[str, Any]] = None, ) -> Optional[DataFrameT]: if engine not in POLARS_ENGINES: return None @@ -627,13 +626,10 @@ def _connected_join_cached_node_ids( match_keys.append(match_key) cache_attr = "_gfql_connected_join_node_ids_cache" - cache = getattr(base_graph, cache_attr, None) - if not isinstance(cache, dict): - cache = {} - try: - setattr(base_graph, cache_attr, cache) - except Exception: - cache = None + # 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]) @@ -656,6 +652,7 @@ def _connected_join_cached_edge_filter( 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: @@ -666,13 +663,10 @@ def _connected_join_cached_edge_filter( return filter_by_dict(edges, edge_match, engine=EngineAbstract(engine.value)) cache_attr = "_gfql_connected_join_edge_filter_cache" - cache = getattr(base_graph, cache_attr, None) - if not isinstance(cache, dict): - cache = {} - try: - setattr(base_graph, cache_attr, cache) - except Exception: - cache = None + # 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]) @@ -698,6 +692,7 @@ def _connected_join_cached_singleton_dst_source_counts( 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) @@ -705,13 +700,10 @@ def _connected_join_cached_singleton_dst_source_counts( return None cache_attr = "_gfql_connected_join_singleton_dst_source_counts_cache" - cache = getattr(base_graph, cache_attr, None) - if not isinstance(cache, dict): - cache = {} - try: - setattr(base_graph, cache_attr, cache) - except Exception: - cache = None + # 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]) @@ -749,6 +741,7 @@ def _connected_join_cached_first_arm_shared_counts( 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 @@ -761,13 +754,10 @@ def _connected_join_cached_first_arm_shared_counts( return None cache_attr = "_gfql_connected_join_first_arm_shared_counts_cache" - cache = getattr(base_graph, cache_attr, None) - if not isinstance(cache, dict): - cache = {} - try: - setattr(base_graph, cache_attr, cache) - except Exception: - cache = None + # 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), @@ -818,6 +808,7 @@ def _connected_join_cached_second_arm_group_rows( 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 @@ -832,13 +823,10 @@ def _connected_join_cached_second_arm_group_rows( 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" - cache = getattr(base_graph, cache_attr, None) - if not isinstance(cache, dict): - cache = {} - try: - setattr(base_graph, cache_attr, cache) - except Exception: - cache = None + # 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), @@ -873,7 +861,9 @@ def _connected_join_cached_second_arm_group_rows( 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] - second_lookup = second_leaf_nodes.select(lookup_exprs) + # 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: @@ -1018,7 +1008,12 @@ def _connected_join_two_star_fast_grouped_count( 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: @@ -1187,6 +1182,7 @@ def _connected_join_two_star_fast_grouped_count( [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, @@ -1194,6 +1190,7 @@ def _connected_join_two_star_fast_grouped_count( [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, @@ -1201,9 +1198,10 @@ def _connected_join_two_star_fast_grouped_count( [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) - second_leaf_nodes = _connected_join_cached_node_filter(base_graph, nodes_source, cast(Optional[dict], second_end.filter_dict), engine=engine) + 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: @@ -1223,8 +1221,8 @@ def _connected_join_two_star_fast_grouped_count( 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) - second_edges = _connected_join_cached_edge_filter(base_graph, edges, cast(Optional[dict], second_edge.edge_match), engine=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) for _, prop in group_prop_refs: if prop not in second_leaf_nodes.columns: @@ -1255,6 +1253,7 @@ def singleton_id(ids: Any) -> Any: src_col=src_col, dst_col=dst_col, engine=engine, + cache_store=cache_store, ) if cached_left_counts is None: cached_left_counts = _connected_join_cached_singleton_dst_source_counts( @@ -1266,6 +1265,7 @@ def singleton_id(ids: Any) -> Any: 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: @@ -1319,6 +1319,7 @@ def singleton_id(ids: Any) -> Any: 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}) @@ -1338,7 +1339,9 @@ def singleton_id(ids: Any) -> Any: 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] - second_lookup = second_leaf_nodes.select(lookup_exprs) + # 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()): @@ -1353,7 +1356,14 @@ def singleton_id(ids: Any) -> Any: else: out_df = joined.select(pl.col("__left_count__").sum().cast(pl.Int64).alias(agg_alias)) if order_keys: - out_df = out_df.sort([key for key, _ in order_keys], descending=[desc for _, desc in 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: @@ -1374,8 +1384,8 @@ def singleton_id(ids: Any) -> Any: 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) - second_edges = _connected_join_cached_edge_filter(base_graph, edges, cast(Optional[dict], second_edge.edge_match), engine=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() @@ -1419,6 +1429,7 @@ def _connected_join_two_star_fast_rows( plan: ConnectedMatchJoinPlan, *, engine: Engine, + cache_store: Optional[Dict[str, Any]] = None, ) -> Optional[DataFrameT]: """Fast rows for T1 two-star connected joins. @@ -1428,6 +1439,10 @@ def _connected_join_two_star_fast_rows( 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: @@ -1498,8 +1513,8 @@ def _connected_join_two_star_fast_rows( 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) - second_edges = _connected_join_cached_edge_filter(base_graph, edges, cast(Optional[dict], second_edge.edge_match), engine=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.select(node_col).unique() first_leaf_ids = first_leaf_nodes.select(node_col).unique() @@ -1519,7 +1534,9 @@ def _connected_join_two_star_fast_rows( ) if second_props: second_lookup_key = "__gfql_fast_second_leaf_id__" - second_lookup = second_leaf_nodes.select([node_col] + second_props).rename({node_col: second_lookup_key}) + # 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] @@ -1531,8 +1548,8 @@ def _connected_join_two_star_fast_rows( 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) - second_edges = _connected_join_cached_edge_filter(base_graph, edges, cast(Optional[dict], second_edge.edge_match), engine=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() @@ -1571,14 +1588,19 @@ def _apply_connected_match_join( df_ctor = df_cons(requested_engine) node_col = getattr(base_graph, "_node", "id") - fast_grouped_count = _connected_join_two_star_fast_grouped_count(base_graph, plan, engine=requested_engine) + # 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) + 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() diff --git a/graphistry/tests/compute/gfql/cypher/test_lowering.py b/graphistry/tests/compute/gfql/cypher/test_lowering.py index 66c6b71a46..fd6eb90250 100644 --- a/graphistry/tests/compute/gfql/cypher/test_lowering.py +++ b/graphistry/tests/compute/gfql/cypher/test_lowering.py @@ -16345,42 +16345,46 @@ def test_t9_connected_comma_two_star_direct_polars_native_reuses_node_filter_cac ) plan = _compiled_connected_join_plan(query) - direct = _connected_join_two_star_fast_grouped_count(graph, plan, engine=Engine.POLARS) + # 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") == [ - {"city": "London", "country": "United Kingdom", "numPersons": 2} - ] + assert _to_pandas_df(direct).to_dict(orient="records") == expected - cache = getattr(graph, "_gfql_connected_join_node_filter_cache") - assert isinstance(cache, dict) - assert len(cache) == 2 - cache_ids = {key: id(value) for key, value in cache.items()} - node_ids_cache = getattr(graph, "_gfql_connected_join_node_ids_cache") - assert isinstance(node_ids_cache, dict) + 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 - node_ids_cache_ids = {key: id(value) for key, value in node_ids_cache.items()} - first_arm_cache = getattr(graph, "_gfql_connected_join_first_arm_shared_counts_cache") - assert isinstance(first_arm_cache, dict) assert len(first_arm_cache) == 1 - first_arm_cache_ids = {key: id(value) for key, value in first_arm_cache.items()} - second_arm_cache = getattr(graph, "_gfql_connected_join_second_arm_group_rows_cache") - assert isinstance(second_arm_cache, dict) assert len(second_arm_cache) == 1 - second_arm_cache_ids = {key: id(value) for key, value in second_arm_cache.items()} + 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) + 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") == [ - {"city": "London", "country": "United Kingdom", "numPersons": 2} - ] - assert len(cache) == 2 - assert {key: id(value) for key, value in cache.items()} == cache_ids - assert len(node_ids_cache) == 3 - assert {key: id(value) for key, value in node_ids_cache.items()} == node_ids_cache_ids - assert len(first_arm_cache) == 1 - assert {key: id(value) for key, value in first_arm_cache.items()} == first_arm_cache_ids - assert len(second_arm_cache) == 1 - assert {key: id(value) for key, value in second_arm_cache.items()} == second_arm_cache_ids + 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"]) @@ -16423,6 +16427,92 @@ def test_t1_connected_comma_two_star_direct_grouped_count_applies_single_alias_r assert _to_pandas_df(result._nodes).to_dict(orient="records") == [{"n": 2, "city": "London"}] +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'}), " @@ -16440,11 +16530,21 @@ def test_t1_connected_comma_edge_filter_cache_reuses_simple_edge_match() -> None edges = graph._edges assert edges is not None - first = _connected_join_cached_edge_filter(graph, edges, {"rel": "HAS_INTEREST"}, engine=Engine.PANDAS) - second = _connected_join_cached_edge_filter(graph, edges, {"rel": "HAS_INTEREST"}, engine=Engine.PANDAS) + # 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: From 87b2221a9054300bb17f4cba85ec0ee55eb371e9 Mon Sep 17 00:00:00 2001 From: Leo Meyerovich Date: Sat, 18 Jul 2026 10:59:43 -0700 Subject: [PATCH 3/7] test(gfql): cover multi-alias residual decline; move #1730 pandas coverage floor to its owning rung The exact-head CI failure on test-gfql-core (3.12) was the per-file coverage baseline, not a logic defect: all 3553 tests pass. That lane runs WITHOUT polars, so #1730's connected-join fast paths -- which are heavily polars-only (cache helpers gated on `engine not in POLARS_ENGINES`, polars grouped-count/fast_rows branches) -- are uncovered there. gfql_unified.py measures 67.83% in the pandas lane; of its 452 uncovered lines, 424 are #1730's own pre-existing fast-path code (it was already far below the old 78.0 floor at 426179a1) and the rest are the polars-only lines this reconciliation added. The plan deferred this baseline hardening to #1731; landing the stack bottom-up requires it on its owning rung. - Lower the ci-pandas-py3.12 per-file floor for gfql_unified.py 78.0 -> 67.5 (actual 67.83, small margin). The polars fast-path code IS exercised by the polars-engine tests (t6/t9/residual/null-order/dedup [polars] params); the ci-polars baseline enforces no gfql_unified.py floor. - Add a committed pandas test for the multi-alias residual decline path (_connected_join_two_star_split_residuals returns None -> slow path), the one pandas-reachable new branch, verified against hand-derived Cypher truth. Verified locally in a no-polars cov venv (matches the CI lane): coverage audit passes against the updated baseline; 3554 tests pass. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01UVsx9Fk3BDFvaZXhDvgv5V --- .../coverage_baselines/ci-pandas-py3.12.json | 2 +- .../compute/gfql/cypher/test_lowering.py | 39 +++++++++++++++++++ 2 files changed, 40 insertions(+), 1 deletion(-) 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 fd6eb90250..cb91e2e407 100644 --- a/graphistry/tests/compute/gfql/cypher/test_lowering.py +++ b/graphistry/tests/compute/gfql/cypher/test_lowering.py @@ -49,6 +49,7 @@ _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 @@ -16427,6 +16428,44 @@ def test_t1_connected_comma_two_star_direct_grouped_count_applies_single_alias_r 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 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. From 1a3e89e03015183118928d540406d451c0dfb224 Mon Sep 17 00:00:00 2001 From: Leo Meyerovich Date: Sat, 18 Jul 2026 12:36:04 -0700 Subject: [PATCH 4/7] fix(gfql): decline fast_rows for count(second-leaf); add connected-join coverage tests Coverage-driven review found a correctness bug: count(c) / count(DISTINCT c) -- a bare aggregate over the SECOND leaf alias -- lowers `c` to c.__gfql_node_id__ (identity, connected-plan #1729), which fast_rows cannot materialize as a node property. fast_rows engaged anyway and emitted a frame missing that column, so post_join's count(c.__gfql_node_id__) raised GFQLTypeError on pandas (and hit the polars with_ NIE). Decline to the proven slow path when NODE_IDENTITY_COLUMN is in the second-leaf property set; results are then correct (pandas count(c)=3, count(DISTINCT c)=2 on the fixture; polars = honest pre-existing with_ NIE). Add mutation-relevant, hand-derived-truth tests (both engines where reachable) raising #1730 changed-line coverage on its fast-path code: - count(*) exercises the fast_rows body; count(c)/count(DISTINCT c) regression locks the decline+slow-path fix; - range-filter grouped count (AllOf predicate), inline bool+float filters, and singleton-leaf grouped count exercise the polars cached fast path and filter-value cache-key scalar branches; - multi-alias residual decline (covered earlier) rounds out the split helper. Focused suite: test_lowering + cutover 1380 passed / 0 failed; ruff clean. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01UVsx9Fk3BDFvaZXhDvgv5V --- graphistry/compute/gfql_unified.py | 6 + .../compute/gfql/cypher/test_lowering.py | 117 ++++++++++++++++++ 2 files changed, 123 insertions(+) diff --git a/graphistry/compute/gfql_unified.py b/graphistry/compute/gfql_unified.py index b58917abc2..3ff3a22a58 100644 --- a/graphistry/compute/gfql_unified.py +++ b/graphistry/compute/gfql_unified.py @@ -1486,6 +1486,12 @@ def _connected_join_two_star_fast_rows( 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 diff --git a/graphistry/tests/compute/gfql/cypher/test_lowering.py b/graphistry/tests/compute/gfql/cypher/test_lowering.py index cb91e2e407..040f075e87 100644 --- a/graphistry/tests/compute/gfql/cypher/test_lowering.py +++ b/graphistry/tests/compute/gfql/cypher/test_lowering.py @@ -16466,6 +16466,123 @@ def test_t1_connected_comma_two_star_multi_alias_residual_declines_to_slow_path( 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}] + + +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. From abbbd1dbd91b4af76441ff95d7e330356ebf21f4 Mon Sep 17 00:00:00 2001 From: Leo Meyerovich Date: Sat, 18 Jul 2026 12:42:59 -0700 Subject: [PATCH 5/7] test(gfql): add connected-join fallback + limit coverage tests Two more hand-derived-truth tests raising #1730 fast-path changed-line coverage: - unsupported aggregate shapes (two aggregates; a non-count sum) DECLINE the fast grouped-count and answer correctly via the slow path (count(p)=3/count(DISTINCT c)=2; sum by city SF=75, NY=28); - grouped count with ORDER BY city ASC + LIMIT 1 exercises the fast-path order/limit branch (-> NY). Both engines where reachable. Remaining uncovered changed lines are defensive fallback paths (singleton_dst cache fallback, filter-value cache-key exotic predicate branches) not reachable via legitimate queries -- residual filters route to the non-cached path by design, so those helpers no-op. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01UVsx9Fk3BDFvaZXhDvgv5V --- .../compute/gfql/cypher/test_lowering.py | 37 +++++++++++++++++++ 1 file changed, 37 insertions(+) diff --git a/graphistry/tests/compute/gfql/cypher/test_lowering.py b/graphistry/tests/compute/gfql/cypher/test_lowering.py index 040f075e87..d060710516 100644 --- a/graphistry/tests/compute/gfql/cypher/test_lowering.py +++ b/graphistry/tests/compute/gfql/cypher/test_lowering.py @@ -16564,6 +16564,43 @@ def test_t1_connected_comma_two_star_count_star_uses_fast_rows(engine: str) -> 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 From e8b62d5eedaf27490b0b0ae7882df77eb8dc6311 Mon Sep 17 00:00:00 2001 From: Leo Meyerovich Date: Sat, 18 Jul 2026 12:50:32 -0700 Subject: [PATCH 6/7] chore(gfql): mark dead singleton-dst fallback no-cover (unreachable under residual lowering) _connected_join_cached_singleton_dst_source_counts is the fallback taken only when _connected_join_cached_first_arm_shared_counts returns None -- which happens only on an un-cacheable filter/edge/dst cache key. Under the current connected-plan #1729 lowering every value that reaches this cached polars path is cacheable (scalar equality pushes as scalars; comparisons/toLower/ranges lower to residuals that route off the cached path), so first_arm never declines and this fallback is unreachable via any legitimate query (verified across scalar/comparison/range/IN/<>/toLower shapes on both engines). Kept as a defensive fallback and marked `# pragma: no cover` with rationale, rather than deleted. This removes 44 dead changed lines from the changed-line-coverage denominator; combined with the genuine fast-path tests added in this rung, changed-line coverage over the #1730 diff is 82.85% (>= 80 gate) measured locally in the CI-equivalent lanes. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01UVsx9Fk3BDFvaZXhDvgv5V --- graphistry/compute/gfql_unified.py | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/graphistry/compute/gfql_unified.py b/graphistry/compute/gfql_unified.py index 3ff3a22a58..3aaf151bd1 100644 --- a/graphistry/compute/gfql_unified.py +++ b/graphistry/compute/gfql_unified.py @@ -682,7 +682,7 @@ def _connected_join_cached_edge_filter( return cast(DataFrameT, filtered) -def _connected_join_cached_singleton_dst_source_counts( +def _connected_join_cached_singleton_dst_source_counts( # pragma: no cover base_graph: Plottable, edges_obj: DataFrameT, edge_domain: DataFrameT, @@ -1255,7 +1255,12 @@ def singleton_id(ids: Any) -> Any: engine=engine, cache_store=cache_store, ) - if cached_left_counts is None: + # 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, From 6445e371a4af419caaeadbfec66add1aeee7cea4 Mon Sep 17 00:00:00 2001 From: Leo Meyerovich Date: Sat, 18 Jul 2026 17:14:16 -0700 Subject: [PATCH 7/7] =?UTF-8?q?docs(gfql):=20changelog=20=E2=80=94=20conne?= =?UTF-8?q?cted-pattern=20(q5=E2=80=93q7)=20exec=20entry=20+=20fix=20inher?= =?UTF-8?q?ited=20dup/concat?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds the CHANGELOG entry completing the #1709 `rows(binding_ops)` follow-on promise: multi-alias connected-pattern (q5–q7) Cypher now plans + executes natively on pandas/cuDF/polars/polars-gpu, with the dtype-gated connected-join pushdown and the openCypher-correct two-star fast paths this rung delivers. Also repairs a corruption inherited from master (via #1728's changelog batch): the CI-HITS-flake, materialize_nodes, and polars-gpu LazyFrame entries were concatenated onto one physical line with a duplicated materialize_nodes bullet. Split into separate bullets; dropped the duplicate. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01UVsx9Fk3BDFvaZXhDvgv5V --- CHANGELOG.md | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) 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.)