diff --git a/bin/ci_cypher_surface_guard_baseline.json b/bin/ci_cypher_surface_guard_baseline.json index 046156097c..72fc38aa20 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": 8779 + "lowering_py_max_lines": 9237 } diff --git a/graphistry/compute/filter_by_dict.py b/graphistry/compute/filter_by_dict.py index 1a28b71b02..837d4f9096 100644 --- a/graphistry/compute/filter_by_dict.py +++ b/graphistry/compute/filter_by_dict.py @@ -1,11 +1,11 @@ -from typing import Any, Dict, Optional, Tuple, Union +from typing import Any, Dict, Mapping, Optional, Tuple, Union, cast import pandas as pd from graphistry.Engine import EngineAbstract, df_to_engine, resolve_engine from graphistry.util import setup_logger from graphistry.Plottable import Plottable from .predicates.ASTPredicate import ASTPredicate -from .typing import DataFrameT +from .typing import DataFrameT, DType, NodeDtypes logger = setup_logger(__name__) @@ -219,3 +219,119 @@ def filter_edges_by_dict(self: Plottable, filter_dict: Optional[dict] = None, en """ edges2 = filter_by_dict(self._edges, filter_dict, engine) return self.edges(edges2) + + +class _LazyNodeDtypes(Mapping[str, DType]): + """Node dtypes, materialized on first lookup. + + Reading them costs a full engine conversion, and only connected-join pushdown ever asks -- + a path most queries never reach. Computing eagerly charged every string query for a frame + it then discarded (~65ms per million polars rows). Stay a Mapping so callers are unchanged. + """ + + def __init__(self, g: Plottable, engine: Union[EngineAbstract, str]) -> None: + self._g = g + self._engine = engine + self._resolved: Optional[NodeDtypes] = None + + def _materialize(self) -> NodeDtypes: + if self._resolved is None: + self._resolved = _read_node_dtypes(self._g, self._engine) + return self._resolved + + def __getitem__(self, key: str) -> DType: + return self._materialize()[key] + + def __iter__(self) -> Any: + return iter(self._materialize()) + + def __len__(self) -> int: + return len(self._materialize()) + + +def _node_dtypes_for_pushdown( + g: Plottable, + engine: Union[EngineAbstract, str] = EngineAbstract.AUTO, +) -> Optional[NodeDtypes]: + """Node dtypes for the pushdown gate, or None when there is no graph to read.""" + if g._nodes is None: + # No graph: the caller falls back to value-type rules, as a bare compile_cypher() does. + # Distinct from failing to read a graph we have, which must fail closed. + return None + return _LazyNodeDtypes(g, engine) + + +def _object_column_holds_non_strings(frame: DataFrameT, column: str, dtype: DType) -> bool: + """Whether `column` is an object column whose values `.str` would reject. + + `object` says nothing about contents: pandas stores ordinary strings that way, and also a + numeric column that acquired a `None`. String predicates are admitted on dtype alone -- by + this gate and by `filter_by_dict` alike -- but `.str` fails on the values, leaking a raw + AttributeError. Dropping the column leaves the gate with nothing to look up, so it fails + closed and the residual answers. + + Keep only the kinds that stay valid under SUBSETTING. `StringMethods._validate` admits + {string, empty, bytes, mixed, mixed-integer}, but it runs on the frame it is handed -- and + the frame we inspect is the source, while the pushed filter runs on the join's candidate + subset. `string` and `empty` survive that (a subset of strings is string or empty), but + `mixed`/`mixed-integer` do not: drop the strings and they collapse to integer/floating/ + boolean, which `.str` rejects. `bytes` is excluded separately -- it passes `_validate` yet + `str.contains` forbids it. + + Mirroring a rule is not enough when we evaluate it against a different frame than the + executor does, so admit only what no subset can invalidate. + """ + import pandas as _pd + + try: + is_object = bool(_pd.api.types.is_object_dtype(dtype)) + except Exception: + return False + if not is_object: + return False + try: + inferred = _pd.api.types.infer_dtype(frame[column], skipna=True) + except Exception: + # It IS an object column and we cannot read its values, so we cannot tell whether + # `.str` would reject them. Omit it and let the residual answer, matching + # `_read_node_dtypes`, which returns {} for a schema it cannot read. + return True + return inferred not in {"string", "empty"} + + +def _read_node_dtypes( + g: Plottable, + engine: Union[EngineAbstract, str] = EngineAbstract.AUTO, +) -> NodeDtypes: + """Column dtypes the executor will filter on, for the pushdown gate. + + A pushed filter is schema-validated against the real column while the row residual + evaluates leniently, so the planner needs dtypes to avoid turning a correct empty result + into a type error. An unreadable schema yields an empty mapping, which fails every column + lookup and so falls back to the residual. + """ + nodes = g._nodes + if nodes is None: + return {} + try: + # Classify the frame the EXECUTOR will filter, not the one the caller handed in. + # `filter_by_dict` validates post-materialization dtypes, so judging the pre-conversion + # frame lets the two disagree (polars Decimal reads numeric here, object there). + # + # Convert the whole frame, not an empty probe: polars -> pandas is DATA-dependent, so + # `head(0)` reports a different class than the real conversion for a nullable Boolean + # (`bool` empty, `object` with a null in it). Identity for pandas; for other engines + # this costs a real conversion, which is why the caller defers it until a pushdown + # decision actually needs it. + probe = df_to_engine(nodes, resolve_engine(cast(Any, engine), nodes), warn=False) + # zip rather than `.items()`: pandas/cuDF expose a column-indexed Series, polars a list. + dtypes = {str(col): dtype for col, dtype in zip(list(probe.columns), list(probe.dtypes))} + return { + col: dtype + for col, dtype in dtypes.items() + if not _object_column_holds_non_strings(probe, col, dtype) + } + except Exception: + # We have a graph but could not read its schema, so we know nothing about any column. + # An empty mapping fails every column lookup, which fails closed to the residual. + return {} diff --git a/graphistry/compute/gfql/cypher/lowering.py b/graphistry/compute/gfql/cypher/lowering.py index 5606a84b0d..7f1b56fbd2 100644 --- a/graphistry/compute/gfql/cypher/lowering.py +++ b/graphistry/compute/gfql/cypher/lowering.py @@ -109,9 +109,11 @@ SourceSpan, UnwindClause, WhereClause, + WhereOp, WherePredicate, WherePatternPredicate, ) +from graphistry.compute.typing import DType, NodeDtypes from graphistry.compute.gfql.cypher._boolean_expr_text import boolean_expr_to_text from graphistry.compute.gfql.cypher.expression_text import ( cypher_literal_expr_text as _cypher_literal_expr_text, @@ -7677,6 +7679,7 @@ class ConnectedMatchJoinPlan: pattern_chains: Tuple[Chain, ...] pattern_shared_node_aliases: Tuple[Tuple[str, ...], ...] post_join_chain: Chain + pattern_attach_prop_aliases: Tuple[Optional[Tuple[str, ...]], ...] = () def _pattern_alias_lists(pattern: Sequence[PatternElement]) -> Tuple[List[str], List[str]]: @@ -7772,15 +7775,454 @@ def _next_id() -> int: ) +def _connected_join_literal_op(op: str, *, reverse: bool = False) -> Optional[WhereOp]: + normalized = "==" if op == "=" else op + if normalized not in {"==", "!=", "<", "<=", ">", ">="}: + return None + if not reverse: + return cast(WhereOp, normalized) + return cast(WhereOp, { + "==": "==", + "!=": "!=", + "<": ">", + "<=": ">=", + ">": "<", + ">=": "<=", + }[normalized]) + + +def _connected_join_expr_property_ref( + node: ExprNode, + *, + span: SourceSpan, +) -> Optional[PropertyRef]: + if isinstance(node, PropertyAccessExpr) and isinstance(node.value, Identifier): + if "." in node.value.name: + return None + return PropertyRef(alias=node.value.name, property=node.property, span=span) + if isinstance(node, Identifier): + parts = node.name.split(".") + if len(parts) == 2 and all(parts): + return PropertyRef(alias=parts[0], property=parts[1], span=span) + return None + + +def _connected_join_expr_literal_value(node: ExprNode) -> Tuple[bool, Optional[CypherLiteral]]: + # `NUMBER` carries an optional sign as a lexer terminal, so a signed literal only + # folds when the sign is lexically adjacent (`-1`). With a space or parens + # (`- 1`, `-(1)`) the unary survives as a node and must be folded here, or the + # comparison silently falls back to a row residual instead of pushing down. + if isinstance(node, ExprLiteral): + return True, cast(Optional[CypherLiteral], node.value) + if isinstance(node, UnaryOp) and node.op in {"+", "-"} and isinstance(node.operand, ExprLiteral): + value = node.operand.value + if isinstance(value, (int, float)) and not isinstance(value, bool): + return True, cast(CypherLiteral, value if node.op == "+" else -value) + return False, None + + +_CONNECTED_JOIN_STRING_OPS = frozenset({"contains", "starts_with", "ends_with", "regex"}) +_CONNECTED_JOIN_ORDERING_OPS = frozenset({"!=", "<", "<=", ">", ">="}) +def _connected_join_dtype_classes(dtype: Any) -> Tuple[bool, bool]: + """Classify `dtype` as (numeric, string) using the validator's own helpers, verbatim. + + `_node_dtypes_for_pushdown` hands us the dtype the executor will actually filter on, so + asking `filter_by_dict`'s helpers is asking the executor itself -- planner and validator + agree by construction, and anything they both decline fails closed at the caller. + + Deliberately no kind/text fallback. `pd.api.types.is_numeric_dtype` returns False without + raising both for a dtype pandas cannot parse and for one it authoritatively knows is not + numeric, so a "both False" fallback cannot tell those apart -- and overriding the + authoritative answer is what made Arrow bool/dictionary columns raise where the residual + answers. + """ + from graphistry.compute.filter_by_dict import _is_numeric_dtype_safe, _is_string_dtype_safe + + return bool(_is_numeric_dtype_safe(dtype)), bool(_is_string_dtype_safe(dtype)) + + +def _connected_join_dtype_widens(dtype: Any) -> bool: + """Whether the join can change `dtype`'s class before the filter runs. + + The gate reads the source nodes frame, but the executor filters a joined one, and an + unmatched row introduces NaN. `bool` cannot hold NaN so pandas widens it to `object` -- + numeric to the gate, string to the validator. `int64` widens to `float64`, which stays + numeric and is harmless; `bool` is the only common dtype that crosses the boundary. + """ + import pandas as _pd + + try: + return bool(_pd.api.types.is_bool_dtype(dtype)) and not bool(_pd.api.types.is_extension_array_dtype(dtype)) + except Exception: + return False + + +def _connected_join_dtype_admits(op: WhereOp, value: Optional[Union[str, int, float, bool]], dtype: DType) -> bool: + """Whether pushing `op`/`value` onto a column of `dtype` matches residual semantics. + + A pushed `filter_dict` is schema-validated against the real column, while the row + residual evaluates leniently (Cypher 3-valued logic yields no rows). Pushing a + type-incompatible atom therefore turns a correct empty result into a hard error, so + those atoms have to stay residual. + + Mirrors the validator that actually runs, `compute/filter_by_dict.py`, and reuses its + dtype helpers: they agree by construction (`bool` counts as numeric there) and stay + correct on non-pandas dtypes. `compute/validate_schema.py` looks similar but is dead + code -- only a docs test imports it -- and it disagrees on `bool`. + """ + if _connected_join_dtype_widens(dtype): + return False + is_numeric_col, is_string_col = _connected_join_dtype_classes(dtype) + if not is_numeric_col and not is_string_col: + # Unrecognized dtype: fail closed. Pushing on a guess is how a correct empty + # result becomes a type error. + return False + if op in _CONNECTED_JOIN_STRING_OPS: + return is_string_col + if op in _CONNECTED_JOIN_ORDERING_OPS: + # These lower to NumericASTPredicate, rejected on a non-numeric column. + return is_numeric_col + if isinstance(value, str): + return not is_numeric_col + if isinstance(value, bool): + return True + if isinstance(value, (int, float)): + return not is_string_col + return True + + +def _connected_join_pushable_value( + op: WhereOp, + value: Optional[CypherLiteral], + *, + params: Optional[Mapping[str, Any]], + column: Optional[str] = None, + node_dtypes: Optional[NodeDtypes] = None, +) -> bool: + """Whether `op`/`value` survives the trip into a node `filter_dict`. + + Pushdown is an optimization: anything not exactly representable must stay a + `where_rows` residual rather than push a filter that means something else. + + - `None` is unrepresentable: `_filter_dict_to_json` drops null-valued entries, so a + pushed `nick = null` would vanish on the executor's serialization round-trip and + silently return unfiltered rows. + - Ordering/inequality ops lower to `NumericASTPredicate`, which admits only int/float + (`bool` is an `int` subclass but is not a numeric column predicate), so pushing a + string or bool would raise where the residual answers correctly. + + Parameters must be resolved first: `$v` arrives as an unresolved `ParameterRef`, and + judging that ref instead of its value is how `nick = $v`/`{'v': None}` slipped through. + An unresolvable ref stays a residual, which reports the missing parameter itself. + """ + try: + resolved = _resolve_literal(cast(CypherLiteral, value), params=params, field="where") if value is not None else None + except GFQLValidationError: + return False + if resolved is None: + return False + if isinstance(resolved, (dict, list, tuple, set)): + # A dict survives `_filter_dict_to_json` verbatim and `maybe_filter_dict_from_json` + # revives it via `predicates_from_json`, so a param like {'type': 'GT', 'val': 26} + # would execute as `> 26` instead of an equality against a map. + return False + if ( + isinstance(resolved, int) + and not isinstance(resolved, bool) + and not (_CYPHER_INT64_MIN <= resolved <= _CYPHER_INT64_MAX) + ): + # The 64-bit literal guard lives on the row-expr path, so pushing an out-of-range + # int would evade it and reach pandas, which overflows with a raw OverflowError. + return False + if op in _CONNECTED_JOIN_STRING_OPS: + if not isinstance(resolved, str): + return False + elif op == "==": + if not isinstance(resolved, (str, int, float, bool)): + return False + elif not (isinstance(resolved, (int, float)) and not isinstance(resolved, bool)): + return False + if node_dtypes is None: + # No schema (e.g. bare `compile_cypher()` with no graph). Keep the value-type + # decision; execution always goes through `gfql()`, which supplies dtypes. + return True + if column is None or column not in node_dtypes: + return False + return _connected_join_dtype_admits(op, resolved, node_dtypes[column]) + + +def _connected_join_mergeable_filter(target: ASTNode, prop: str, *, op: WhereOp, resolved: Optional[Union[str, int, float, bool]]) -> bool: + """Whether pushing `op`/`resolved` onto `prop` can merge with what is already there. + + `_merge_filter_predicates` wraps raw scalars with `comparison.eq`, which serializes to + `{'type': 'EQ'}` -- but `predicates/from_json.py` binds that tag to `predicates.numeric.EQ`, + which admits only int/float. Merging a non-numeric scalar on EITHER side therefore builds + a filter that raises when the executor rehydrates it. That write/read split is not ours to + fix here; just don't create it. + + Both sides must be checked. `filter_dict` is mutated as earlier atoms in the same WHERE + push, so the existing value may be an inline scalar OR a predicate a previous push left + behind -- checking only the existing side made this order-dependent + (`= 'a' AND CONTAINS 'x'` blocked, `CONTAINS 'x' AND = 'a'` not). + """ + existing_filter = _target_filter_dict(target) or {} + if prop not in existing_filter: + return True + existing = existing_filter[prop] + if not (isinstance(existing, ASTPredicate) or (isinstance(existing, (int, float)) and not isinstance(existing, bool))): + return False + # `_predicate_value` keeps `==` as a raw value; every other supported op builds a real + # predicate, which round-trips on its own tag. + if op == "==" and not isinstance(resolved, (int, float)): + return False + return True + + +def _apply_connected_join_node_filter( + alias_targets_by_pattern: Sequence[Mapping[str, ASTObject]], + *, + prop_ref: PropertyRef, + op: WhereOp, + value: Optional[CypherLiteral], + params: Optional[Mapping[str, Any]], + node_dtypes: Optional[NodeDtypes] = None, +) -> bool: + if not _connected_join_pushable_value( + op, value, params=params, column=prop_ref.property, node_dtypes=node_dtypes + ): + return False + try: + resolved_value = ( + _resolve_literal(cast(CypherLiteral, value), params=params, field="where") + if value is not None + else None + ) + except GFQLValidationError: + return False + for alias_targets in alias_targets_by_pattern: + target = alias_targets.get(prop_ref.alias) + if not isinstance(target, ASTNode): + continue + if not _connected_join_mergeable_filter(target, prop_ref.property, op=op, resolved=resolved_value): + return False + pushed = False + for alias_targets in alias_targets_by_pattern: + target = alias_targets.get(prop_ref.alias) + if not isinstance(target, ASTNode): + continue + _apply_literal_where( + cast(Dict[str, ASTObject], alias_targets), + left=prop_ref, + op=op, + right=value, + params=params, + ) + pushed = True + return pushed + + +def _pushdown_connected_join_atom_filter( + expr: ExpressionText, + alias_targets_by_pattern: Sequence[Mapping[str, ASTObject]], + alias_targets: Mapping[str, ASTObject], + *, + params: Optional[Mapping[str, Any]], + node_dtypes: Optional[NodeDtypes] = None, +) -> bool: + try: + node = _parse_row_expr( + expr.text, + params=params, + alias_targets=alias_targets, + field="where", + line=expr.span.line, + column=expr.span.column, + ) + except GFQLValidationError: + return False + if not isinstance(node, BinaryOp): + return False + + op = _connected_join_literal_op(node.op) + if op is not None: + left_ref = _connected_join_expr_property_ref(node.left, span=expr.span) + right_is_literal, right_value = _connected_join_expr_literal_value(node.right) + if left_ref is not None and right_is_literal: + return _apply_connected_join_node_filter( + alias_targets_by_pattern, + prop_ref=left_ref, + op=op, + value=right_value, + params=params, + node_dtypes=node_dtypes, + ) + reverse_op = _connected_join_literal_op(node.op, reverse=True) + right_ref = _connected_join_expr_property_ref(node.right, span=expr.span) + left_is_literal, left_value = _connected_join_expr_literal_value(node.left) + if right_ref is not None and left_is_literal and reverse_op is not None: + return _apply_connected_join_node_filter( + alias_targets_by_pattern, + prop_ref=right_ref, + op=reverse_op, + value=left_value, + params=params, + node_dtypes=node_dtypes, + ) + + return False + + +def _pushdown_connected_join_where_filters( + where: Optional[WhereClause], + alias_targets_by_pattern: Sequence[Mapping[str, ASTObject]], + alias_targets: Mapping[str, ASTObject], + *, + params: Optional[Mapping[str, Any]], + node_dtypes: Optional[NodeDtypes] = None, +) -> Optional[List[ExpressionText]]: + if where is None: + return [] + + # Push structured predicates on the expr_tree parser path. When no expr_tree is + # present, the residual loop below treats predicates as an implicit AND list and + # applies pushdown exactly once. + if where.expr_tree is not None: + for predicate in where.predicates: + if not isinstance(predicate, WherePredicate): + continue + if isinstance(predicate.left, LabelRef): + for pattern_targets in alias_targets_by_pattern: + target = pattern_targets.get(predicate.left.alias) + if isinstance(target, ASTNode): + _apply_label_where(cast(Dict[str, ASTObject], pattern_targets), left=predicate.left) + continue + if isinstance(predicate.right, PropertyRef): + continue + if not isinstance(predicate.left, PropertyRef): + continue + _apply_connected_join_node_filter( + alias_targets_by_pattern, + prop_ref=predicate.left, + op=predicate.op, + value=cast(Optional[CypherLiteral], predicate.right), + params=params, + node_dtypes=node_dtypes, + ) + + residuals: List[ExpressionText] = [] + + def _walk(expr_node: BooleanExpr) -> None: + if expr_node.op == "and" and expr_node.left is not None and expr_node.right is not None: + _walk(expr_node.left) + _walk(expr_node.right) + return + if expr_node.op == "atom" and expr_node.atom_text is not None: + atom = ExpressionText(text=expr_node.atom_text, span=expr_node.atom_span or expr_node.span) + pushed = _pushdown_connected_join_atom_filter( + atom, + alias_targets_by_pattern, + alias_targets, + params=params, + node_dtypes=node_dtypes, + ) + if not pushed: + residuals.append(atom) + return + synthesized = _where_clause_expr_text(where) + if synthesized is not None: + residuals.append(synthesized) + + if where.expr_tree is not None: + _walk(where.expr_tree) + return residuals + + for predicate in where.predicates: + if not isinstance(predicate, WherePredicate): + return None + pushed = False + if isinstance(predicate.left, LabelRef): + for pattern_targets in alias_targets_by_pattern: + target = pattern_targets.get(predicate.left.alias) + if isinstance(target, ASTNode): + _apply_label_where(cast(Dict[str, ASTObject], pattern_targets), left=predicate.left) + pushed = True + elif isinstance(predicate.left, PropertyRef) and not isinstance(predicate.right, PropertyRef): + pushed = _apply_connected_join_node_filter( + alias_targets_by_pattern, + prop_ref=predicate.left, + op=predicate.op, + value=cast(Optional[CypherLiteral], predicate.right), + params=params, + node_dtypes=node_dtypes, + ) + if pushed: + continue + row_text = _row_where_predicate_text(predicate) + if row_text is None: + return None + residuals.append(ExpressionText(text=row_text, span=where.span)) + return residuals + + +def _connected_join_required_property_aliases( + query: CypherQuery, + *, + alias_targets: Mapping[str, ASTObject], + residual_filters: Sequence[ExpressionText], + params: Optional[Mapping[str, Any]], +) -> Optional[Set[str]]: + if query.with_stages: + return None + node_aliases = {alias for alias, target in alias_targets.items() if isinstance(target, ASTNode)} + required: Set[str] = set() + + exprs: List[Tuple[str, int, int, str]] = [] + for expr in residual_filters: + exprs.append((expr.text, expr.span.line, expr.span.column, "where")) + for item in query.return_.items: + exprs.append((item.expression.text, item.span.line, item.span.column, query.return_.kind)) + if query.order_by is not None: + for order_item in query.order_by.items: + exprs.append((order_item.expression.text, order_item.span.line, order_item.span.column, "order_by")) + + for text, line, column, field_name in exprs: + if text == "*": + continue + try: + non_aggregate_aliases, _aggregate_aliases = _expr_match_alias_usage( + text, + alias_targets=alias_targets, + params=params, + field=field_name, + line=line, + column=column, + ) + prop_aliases = _expr_property_access_node_aliases( + text, + alias_targets=alias_targets, + params=params, + field=field_name, + line=line, + column=column, + ) + except GFQLValidationError: + return None + required.update(alias for alias in non_aggregate_aliases if alias in node_aliases) + required.update(alias for alias in prop_aliases if alias in node_aliases) + return required + + def _compile_connected_match_join( query: CypherQuery, *, params: Optional[Mapping[str, Any]] = None, semantic_entity_kinds: Optional[Mapping[str, Literal["node", "edge", "scalar"]]] = None, + node_dtypes: Optional[NodeDtypes] = None, ) -> CompiledCypherQuery: clause = query.matches[0] pattern_chains: List[Chain] = [] pattern_node_aliases: List[Set[str]] = [] + alias_targets_by_pattern: List[Mapping[str, ASTObject]] = [] combined_alias_targets: Dict[str, ASTObject] = {} pre_join_filters: List[ExpressionText] = [] @@ -7794,6 +8236,7 @@ def _compile_connected_match_join( ) ops, duplicate_alias_where = _lower_match_clause_with_alias_equalities(single_clause, params=params) alias_targets = _alias_target(ops) + alias_targets_by_pattern.append(alias_targets) dynamic_where_out, dynamic_row_preds = _dynamic_property_entry_constraints( single_clause, alias_targets=alias_targets, @@ -7825,15 +8268,14 @@ def _compile_connected_match_join( accumulated_aliases.update(node_aliases) if query.where is not None: - # #1712: a WHERE on a connected comma-pattern join must be applied as a - # post-join row filter. Previously this only handled the ``expr_tree`` form - # and SILENTLY DROPPED the structured ``predicates`` form (e.g. a plain - # ``WHERE i.k = 'x'``), returning an unfiltered — wrong — result. - # ``_where_clause_expr_text`` renders BOTH forms; a WHERE it can't render - # (has-labels / outside the row-renderable subset) must NIE honestly, never - # be dropped. - synthesized = _where_clause_expr_text(query.where) - if synthesized is None: + residual_filters = _pushdown_connected_join_where_filters( + query.where, + alias_targets_by_pattern, + combined_alias_targets, + params=params, + node_dtypes=node_dtypes, + ) + if residual_filters is None: raise _unsupported( "Cypher connected comma-pattern join lowering cannot render this WHERE clause " "to a row filter; use engine='pandas' with a supported WHERE shape", @@ -7842,7 +8284,20 @@ def _compile_connected_match_join( line=clause.span.line, column=clause.span.column, ) - pre_join_filters.append(synthesized) + pre_join_filters.extend(residual_filters) + + required_prop_aliases = _connected_join_required_property_aliases( + query, + alias_targets=combined_alias_targets, + residual_filters=pre_join_filters, + params=params, + ) + pattern_attach_prop_aliases: List[Optional[Tuple[str, ...]]] = [] + for pattern_alias_targets in alias_targets_by_pattern: + if required_prop_aliases is None: + pattern_attach_prop_aliases.append(None) + else: + pattern_attach_prop_aliases.append(tuple(sorted(required_prop_aliases & set(pattern_alias_targets.keys())))) for projection_clause in [stage.clause for stage in query.with_stages] + [query.return_]: _reject_unsupported_connected_join_clause_shapes( @@ -7950,6 +8405,7 @@ def _compile_connected_match_join( pattern_chains=tuple(pattern_chains), pattern_shared_node_aliases=tuple(shared_aliases_per_pattern), post_join_chain=Chain(row_steps), + pattern_attach_prop_aliases=tuple(pattern_attach_prop_aliases), ), query_graph=query_graph, logical_plan=logical_plan, @@ -8358,6 +8814,7 @@ def compile_cypher_query( query: Union[CypherQuery, CypherUnionQuery, CypherGraphQuery], *, params: Optional[Mapping[str, Any]] = None, + node_dtypes: Optional[NodeDtypes] = None, ) -> Union[CompiledCypherQuery, CompiledCypherUnionQuery, CompiledCypherGraphQuery]: from graphistry.compute.gfql.cypher import projection_planning as _projection @@ -8510,6 +8967,7 @@ def _attach_graph_context(result: CompiledCypherQuery) -> CompiledCypherQuery: query, params=params, semantic_entity_kinds=bound_context.entity_kinds, + node_dtypes=node_dtypes, ) ) if _is_connected_optional_match_query(query): diff --git a/graphistry/compute/gfql/row/pipeline.py b/graphistry/compute/gfql/row/pipeline.py index 836bbf3097..6c774ff86c 100644 --- a/graphistry/compute/gfql/row/pipeline.py +++ b/graphistry/compute/gfql/row/pipeline.py @@ -3787,6 +3787,75 @@ def _gfql_connected_bindings_row_table_from_ops( setattr(out, "_gfql_rows_edge_aliases", edge_aliases) return out + def _gfql_add_missing_binding_columns( + self, bindings: DataFrameT, ops: Sequence["ASTObject"] + ) -> DataFrameT: + """Ensure an EMPTY bindings frame still carries every declared alias column. + + The walk assembles the schema incrementally, so an emptied hop returns a frame that + never acquired the columns later steps would have added -- edge-alias columns + (``e1.w``) and ``alias.alias`` markers. Node aliases survive via the row-frame merge's + empty lookup, but edge aliases have no ``.id`` fallback, so ``count(e1)`` / + ``e1.w``-in-WHERE dereferenced a missing column. Declared columns are derivable from + ``ops`` + the base edge frame, so add whatever is missing here (0-row only, no effect + on non-empty output). See #25/#27. + """ + from graphistry.compute.ast import ASTEdge, ASTNode + + base_graph = self._gfql_base_graph() + edges = base_graph._edges if base_graph is not None else None + src_col = base_graph._source if base_graph is not None else None + dst_col = base_graph._destination if base_graph is not None else None + # (column_name, source_series) — carry the source dtype so an edge property like + # `e1.w` stays int64 at 0 rows instead of an untyped object column (which would upcast + # a sum/avg and escape via UNION ALL). Mirrors the node path's typed empty slice. The + # `alias.alias` marker has no source dtype, so None-broadcast is fine for it. + missing: List[Tuple[str, Optional[Any]]] = [] + for op in ops: + alias = getattr(op, "_name", None) + if not isinstance(alias, str): + continue + missing.append((f"{alias}.{alias}", None)) # the alias.alias marker + if isinstance(op, ASTEdge) and edges is not None: + for col in edges.columns: + if col in (src_col, dst_col): + continue + missing.append((f"{alias}.{col}", edges[col])) + for col, source in missing: + if col not in bindings.columns: + if source is not None: + bindings[col] = source.iloc[0:0].reindex(bindings.index) + else: + bindings[col] = self._gfql_broadcast_scalar(bindings, None) + # Downstream node aliases (every node op after the first) reach the row via a hop + # whose unmatched rows introduce NaN, so the non-empty path widens the columns that + # cannot hold NaN: numpy int -> float64, numpy bool -> object. The 0-row path sourced + # them from the base node frame (int/bool), so match that widening or an emptied + # `sum(b.i)`/`max(b.bv)` returns int64/bool where the non-empty run and master give + # float64/object -- observable through UNION ALL (#31, Wave 37 Finding 2). Extension + # dtypes (`Int64`, `boolean`) hold NA natively and stay put in the non-empty path, so + # they must NOT be touched here (widening them re-introduced the divergence -- Wave 37 + # Finding 1). + import pandas as _pd + + node_ops = [op for op in ops if isinstance(op, ASTNode)] + for op in node_ops[1:]: + alias = getattr(op, "_name", None) + if not isinstance(alias, str): + continue + for col in [c for c in bindings.columns if c == alias or str(c).startswith(f"{alias}.")]: + try: + dtype = bindings[col].dtype + if _pd.api.types.is_extension_array_dtype(dtype): + continue + if dtype.kind in ("i", "u"): + bindings[col] = bindings[col].astype("float64") + elif dtype.kind == "b": + bindings[col] = bindings[col].astype("object") + except Exception: + continue + return bindings + def _gfql_connected_bindings_row_frame_from_state( self, ops: Sequence["ASTObject"], @@ -3848,6 +3917,8 @@ def _gfql_connected_bindings_row_frame_from_state( drop_cols = ["__current__"] bindings = bindings.drop(columns=[col for col in drop_cols if col in bindings.columns]) + if len(bindings) == 0: + bindings = self._gfql_add_missing_binding_columns(bindings, ops) return bindings def _gfql_shortest_path_scalar_native( diff --git a/graphistry/compute/gfql_unified.py b/graphistry/compute/gfql_unified.py index 0583073072..179c14c98c 100644 --- a/graphistry/compute/gfql_unified.py +++ b/graphistry/compute/gfql_unified.py @@ -27,7 +27,7 @@ parse_where_json, ) from graphistry.compute.exceptions import ErrorCode, GFQLValidationError -from graphistry.compute.gfql.cypher.api import compile_cypher +from graphistry.compute.gfql.cypher.parser import parse_cypher from graphistry.compute.gfql.cypher.lowering import ( ConnectedMatchJoinPlan, CompiledCypherGraphQuery, @@ -35,7 +35,9 @@ CompiledCypherUnionQuery, CompiledGraphResidualFilter, ConnectedOptionalMatchPlan, + compile_cypher_query, ) +from graphistry.compute.filter_by_dict import _node_dtypes_for_pushdown from graphistry.compute.gfql.cypher.reentry.execution import ( REENTRY_DUPLICATE_CARRIED_ROWS_REASON as _REENTRY_DUPLICATE_CARRIED_ROWS_REASON, REENTRY_WHOLE_ROW_SUGGESTION as _REENTRY_WHOLE_ROW_SUGGESTION, @@ -69,7 +71,7 @@ from graphistry.compute.gfql.passes import DEFAULT_LOGICAL_PASSES, DEFAULT_TIER2_PASSES, PassManager from graphistry.compute.gfql.row.pipeline import _RowPipelineAdapter, is_row_pipeline_call from graphistry.compute.gfql.search_any import search_any_mask -from graphistry.compute.typing import DataFrameT, SeriesT +from graphistry.compute.typing import DataFrameT, SeriesT, NodeDtypes from graphistry.compute.util.generate_safe_column_name import generate_safe_column_name from graphistry.compute.validate.validate_schema import validate_chain_schema from graphistry.compute.gfql_validate import gfql_validate as gfql_preflight_validate @@ -493,11 +495,14 @@ def _apply_connected_match_join( ) pattern_result = _chain_dispatch(base_graph, with_rows, dispatch_engine, policy, context) pattern_rows = cast(Optional[DataFrameT], pattern_result._nodes) - if pattern_rows is None or len(pattern_rows) == 0: + if pattern_rows is None: out = base_graph.bind() out._nodes = df_ctor() out._edges = df_ctor() return out + # The rows op now emits the full binding schema even at 0 rows (#25), so an emptied + # pattern carries its columns and flows through post_join_chain -- which is where the + # aggregate RETURN lives. Short-circuiting here dropped that column. pattern_rows = cast(DataFrameT, pattern_rows[_binding_join_columns(pattern_rows)]) if joined_rows is None: joined_rows = pattern_rows @@ -526,7 +531,7 @@ def _apply_connected_match_join( engine=requested_engine, ) - if joined_rows is None or len(joined_rows) == 0: + if joined_rows is None: out = base_graph.bind() out._nodes = df_ctor() out._edges = df_ctor() @@ -1346,6 +1351,7 @@ def _compile_string_query( *, language: Optional[Literal["cypher", "gremlin"]], params: Optional[Mapping[str, Any]], + node_dtypes: Optional[NodeDtypes] = None, ) -> Any: query_language = language or "cypher" if query_language != "cypher": @@ -1357,7 +1363,7 @@ def _compile_string_query( suggestion="Use language='cypher' for now; Gremlin string compilation is not implemented yet.", language="gfql", ) - return compile_cypher(query, params=params, _warn_deprecated=False) + return compile_cypher_query(parse_cypher(query), params=params, node_dtypes=node_dtypes) def _compile_value_repr(value: Any) -> str: @@ -1653,7 +1659,12 @@ def gfql(self: Plottable, if isinstance(query, str): query_language = language or "cypher" try: - compiled_query = _compile_string_query(query, language=language, params=params) + compiled_query = _compile_string_query( + query, + language=language, + params=params, + node_dtypes=_node_dtypes_for_pushdown(self, engine), + ) except GFQLValidationError as exc: _fire_postcompile_policy( expanded_policy, diff --git a/graphistry/compute/typing.py b/graphistry/compute/typing.py index d17805fc0b..a72c287274 100644 --- a/graphistry/compute/typing.py +++ b/graphistry/compute/typing.py @@ -1,5 +1,5 @@ import pandas as pd -from typing import Any, Protocol, TYPE_CHECKING, Tuple, TypeVar, Union +from typing import Any, Mapping, Protocol, TYPE_CHECKING, Tuple, TypeVar, Union # TODO stubs for Union[cudf.DataFrame, dask.DataFrame, ..] at checking time if TYPE_CHECKING: @@ -13,6 +13,12 @@ IndexT = Any DomainT = Any +# Engine-polymorphic column dtype: numpy dtype / pandas ExtensionDtype / polars DataType. +# Honestly Any -- the concrete type is engine-dependent and only ever passed to dtype-inspection +# helpers that accept Any and fail closed. +DType = Any +NodeDtypes = Mapping[str, DType] + # Type variable for return type preservation in predicates T = TypeVar('T') diff --git a/graphistry/tests/compute/gfql/cypher/test_lowering.py b/graphistry/tests/compute/gfql/cypher/test_lowering.py index 57e2c30cda..6a11c5cadb 100644 --- a/graphistry/tests/compute/gfql/cypher/test_lowering.py +++ b/graphistry/tests/compute/gfql/cypher/test_lowering.py @@ -38,12 +38,15 @@ ) from graphistry.plugins.networkx.policy import NETWORKX_SCIPY_EXTRA_REQUIREMENTS, NETWORKX_VERSION_SPEC, SCIPY_VERSION_SPEC from graphistry.compute.gfql.cypher.ast import ExpressionText, OrderByClause, OrderItem, ReturnClause, ReturnItem, SourceSpan -from graphistry.compute.gfql.cypher.lowering import CompiledCypherExecutionExtras, CompiledCypherGraphQuery +from graphistry.compute.gfql.cypher.lowering import CompiledCypherExecutionExtras, CompiledCypherGraphQuery, compile_cypher_query from graphistry.compute.gfql.cypher.lowering import _logical_plan_route_for_query from graphistry.compute.gfql.frontends.cypher.binder import FrontendBinder from graphistry.compute.gfql.ir.bound_ir import BoundIR, BoundQueryPart, SemanticTable from graphistry.compute.gfql.ir.compilation import PlanContext from graphistry.compute.gfql.ir.logical_plan import CHILD_SLOTS, Filter, PatternMatch, ProcedureCall as LogicalProcedureCall +from graphistry.compute.filter_by_dict import _node_dtypes_for_pushdown +from graphistry.compute.gfql.cypher.lowering import _connected_join_dtype_admits, _connected_join_dtype_classes +from graphistry.Plottable import Plottable from graphistry.tests.test_compute import CGFull from graphistry.tests.compute.gfql.cypher._whole_entity_compat import entity_text_records @@ -14933,6 +14936,7 @@ 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 @@ -14946,6 +14950,1177 @@ def test_issue_1712_connected_comma_pattern_where_intersects() -> None: assert result._nodes.to_dict(orient="records") == [{"numPersons": 2}] +def _mk_graph_benchmark_t1_shape_graph() -> "_CypherTestGraph": + nodes = pd.DataFrame({ + "id": [0, 1, 2, 10, 11, 20, 21], + "node_type": ["Person", "Person", "Person", "City", "City", "Interest", "Interest"], + "gender": ["MALE", "female", "male", None, None, None, None], + "age": [25, 27, 35, None, None, None, None], + "city": [None, None, None, "London", "Paris", None, None], + "country": [None, None, None, "United Kingdom", "France", None, None], + "state": [None, None, None, "England", "Ile-de-France", None, None], + "interest": [None, None, None, None, None, "Fine Dining", "photography"], + }) + edges = pd.DataFrame({ + "s": [0, 1, 2, 0, 1, 2], + "d": [20, 20, 21, 10, 10, 11], + "rel": ["HAS_INTEREST", "HAS_INTEREST", "HAS_INTEREST", "LIVES_IN", "LIVES_IN", "LIVES_IN"], + }) + return _mk_graph(nodes, edges) + + +def _compiled_connected_join_filters(query: str) -> list[dict[str, Any]]: + compiled = _compile_query(query) + plan = _compiled_execution_extras(compiled).connected_match_join + assert plan is not None + out: list[dict[str, Any]] = [] + for chain in plan.pattern_chains: + for op in chain.chain: + if isinstance(op, ASTNode) and isinstance(op._name, str): + out.append({op._name: dict(op.filter_dict or {})}) + return out + + +def _compiled_connected_join_plan(query: str) -> Any: + compiled = _compile_query(query) + plan = _compiled_execution_extras(compiled).connected_match_join + assert plan is not None + return plan + + +def _post_join_functions(query: str) -> list[str]: + return [op.function for op in _compiled_connected_join_plan(query).post_join_chain.chain if isinstance(op, ASTCall)] + + +def _mk_graph_benchmark_t1_labelled_shape_graph() -> "_CypherTestGraph": + nodes = pd.DataFrame({ + "id": [0, 1, 2, 10, 11, 20, 21], + "node_type": ["Person", "Person", "Person", "City", "City", "Interest", "Interest"], + "age": [25, 27, 35, None, None, None, None], + "city": [None, None, None, "London", "Paris", None, None], + "interest": [None, None, None, None, None, "Fine Dining", "photography"], + "label__Person": [True, True, True, False, False, False, False], + "label__City": [False, False, False, True, True, False, False], + "label__Interest": [False, False, False, False, False, True, True], + }) + edges = pd.DataFrame({ + "s": [0, 1, 2, 0, 1, 2], + "d": [20, 20, 21, 10, 10, 11], + "rel": ["HAS_INTEREST", "HAS_INTEREST", "HAS_INTEREST", "LIVES_IN", "LIVES_IN", "LIVES_IN"], + }) + return _mk_graph(nodes, edges) + + +def test_t1_connected_comma_pushes_label_predicate_with_property_filter() -> None: + query = ( + "MATCH (p {node_type:'Person'})-[{rel:'HAS_INTEREST'}]->(i {node_type:'Interest'}), " + "(p)-[{rel:'LIVES_IN'}]->(c {node_type:'City'}) " + "WHERE p:Person AND p.age >= 26 " + "RETURN count(p) AS numPersons" + ) + + result = _mk_graph_benchmark_t1_labelled_shape_graph().gfql(query) + assert result._nodes.to_dict(orient="records") == [{"numPersons": 2}] + + filters_by_alias = _compiled_connected_join_filters(query) + assert any("label__Person" in entry.get("p", {}) for entry in filters_by_alias) + assert any("age" in entry.get("p", {}) for entry in filters_by_alias) + + +def test_t1_connected_comma_mixes_signed_literal_pushdown_and_in_residual() -> 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') " + "AND p.age >= -1 AND p.age IN [25, 27] " + "RETURN count(p) AS numPersons" + ) + + result = _mk_graph_benchmark_t1_shape_graph().gfql(query) + assert result._nodes.to_dict(orient="records") == [{"numPersons": 2}] + + plan = _compiled_connected_join_plan(query) + assert "where_rows" in _post_join_functions(query) + assert plan.pattern_attach_prop_aliases == (("i", "p"), ("p",)) + filters_by_alias = _compiled_connected_join_filters(query) + assert any("age" in entry.get("p", {}) for entry in filters_by_alias) + + +def _mk_graph_benchmark_t1_nullable_shape_graph() -> "_CypherTestGraph": + nodes = pd.DataFrame({ + "id": [0, 1, 2, 3, 10, 11, 20, 21], + "node_type": ["Person"] * 4 + ["City"] * 2 + ["Interest"] * 2, + "age": [25, 27, 35, None, None, None, None, None], + "nick": ["a", "b", "c", None, None, None, None, None], + "flag": [True, False, True, None, None, None, None, None], + "city": [None] * 4 + ["London", "Paris", None, None], + "interest": [None] * 6 + ["Fine Dining", "photography"], + }) + edges = pd.DataFrame({ + "s": [0, 1, 2, 3, 0, 1, 2, 3], + "d": [20, 20, 21, 20, 10, 10, 11, 11], + "rel": ["HAS_INTEREST"] * 4 + ["LIVES_IN"] * 4, + }) + return _mk_graph(nodes, edges) + + +def _real_t1_nullable_graph() -> Plottable: + # A real Plottable, not `_CypherTestGraph`: the executor's serialize/deserialize + # round-trip is where dropped-null and revived-predicate filters actually bite, and + # the test double does not reproduce it. + nodes = pd.DataFrame({ + "id": [0, 1, 2, 3, 10, 11, 20, 21], + "node_type": ["Person"] * 4 + ["City"] * 2 + ["Interest"] * 2, + "age": [25, 27, 35, None, None, None, None, None], + "nick": ["a", "b", "c", None, None, None, None, None], + "flag": [True, False, True, None, None, None, None, None], + "city": [None] * 4 + ["London", "Paris", None, None], + "interest": [None] * 6 + ["Fine Dining", "photography"], + }) + edges = pd.DataFrame({ + "s": [0, 1, 2, 3, 0, 1, 2, 3], + "d": [20, 20, 21, 20, 10, 10, 11, 11], + "rel": ["HAS_INTEREST"] * 4 + ["LIVES_IN"] * 4, + }) + return graphistry.nodes(nodes, "id").edges(edges, "s", "d") + + +def _t1_nullable_query(predicate: str) -> str: + return ( + "MATCH (p {node_type:'Person'})-[{rel:'HAS_INTEREST'}]->(i {node_type:'Interest'}), " + "(p)-[{rel:'LIVES_IN'}]->(c {node_type:'City'}) " + f"WHERE {predicate} " + "RETURN count(p) AS n" + ) + + +def test_connected_join_does_not_push_null_equality() -> None: + # `filter_dict` serialization drops null values, so a pushed `nick = null` would + # vanish and return every row unfiltered. It must stay a residual instead. + query = _t1_nullable_query("p.nick = $v") + + result = _mk_graph_benchmark_t1_nullable_shape_graph().gfql(query, params={"v": None}) + assert result._nodes.to_dict(orient="records") == [] + + compiled = cast(CompiledCypherQuery, compile_cypher(query, params={"v": None})) + plan = _compiled_execution_extras(compiled).connected_match_join + assert plan is not None + pushed_props = { + prop + for chain in plan.pattern_chains + for op in chain.chain + if isinstance(op, ASTNode) + for prop in (op.filter_dict or {}) + } + assert "nick" not in pushed_props + residuals = [op.function for op in plan.post_join_chain.chain if isinstance(op, ASTCall)] + assert "where_rows" in residuals + + +@pytest.mark.parametrize( + "predicate,expected", + [ + ("p.nick <> 'a'", 2), + ("i.interest <> 'photography'", 3), + ("p.nick < 'c'", 2), + ("p.flag >= true", 2), + ], +) +def test_connected_join_does_not_push_non_numeric_ordering(predicate: str, expected: int) -> None: + # Ordering/inequality ops lower to numeric-only predicates; pushing a string or bool + # raises, so these must fall back to a residual and still answer correctly. + query = _t1_nullable_query(predicate) + + result = _mk_graph_benchmark_t1_nullable_shape_graph().gfql(query) + assert result._nodes.to_dict(orient="records") == [{"n": expected}] + + assert "where_rows" in _post_join_functions(query) + + +@pytest.mark.parametrize( + "value", + [ + {"type": "GT", "val": 26}, + {"type": "Between", "lower": 0, "upper": 100, "inclusive": True}, + [25, 27], + ], +) +def test_connected_join_does_not_push_non_scalar_param(value: Any) -> None: + # `_filter_dict_to_json` passes a dict through verbatim and `maybe_filter_dict_from_json` + # revives it through `predicates_from_json`, so pushing a caller-supplied map would run + # an arbitrary predicate (`age > 26`) instead of an equality against a map. + # Must use a real Plottable: the serialization round-trip that resurrects the predicate + # only happens on the real executor, so `_CypherTestGraph` cannot observe this. + query = _t1_nullable_query("p.age = $v") + + result = _real_t1_nullable_graph().gfql(query, params={"v": value}) + assert result._nodes.to_dict(orient="records") == [] + + +@pytest.mark.parametrize( + "predicate,expected", + [ + ("p.nick STARTS WITH 'a'", 1), + ("p.nick CONTAINS 'a'", 1), + ("p.nick ENDS WITH 'a'", 1), + ("p.nick =~ 'a'", 1), + ], +) +def test_connected_join_pushes_string_predicates(predicate: str, expected: int) -> None: + # These lower to real round-trippable ASTPredicates, and the residual cannot render + # them at all, so they must push rather than fall back. + query = _t1_nullable_query(predicate) + + result = _mk_graph_benchmark_t1_nullable_shape_graph().gfql(query) + assert result._nodes.to_dict(orient="records") == [{"n": expected}] + + filters_by_alias = _compiled_connected_join_filters(query) + assert any("nick" in entry.get("p", {}) for entry in filters_by_alias) + + +@pytest.mark.parametrize( + "predicate,expected", + [ + # string literal vs numeric column, and numeric literal vs string column: the + # residual answers these leniently, so pushing them must not turn them into errors + ("p.age = 'foo'", []), + ("p.age = date('2020-01-01')", []), + ("p.nick >= 26", []), + ("p.nick < 26", []), + ("p.nick <> 26", [{"n": 3}]), + ], +) +def test_connected_join_does_not_push_dtype_incompatible_atoms(predicate: str, expected: Any) -> None: + query = _t1_nullable_query(predicate) + + result = _real_t1_nullable_graph().gfql(query) + assert result._nodes.to_dict(orient="records") == expected + + +def _real_bool_shape_graph() -> Plottable: + nodes = pd.DataFrame({ + "id": ["a", "b", "c", "d"], + "flag": pd.Series([True, False, True, False], dtype="bool"), + "age": pd.Series([25, 27, 35, 40], dtype="int64"), + }) + edges = pd.DataFrame({"s": ["a", "b", "c", "d"], "d": ["b", "c", "d", "a"]}) + return graphistry.nodes(nodes, "id").edges(edges, "s", "d") + + +@pytest.mark.parametrize( + "predicate,expected", + [ + # `bool` counts as numeric to the live validator, so a string value against a bool + # column must stay residual rather than push into a type error. + ("p.flag = 'yes'", []), + ("p.flag = true", [{"n": 2}]), + ("p.age >= 26", [{"n": 3}]), + ], +) +def test_connected_join_bool_column_matches_validator(predicate: str, expected: Any) -> None: + query = f"MATCH (p)-[]->(x), (p)-[]->(y) WHERE {predicate} RETURN count(p) AS n" + + result = _real_bool_shape_graph().gfql(query) + assert result._nodes.to_dict(orient="records") == expected + + +@pytest.mark.parametrize( + "predicate", + [ + "p.age = 99999999999999999999", + "p.age <> 99999999999999999999", + "p.flag = 99999999999999999999", + ], +) +def test_connected_join_out_of_range_int_still_reports_range_error(predicate: str) -> None: + # The 64-bit literal guard lives on the row-expr path; pushing an out-of-range int + # would evade it and surface a raw OverflowError from pandas instead. + query = f"MATCH (p)-[]->(x), (p)-[]->(y) WHERE {predicate} RETURN count(p) AS n" + + with pytest.raises(GFQLValidationError): + _real_bool_shape_graph().gfql(query) + + +@pytest.mark.parametrize( + "predicate,expected", + [("p.iv > 1", []), ("p.iv >= 1", []), ("p.iv <> 1", [{"n": 8}]), ("p.iv = 1", [])], +) +def test_connected_join_interval_column_matches_master(predicate: str, expected: Any) -> None: + # `interval[int64, right]` contains "int": classified numeric, a comparison pushed onto + # it raised a raw ValueError out of pandas where the residual answers correctly. + nodes = pd.DataFrame({"id": ["p1", "p2", "a1", "b1", "a2", "b2"]}) + nodes["iv"] = pd.arrays.IntervalArray.from_breaks([0, 1, 2, 3, 4, 5, 6]) + edges = pd.DataFrame({"s": ["p1", "p1", "p2", "p2"], "d": ["a1", "b1", "a2", "b2"]}) + g = graphistry.nodes(nodes, "id").edges(edges, "s", "d") + query = f"MATCH (p)-[]->(x), (p)-[]->(y) WHERE {predicate} RETURN count(p) AS n" + + assert g.gfql(query)._nodes.to_dict(orient="records") == expected + + +def _real_labelled_inline_graph() -> Plottable: + nodes = pd.DataFrame([ + {"id": "p1", "label__Person": True, "label__Place": False, "nick": "aa", "age": 30}, + {"id": "p2", "label__Person": True, "label__Place": False, "nick": "bb", "age": 40}, + {"id": "c1", "label__Person": False, "label__Place": True, "nick": None, "age": None}, + ]) + edges = pd.DataFrame([{"s": "p1", "d": "c1", "type": "L"}, {"s": "p2", "d": "c1", "type": "L"}]) + return graphistry.nodes(nodes, "id").edges(edges, "s", "d") + + +@pytest.mark.parametrize( + "inline,predicate,expected", + [ + # Merging onto an existing STRING inline value wraps it with comparison.eq, which + # serializes to {'type':'EQ'} -- a tag from_json binds to the numeric-only EQ, so the + # executor raises when it rehydrates. Don't create that shape; stay residual. + ("{nick:'aa'}", "friend.nick = 'aa'", [{"n": 1}]), + ("{nick:'aa'}", "friend.nick = 'bb'", []), + # A different property never merges, and a numeric inline value rehydrates fine. + ("{nick:'aa'}", "friend.age > 20", [{"n": 1}]), + ("{age:30}", "friend.age > 20", [{"n": 1}]), + ("{}", "friend.nick = 'aa'", [{"n": 1}]), + ], +) +def test_connected_join_inline_string_property_merge_matches_master( + inline: str, predicate: str, expected: Any +) -> None: + query = ( + "MATCH (person:Person {id:'p1'})-[:L]->(city:Place), " + f"(friend:Person {inline})-[:L]->(city) " + f"WHERE {predicate} RETURN count(friend) AS n" + ) + + result = _real_labelled_inline_graph().gfql(query) + assert result._nodes.to_dict(orient="records") == expected + + +def _real_string_merge_graph() -> Plottable: + nodes = pd.DataFrame({ + "id": ["p1", "p2", "p3", "aa", "bb"], + "name": ["alice", "bob", "carol", "aa", "bb"], + "age": [30, 40, 50, 1, 2], + }) + edges = pd.DataFrame({ + "s": ["p1", "p1", "p2", "p2", "p3", "p3"], + "d": ["aa", "bb", "aa", "bb", "aa", "bb"], + }) + return graphistry.nodes(nodes, "id").edges(edges, "s", "d") + + +@pytest.mark.parametrize( + "predicate,expected", + [ + # filter_dict is mutated as earlier atoms push, so a previously-pushed string + # predicate must not green-light merging a raw string behind it. Checking only the + # existing side made this order-dependent. + ("p.name CONTAINS 'al' AND p.name = 'alice'", [{"n": 4}]), + ("p.name STARTS WITH 'al' AND p.name = 'alice'", [{"n": 4}]), + ("p.name CONTAINS 'a' AND p.name CONTAINS 'l'", [{"n": 8}]), + ("p.name CONTAINS 'al'", [{"n": 4}]), + # Numeric merges are representable and must still push. + ("p.age > 20 AND p.age = 30", [{"n": 4}]), + ], +) +def test_connected_join_string_predicate_merge_matches_cypher(predicate: str, expected: Any) -> None: + query = f"MATCH (p)-[]->(a), (p)-[]->(b) WHERE {predicate} RETURN count(p) AS n" + + result = _real_string_merge_graph().gfql(query) + assert result._nodes.to_dict(orient="records") == expected + + +def _real_pandas_bool_graph() -> Plottable: + nodes = pd.DataFrame({"id": ["a", "b", "c", "d", "e"], "flag": [True, False, True, False, True], "age": [1, 2, 3, 4, 5]}) + edges = pd.DataFrame({"s": ["a", "b", "c", "a", "b"], "d": ["b", "c", "d", "c", "e"]}) + return graphistry.nodes(nodes, "id").edges(edges, "s", "d") + + +@pytest.mark.parametrize( + "predicate,expected", + [ + # The gate reads the source frame but the executor filters a joined one, where an + # unmatched row introduces NaN and pandas widens bool -> object: numeric to the gate, + # string to the validator. Declining keeps the residual's answer. + ("p.flag > 0", [{"n": 2}]), + ("p.flag >= 1", [{"n": 2}]), + ("p.flag <> 0", [{"n": 2}]), + ("p.flag = 1", [{"n": 2}]), + ("p.flag = true", [{"n": 2}]), + # int64 widens to float64, which stays numeric, so this must still push and answer. + ("p.age >= 2", [{"n": 4}]), + ], +) +def test_connected_join_bool_column_survives_join_widening(predicate: str, expected: Any) -> None: + query = f"MATCH (i)-->(p), (p)-->(c) WHERE {predicate} RETURN count(p) AS n" + + result = _real_pandas_bool_graph().gfql(query) + assert result._nodes.to_dict(orient="records") == expected + + +def test_connected_join_bool_dtype_never_pushes() -> None: + # bool cannot hold NaN, so the join widens it out of its class. int64 -> float64 stays + # numeric and must keep pushing. + assert _connected_join_dtype_admits(">", 0, pd.Series([True]).dtype) is False + assert _connected_join_dtype_admits("==", True, pd.Series([True]).dtype) is False + assert _connected_join_dtype_admits(">", 0, pd.Series([1]).dtype) is True + + +def _real_object_content_graph() -> Plottable: + nodes = pd.DataFrame({ + "id": ["n1", "n2", "n3", "n4"], + "flag": [True, False, None, True], # bool + null -> object, values are not strings + "name": ["ann", "bob", "cid", "dee"], + "mix": ["a", 1, None, "c"], + "num": [1, 2, 3, 4], + }) + edges = pd.DataFrame({"s": ["n1", "n2", "n3", "n1"], "d": ["n2", "n3", "n4", "n3"]}) + return graphistry.nodes(nodes, "id").edges(edges, "s", "d") + + +@pytest.mark.parametrize( + "column", + [ + # `.str` admits only {string, empty, bytes, mixed, mixed-integer}; everything else must + # be omitted. A float column with a None infers as mixed-integer-float -- unlike np.nan, + # which infers as floating -- so enumerating what to reject misses it. + "score", + "clock", + "flag", + "plain_float", + "plain_int", + ], +) +def test_connected_join_string_op_on_non_str_kinds_stays_typed(column: str) -> None: + query = f"MATCH (i)-->(p), (p)-->(c) WHERE p.{column} CONTAINS 'a' RETURN count(p) AS n" + + with pytest.raises(GFQLValidationError): + _real_infer_kind_graph().gfql(query) + + +def _real_edge_alias_graph() -> Plottable: + nodes = pd.DataFrame({ + "id": ["n1", "n2", "n3", "n4"], + "s_col": pd.Series(["a", "b", "c", "d"], dtype=object), + "i_col": pd.Series([1, 2, 3, 4], dtype="int64"), + }) + edges = pd.DataFrame({ + "s": ["n1", "n2", "n3"], "d": ["n2", "n3", "n4"], "w": pd.Series([1, 2, 3], dtype="int64"), + }) + return graphistry.nodes(nodes, "id").edges(edges, "s", "d") + + +@pytest.mark.parametrize( + "predicate,ret", + [ + # A pushed filter that empties the pattern must still run post_join_chain, so the + # aggregate RETURN column survives -- including for edge aliases (#27), which have no + # `.id` fallback and raised until the rows op emitted the full binding schema at 0 rows. + ("p.i_col = 999", "count(e1) AS n"), + ("p.s_col = 'zzz'", "count(e1) AS n"), + ("p.s_col = 'zzz' AND e1.w > 0", "count(p) AS n"), + ("p.s_col = 'zzz'", "count(p) AS n"), + ], +) +def test_connected_join_empty_pattern_keeps_edge_alias_aggregate(predicate: str, ret: str) -> None: + query = f"MATCH (p)-[e1]->(q), (p)-[e2]->(r) WHERE {predicate} RETURN {ret}" + result = _real_edge_alias_graph().gfql(query)._nodes + assert list(result.columns) == ["n"] + assert len(result) == 0 + + +@pytest.mark.parametrize( + "ret,dtype", + [ + # The 0-row schema fill must carry the source dtype: an untyped None gives object, + # which upcasts sum/avg and escapes via UNION ALL. Edge property must stay int64/float. + ("sum(e1.w) AS c", "int64"), + ("avg(e1.w) AS c", "float64"), + ("max(e1.w) AS c", "int64"), + ], +) +def test_connected_join_empty_edge_aggregate_keeps_numeric_dtype(ret: str, dtype: str) -> None: + query = f"MATCH (a)-[e1]->(b), (b)-[e2]->(c) WHERE a.i_col > 99999 RETURN {ret}" + result = _real_edge_alias_graph().gfql(query)._nodes + assert len(result) == 0 + assert str(result["c"].dtype) == dtype + + +@pytest.mark.parametrize( + "ret,dtype", + [ + # A downstream (non-anchor) node reaches the row via a hop whose NaN widens its + # integer columns to float in the non-empty run; the 0-row path must match, or an + # emptied sum(b.iv) returns int64 and escapes via UNION ALL (#31). The anchor never + # NaN-widens, so it stays int; float columns are unchanged. + ("sum(b.iv) AS c", "float64"), + ("max(b.iv) AS c", "float64"), + ("sum(a.iv) AS c", "int64"), + ("sum(b.fv) AS c", "float64"), + ], +) +def test_connected_join_empty_node_aggregate_matches_nonempty_dtype(ret: str, dtype: str) -> None: + nodes = pd.DataFrame({ + "id": ["n1", "n2", "n3", "n4"], + "iv": pd.Series([1, 2, 3, 4], dtype="int64"), + "fv": pd.Series([1.0, 2.0, 3.0, 4.0], dtype="float64"), + }) + edges = pd.DataFrame({"s": ["n1", "n2", "n3"], "d": ["n2", "n3", "n4"]}) + g = graphistry.nodes(nodes, "id").edges(edges, "s", "d") + query = f"MATCH (a)-[e1]->(b), (b)-[e2]->(c) WHERE a.iv = 999 RETURN {ret}" + result = g.gfql(query)._nodes + assert len(result) == 0 + assert str(result["c"].dtype) == dtype + + +@pytest.mark.parametrize( + "ret,dtype", + [ + # A nullable *extension* Int64 on a downstream node stays Int64 through the non-empty + # hop's left-join (avg -> Float64); the 0-row widening must NOT collapse it to numpy + # float64. `pd.api.types.is_integer_dtype(Int64)` is True, so a naive int check + # over-reaches and diverges from both the non-empty run and master -- observable via + # UNION ALL, the same escape as #31. This is the regression Wave 37 caught. + ("sum(b.iv) AS c", "Int64"), + ("max(b.iv) AS c", "Int64"), + ("min(b.iv) AS c", "Int64"), + ("avg(b.iv) AS c", "Float64"), + ], +) +def test_connected_join_empty_node_aggregate_keeps_nullable_int(ret: str, dtype: str) -> None: + nodes = pd.DataFrame({ + "id": ["n1", "n2", "n3", "n4"], + "iv": pd.Series([1, 2, None, 4], dtype="Int64"), + }) + edges = pd.DataFrame({"s": ["n1", "n2", "n3"], "d": ["n2", "n3", "n4"]}) + g = graphistry.nodes(nodes, "id").edges(edges, "s", "d") + query = f"MATCH (a)-[e1]->(b), (b)-[e2]->(c) WHERE a.iv = 999 RETURN {ret}" + empty = g.gfql(query)._nodes + assert len(empty) == 0 + assert str(empty["c"].dtype) == dtype + # The empty schema must match the non-empty run of the identical query. + full_query = f"MATCH (a)-[e1]->(b), (b)-[e2]->(c) RETURN {ret}" + full = g.gfql(full_query)._nodes + assert str(full["c"].dtype) == dtype + + +@pytest.mark.parametrize( + "bv_dtype,expected", + [ + # numpy bool cannot hold the hop's left-join NaN, so the non-empty path widens a + # downstream node's bool column to object; the 0-row path must match, or an emptied + # max(b.bv) returns raw bool where non-empty and master give object (Wave 37 Finding 2). + ("bool", "object"), + # Nullable `boolean` holds NA natively and stays `boolean` in the non-empty path, so it + # must be left untouched -- widening it would be the Finding-1 over-reach in bool form. + ("boolean", "boolean"), + ], +) +def test_connected_join_empty_node_bool_aggregate_matches_nonempty(bv_dtype: str, expected: str) -> None: + nodes = pd.DataFrame({ + "id": ["n1", "n2", "n3", "n4"], + "bv": pd.Series([True, False, True, False], dtype=bv_dtype), + }) + edges = pd.DataFrame({"s": ["n1", "n2", "n3"], "d": ["n2", "n3", "n4"]}) + g = graphistry.nodes(nodes, "id").edges(edges, "s", "d") + empty = g.gfql("MATCH (a)-[e1]->(b), (b)-[e2]->(c) WHERE a.id = 'zzz' RETURN max(b.bv) AS c")._nodes + assert len(empty) == 0 + assert str(empty["c"].dtype) == expected + full = g.gfql("MATCH (a)-[e1]->(b), (b)-[e2]->(c) RETURN max(b.bv) AS c")._nodes + assert str(full["c"].dtype) == expected + + +def test_connected_join_non_empty_edge_alias_aggregate_answers() -> None: + query = "MATCH (p)-[e1]->(q), (p)-[e2]->(r) WHERE e1.w > 0 RETURN count(p) AS n" + assert _real_edge_alias_graph().gfql(query)._nodes.to_dict(orient="records") == [{"n": 3}] + + +def test_object_column_holds_non_strings_fails_closed_when_unreadable() -> None: + # An object column whose values cannot be read tells us nothing about whether `.str` would + # reject them, so omit it and let the residual answer -- matching `_read_node_dtypes`, which + # returns {} for a schema it cannot read. Keeping it would push blind. + from graphistry.compute.filter_by_dict import _object_column_holds_non_strings + + class _UnreadableFrame: + def __getitem__(self, key: str) -> Any: + raise RuntimeError("cannot read column") + + object_dtype = pd.Series(["a"], dtype=object).dtype + assert _object_column_holds_non_strings(_UnreadableFrame(), "x", object_dtype) is True + # Non-object dtypes are not this helper's concern, and real strings still push. + assert _object_column_holds_non_strings(pd.DataFrame({"x": [1]}), "x", pd.Series([1]).dtype) is False + strings = pd.DataFrame({"x": pd.Series(["a", "b"], dtype=object)}) + assert _object_column_holds_non_strings(strings, "x", strings["x"].dtype) is False + + +def _real_all_null_object_graph() -> Plottable: + nodes = pd.DataFrame({"id": ["n1", "n2", "n3", "n4"]}) + nodes["note"] = pd.Series([None] * 4, dtype=object) # infers `empty` + edges = pd.DataFrame({"s": ["n1", "n2", "n3", "n1"], "d": ["n2", "n3", "n4", "n3"]}) + return graphistry.nodes(nodes, "id").edges(edges, "s", "d") + + +@pytest.mark.parametrize( + "predicate", + ["p.note CONTAINS 'a'", "p.note STARTS WITH 'a'", "p.note ENDS WITH 'a'", "p.note =~ 'a.*'"], +) +def test_connected_join_empty_object_column_still_answers(predicate: str) -> None: + # `empty` (an all-null object column) is subset-closed and `.str` handles it, so it must keep + # pushing: 3VL says NULL CONTAINS 'a' is NULL, hence no rows. Dropping `empty` from the + # keep-set silently reverts these to master's unsupported error, and nothing else catches it. + query = f"MATCH (i)-->(p), (p)-->(c) WHERE {predicate} RETURN count(p) AS n" + + assert _real_all_null_object_graph().gfql(query)._nodes.to_dict(orient="records") == [] + + +def test_connected_join_mixed_column_stays_typed() -> None: + # `mixed`/`mixed-integer` pass pandas' accessor rule on the SOURCE frame but are not closed + # under subsetting: the executor filters the join's candidates, and dropping the strings + # leaves integers that `.str` rejects. Declining matches master, which cannot render + # CONTAINS at all. + nodes = pd.DataFrame({"id": ["u1", "u2", "u3", "u4", "u5"]}) + nodes["value"] = pd.Series(["alpha", 10, 20, 30, "beta"], dtype=object) # mixed-integer + edges = pd.DataFrame({"s": ["u1", "u2", "u5"], "d": ["u2", "u3", "u2"]}) + g = graphistry.nodes(nodes, "id").edges(edges, "s", "d") + query = "MATCH (i)-->(p), (p)-->(c) WHERE p.value CONTAINS 'a' RETURN count(p) AS n" + + with pytest.raises(GFQLValidationError): + g.gfql(query) + + +def test_connected_join_bytes_column_stays_typed() -> None: + # bytes passes StringMethods._validate but the methods themselves forbid it via + # @forbid_nonstring_types, so it must be omitted despite being in pandas' accessor + # allowlist. Without the carve-out this leaks a raw TypeError. + nodes = pd.DataFrame({"id": ["n1", "n2", "n3", "n4"]}) + nodes["raw"] = pd.Series([b"abc", b"def", None, b"ghi"], dtype=object) + edges = pd.DataFrame({"s": ["n1", "n2", "n3", "n1"], "d": ["n2", "n3", "n4", "n1"]}) + g = graphistry.nodes(nodes, "id").edges(edges, "s", "d") + query = "MATCH (i)-->(p), (p)-->(c) WHERE p.raw CONTAINS 'a' RETURN count(p) AS n" + + with pytest.raises(GFQLValidationError): + g.gfql(query) + + +def _real_infer_kind_graph() -> Plottable: + import datetime + + nodes = pd.DataFrame({"id": ["n1", "n2", "n3", "n4"]}) + nodes["score"] = pd.Series([1.5, 2.5, None, 4.5], dtype=object) # mixed-integer-float + nodes["clock"] = pd.Series( + [datetime.time(1, 0), datetime.time(2, 0), None, datetime.time(3, 0)], dtype=object + ) # time + nodes["flag"] = pd.Series([True, False, None, True], dtype=object) # boolean + nodes["plain_float"] = pd.Series([1.5, 2.5, 3.5, 4.5], dtype=object) # floating + nodes["plain_int"] = pd.Series([1, 2, 3, 4], dtype=object) # integer + nodes["name"] = ["ann", "bob", "cid", "dee"] # string -- must still push + edges = pd.DataFrame({"s": ["n1", "n2", "n3", "n1"], "d": ["n2", "n3", "n4", "n1"]}) + return graphistry.nodes(nodes, "id").edges(edges, "s", "d") + + +def test_connected_join_string_op_on_non_string_object_column_stays_typed() -> None: + # `object` says nothing about contents: strings live there, and so does a bool column that + # acquired a null. Pushing a string op on dtype alone let `.str` fail on the values and leak + # a raw AttributeError where master raised a GFQL error. + query = "MATCH (i)-->(p), (p)-->(c) WHERE p.flag CONTAINS 'a' RETURN count(p) AS n" + + with pytest.raises(GFQLValidationError): + _real_object_content_graph().gfql(query) + + +@pytest.mark.parametrize( + "predicate,expected", + [ + # Real string columns are object too, and must keep pushing -- master cannot render these. + ("p.name CONTAINS 'o'", [{"n": 1}]), + ("p.name STARTS WITH 'a'", []), + + ("p.num >= 2", [{"n": 3}]), + ], +) +def test_connected_join_object_columns_still_answer(predicate: str, expected: Any) -> None: + query = f"MATCH (i)-->(p), (p)-->(c) WHERE {predicate} RETURN count(p) AS n" + + result = _real_object_content_graph().gfql(query) + assert result._nodes.to_dict(orient="records") == expected + + +def test_node_dtypes_for_pushdown_fails_closed_on_unreadable_nodes() -> None: + # An unreadable schema must yield an empty mapping, not None: None means "no graph, use + # value-type rules" and would push dtype-blind. Pinned directly because the two are + # observationally equivalent end-to-end (such frames fail execution either way). + class _Unreadable: + columns = ["id"] + + @property + def dtypes(self) -> Any: + raise RuntimeError("schema unavailable") + + g = graphistry.nodes(pd.DataFrame({"id": ["a"]}), "id").edges( + pd.DataFrame({"s": ["a"], "d": ["a"]}), "s", "d" + ) + object.__setattr__(g, "_nodes", _Unreadable()) + + dtypes = _node_dtypes_for_pushdown(g) + assert dtypes is not None, "unreadable schema must not fall back to dtype-blind pushing" + assert dict(dtypes) == {} + + +def test_node_dtypes_for_pushdown_defers_the_conversion() -> None: + # Reading dtypes costs a full engine conversion, and only connected-join pushdown asks -- + # a path most queries never reach. Computing eagerly charged every string query for a + # frame it discarded. Nothing may convert until a lookup actually happens. + pl = pytest.importorskip("polars") + + converted = [] + nodes = pl.DataFrame({"id": ["a", "b"], "age": pl.Series([1, 2], dtype=pl.Int64)}) + g = graphistry.nodes(nodes, "id").edges(pl.DataFrame({"s": ["a"], "d": ["b"]}), "s", "d") + + import graphistry.compute.filter_by_dict as filter_by_dict + + original = filter_by_dict._read_node_dtypes + + def spy(*args: Any, **kwargs: Any) -> Any: + converted.append(1) + return original(*args, **kwargs) + + filter_by_dict._read_node_dtypes = spy + try: + dtypes = filter_by_dict._node_dtypes_for_pushdown(g) + assert dtypes is not None + assert converted == [] # built, but nothing read yet + assert _connected_join_dtype_classes(dtypes["age"]) == (True, False) + assert converted == [1] # first lookup materialized + _ = dtypes["id"] + assert converted == [1] # and it is cached + finally: + filter_by_dict._read_node_dtypes = original + + +def test_node_dtypes_for_pushdown_matches_the_full_conversion() -> None: + # polars -> pandas is DATA-dependent: an empty probe reports `bool` for a nullable Boolean + # while the real conversion yields `object`. The gate must report what the executor sees, + # so compare against the full conversion rather than asserting probe dtypes in isolation. + pl = pytest.importorskip("polars") + from graphistry.Engine import df_to_engine, resolve_engine + + nodes = pl.DataFrame({ + "id": ["a", "b", "c", "d"], + "flag": pl.Series([True, False, None, True], dtype=pl.Boolean), + "age": pl.Series([1, 2, None, 4], dtype=pl.Int64), + "name": pl.Series(["w", "x", "y", "z"]), + }) + edges = pl.DataFrame({"s": ["a", "a"], "d": ["b", "c"]}) + g = graphistry.nodes(nodes, "id").edges(edges, "s", "d") + + reported = _node_dtypes_for_pushdown(g) + assert reported is not None + executed = df_to_engine(nodes, resolve_engine("auto", nodes), warn=False) + # The map is deliberately partial: object columns whose values `.str` would reject are + # omitted so the gate fails closed on them. Every column it DOES report must agree with + # the executed frame. + for column, dtype in zip(list(executed.columns), list(executed.dtypes)): + if str(column) not in reported: + continue + assert _connected_join_dtype_classes(reported[str(column)]) == _connected_join_dtype_classes(dtype) + # `flag` is polars Boolean with a null -> pandas object holding bools -> omitted. + assert "flag" not in reported + assert {"id", "age", "name"} <= set(reported) + # The empty probe would have called the nullable bool numeric, which is why it cannot be + # used: the real conversion yields object-holding-bools, which is omitted instead. + empty = df_to_engine(nodes.head(0), resolve_engine("auto", nodes), warn=False) + empty_dtypes = dict(zip([str(name) for name in empty.columns], list(empty.dtypes))) + assert _connected_join_dtype_classes(empty_dtypes["flag"]) == (True, False) + + +@pytest.mark.parametrize( + "predicate,expected", + [("p.flag > 0", [{"n": 4}]), ("p.flag = true", [{"n": 4}]), ("p.age >= 2", [{"n": 4}])], +) +def test_connected_join_polars_nullable_columns_match_master(predicate: str, expected: Any) -> None: + pl = pytest.importorskip("polars") + + nodes = pl.DataFrame({ + "id": ["a", "b", "c", "d"], + "flag": pl.Series([True, False, None, True], dtype=pl.Boolean), + "age": pl.Series([1, 2, None, 4], dtype=pl.Int64), + }) + edges = pl.DataFrame({"s": ["a", "a", "b", "b"], "d": ["b", "c", "a", "c"]}) + g = graphistry.nodes(nodes, "id").edges(edges, "s", "d") + query = f"MATCH (p)-[]->(x), (p)-[]->(y) WHERE {predicate} RETURN count(p) AS n" + + out = g.gfql(query)._nodes + records = out.to_dict(orient="records") if hasattr(out, "to_dict") else out.to_dicts() + assert records == expected + + +def test_node_dtypes_for_pushdown_reads_frames_without_head() -> None: + # pyarrow.Table has no .head(); the probe raised, was swallowed, and returned None -- + # which means "no graph, use value-type rules" and pushes dtype-blind, the exact bug the + # gate exists to stop. Absent graph and unreadable schema must not be conflated. + pa = pytest.importorskip("pyarrow") + + nodes = pa.table({"id": ["a", "b", "c"], "name": ["ann", "bob", "cid"], "age": [30, 40, 50]}) + edges = pa.table({"s": ["a", "a", "b", "b"], "d": ["b", "c", "a", "c"]}) + g = graphistry.nodes(nodes, "id").edges(edges, "s", "d") + + dtypes = _node_dtypes_for_pushdown(g) + assert dtypes is not None + assert _connected_join_dtype_classes(dtypes["name"]) == (False, True) + assert _connected_join_dtype_classes(dtypes["age"]) == (True, False) + + +def test_arrow_table_nodes_match_master_results() -> None: + pa = pytest.importorskip("pyarrow") + + nodes = pa.table({"id": ["a", "b", "c"], "name": ["ann", "bob", "cid"], "age": [30, 40, 50]}) + edges = pa.table({"s": ["a", "a", "b", "b"], "d": ["b", "c", "a", "c"]}) + g = graphistry.nodes(nodes, "id").edges(edges, "s", "d") + + def run(predicate: str) -> Any: + query = f"MATCH (p)-[]->(x), (p)-[]->(y) WHERE {predicate} RETURN count(p) AS n" + out = g.gfql(query)._nodes + return out.to_dict(orient="records") if hasattr(out, "to_dict") else out.to_dicts() + + assert run("p.name = 5") == [] + assert run("p.name = 'ann'") == [{"n": 4}] + assert run("p.age >= 40") == [{"n": 4}] + + +def test_connected_join_dtype_classes_defers_to_the_live_validator() -> None: + # The gate must not second-guess filter_by_dict: it validates whatever + # `_node_dtypes_for_pushdown` classified, so any disagreement turns a correct answer into + # a type error. Asking its own helpers makes them agree by construction. + from graphistry.compute.filter_by_dict import _is_numeric_dtype_safe, _is_string_dtype_safe + import pyarrow as pa + + for dtype in [ + pd.Series([1]).dtype, + pd.Series(["a"]).dtype, + pd.Series([True]).dtype, + pd.ArrowDtype(pa.bool_()), + pd.ArrowDtype(pa.dictionary(pa.int32(), pa.int64())), + pd.CategoricalDtype(["a"]), + pd.IntervalDtype("int64"), + ]: + assert _connected_join_dtype_classes(dtype) == ( + bool(_is_numeric_dtype_safe(dtype)), + bool(_is_string_dtype_safe(dtype)), + ) + + +@pytest.mark.parametrize( + "predicate,expected", + [ + # Decimal materializes to pandas object, so ordering must not push; the residual answers. + ("p.age > 25", [{"n": 8}]), + ("p.age >= 26", [{"n": 8}]), + ("p.age != 26", [{"n": 8}]), + ("p.age = 26", [{"n": 4}]), + # A plain polars numeric column materializes to int64 and must still push and answer. + ("p.n2 >= 26", [{"n": 8}]), + ], +) +def test_connected_join_polars_decimal_matches_master(predicate: str, expected: Any) -> None: + pl = pytest.importorskip("polars") + + nodes = pl.DataFrame({ + "id": ["p1", "p2", "p3", "x1", "y1"], + "age": pl.Series([25, 26, 30, 1, 2], dtype=pl.Int64()).cast(pl.Decimal(10, 2)), + "n2": pl.Series([25, 26, 30, 1, 2], dtype=pl.Int64()), + }) + edges = pl.DataFrame({ + "s": ["p1", "p1", "p2", "p2", "p3", "p3"], + "d": ["x1", "y1", "x1", "y1", "x1", "y1"], + }) + g = graphistry.nodes(nodes, "id").edges(edges, "s", "d") + query = f"MATCH (p)-->(a), (p)-->(b) WHERE {predicate} RETURN count(p) AS n" + + result = g.gfql(query)._nodes + records = result.to_dict(orient="records") if hasattr(result, "to_dict") else result.to_dicts() + assert records == expected + + +def test_connected_join_polars_backed_graph_matches_pandas_results() -> None: + # The whole dtype gate was inert on polars-typed nodes: `p.age = 'foo'` pushed and + # raised where a pandas-backed graph correctly answers []. + pl = pytest.importorskip("polars") + + nodes = pd.DataFrame({"id": ["p1", "p2", "p3", "x1", "y1"], "age": [25, 26, 30, 1, 2]}) + edges = pd.DataFrame({"s": ["p1", "p1", "p2", "p2", "p3", "p3"], "d": ["x1", "y1", "x1", "y1", "x1", "y1"]}) + query = "MATCH (p)-->(a), (p)-->(b) WHERE p.age = 'foo' RETURN count(p) AS n" + + g_polars = graphistry.nodes(pl.from_pandas(nodes), "id").edges(pl.from_pandas(edges), "s", "d") + result = g_polars.gfql(query)._nodes + records = result.to_dict(orient="records") if hasattr(result, "to_dict") else result.to_dicts() + assert records == [] + + +def test_node_dtypes_for_pushdown_reads_columns_not_items() -> None: + # pandas/cuDF expose a column-indexed Series but polars exposes a plain list, so the + # mapping must be built by zipping columns with dtypes rather than calling `.items()`. + g = _real_bool_shape_graph() + dtypes = _node_dtypes_for_pushdown(g) + assert dtypes is not None + assert set(dtypes) == {"id", "flag", "age"} + assert _connected_join_dtype_admits("==", "yes", dtypes["flag"]) is False + assert _connected_join_dtype_admits("==", "yes", dtypes["id"]) is True + + +def test_connected_join_dtype_schema_selects_pushdown() -> None: + # With dtypes the planner pushes what the column admits and leaves the rest residual. + g = _real_t1_nullable_graph() + dtypes = _node_dtypes_for_pushdown(g) + assert dtypes is not None + + def pushed_props(predicate: str) -> set: + compiled = cast(CompiledCypherQuery, compile_cypher_query(parse_cypher(_t1_nullable_query(predicate)), node_dtypes=dtypes)) + plan = _compiled_execution_extras(compiled).connected_match_join + assert plan is not None + return { + prop + for chain in plan.pattern_chains + for op in chain.chain + if isinstance(op, ASTNode) + for prop in (op.filter_dict or {}) + } - {"node_type"} + + assert pushed_props("p.age >= 26") == {"age"} + assert pushed_props("p.nick = 'a'") == {"nick"} + assert pushed_props("p.nick STARTS WITH 'a'") == {"nick"} + assert pushed_props("p.nick >= 26") == set() + assert pushed_props("p.age = 'foo'") == set() + + +@pytest.mark.parametrize("node_id_column", ["id", "nid"]) +def test_connected_join_count_distinct_alias_is_node_identity(node_id_column: str) -> None: + # A bare alias must lower to its identity column. Left unrewritten it only resolves when + # the node id column happens to be named `id`/`p`, and otherwise degrades to a constant, + # collapsing count(DISTINCT p) to 1. + nodes = pd.DataFrame({ + node_id_column: [0, 1, 2, 10, 11, 20, 21], + "node_type": ["Person"] * 3 + ["City"] * 2 + ["Interest"] * 2, + "age": [25, 27, 35, None, None, None, None], + "interest": [None] * 5 + ["Fine Dining", "photography"], + }) + edges = pd.DataFrame({ + "s": [0, 1, 2, 0, 1, 2], + "d": [20, 20, 21, 10, 10, 11], + "rel": ["HAS_INTEREST"] * 3 + ["LIVES_IN"] * 3, + }) + g = graphistry.nodes(nodes, node_id_column).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(DISTINCT p) AS n" + ) + + assert g.gfql(query)._nodes.to_dict(orient="records") == [{"n": 3}] + + +@pytest.mark.parametrize("predicate", ["p.age >= - 26", "p.age >= -(26)", "p.age >= +(26)"]) +def test_t1_connected_comma_pushes_unfolded_unary_literal(predicate: str) -> None: + # `NUMBER` carries its sign as a lexer terminal, so only a lexically adjacent sign + # folds into the literal. A spaced or parenthesised sign stays a UnaryOp node and + # must still push down rather than degrade to a row residual. + query = ( + "MATCH (p {node_type:'Person'})-[{rel:'HAS_INTEREST'}]->(i {node_type:'Interest'}), " + "(p)-[{rel:'LIVES_IN'}]->(c {node_type:'City'}) " + f"WHERE {predicate} " + "RETURN count(p) AS numPersons" + ) + + expected = 3 if predicate != "p.age >= +(26)" else 2 + result = _mk_graph_benchmark_t1_shape_graph().gfql(query) + assert result._nodes.to_dict(orient="records") == [{"numPersons": expected}] + + filters_by_alias = _compiled_connected_join_filters(query) + assert any("age" in entry.get("p", {}) for entry in filters_by_alias) + + +def test_t1_connected_comma_q5_retains_lower_residual_and_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') " + "AND toLower(p.gender) = toLower('male') " + "AND c.city = 'London' AND c.country = 'United Kingdom' " + "RETURN count(p) AS numPersons" + ) + + plan = _compiled_connected_join_plan(query) + assert "where_rows" in _post_join_functions(query) + assert plan.pattern_attach_prop_aliases == (("i", "p"), ("p",)) + + +def test_t1_connected_comma_grouped_projection_attaches_only_city_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') " + "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" + ) + + plan = _compiled_connected_join_plan(query) + assert "where_rows" in _post_join_functions(query) + assert plan.pattern_attach_prop_aliases == (("i",), ("c",)) + + +def test_t1_connected_comma_pushes_q5_literal_filters_and_retains_lower_residual() -> 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') " + "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) + assert result._nodes.to_dict(orient="records") == [{"numPersons": 1}] + + plan = _compiled_connected_join_plan(query) + assert "where_rows" in _post_join_functions(query) + assert plan.pattern_attach_prop_aliases == (("i", "p"), ("p",)) + + filters_by_alias = _compiled_connected_join_filters(query) + assert not any("interest" in entry.get("i", {}) for entry in filters_by_alias) + assert not any("gender" in entry.get("p", {}) for entry in filters_by_alias) + assert any(entry.get("c", {}).get("city") == "London" for entry in filters_by_alias) + assert any(entry.get("c", {}).get("country") == "United Kingdom" for entry in filters_by_alias) + + +def test_t1_connected_comma_pushes_reversed_single_alias_filters_before_join() -> None: + query = ( + "MATCH (p {node_type:'Person'})-[{rel:'HAS_INTEREST'}]->(i {node_type:'Interest'}), " + "(p)-[{rel:'LIVES_IN'}]->(c {node_type:'City'}) " + "WHERE toLower('fine dining') = toLower(i.interest) " + "AND 23 <= p.age AND 30 >= p.age " + "AND 'London' = c.city " + "RETURN count(p) AS numPersons" + ) + + result = _mk_graph_benchmark_t1_shape_graph().gfql(query) + assert result._nodes.to_dict(orient="records") == [{"numPersons": 2}] + + plan = _compiled_connected_join_plan(query) + assert "where_rows" in _post_join_functions(query) + assert plan.pattern_attach_prop_aliases == (("i",), ()) + + filters_by_alias = _compiled_connected_join_filters(query) + assert not any("interest" in entry.get("i", {}) for entry in filters_by_alias) + assert any("age" in entry.get("p", {}) for entry in filters_by_alias) + assert any(entry.get("c", {}).get("city") == "London" for entry in filters_by_alias) + + +def test_t1_connected_comma_retains_lower_property_plain_lowercase_literal() -> 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) = 'fine dining' " + "AND p.age >= 23 AND p.age <= 30 AND c.city = 'London' " + "RETURN count(p) AS numPersons" + ) + + result = _mk_graph_benchmark_t1_shape_graph().gfql(query) + assert result._nodes.to_dict(orient="records") == [{"numPersons": 2}] + + plan = _compiled_connected_join_plan(query) + assert "where_rows" in _post_join_functions(query) + assert plan.pattern_attach_prop_aliases == (("i",), ()) + + +def test_t1_connected_comma_retains_reversed_uppercase_plain_literal_residual() -> None: + query = ( + "MATCH (p {node_type:'Person'})-[{rel:'HAS_INTEREST'}]->(i {node_type:'Interest'}), " + "(p)-[{rel:'LIVES_IN'}]->(c {node_type:'City'}) " + "WHERE 'FINE DINING' = toLower(i.interest) " + "AND p.age >= 23 AND p.age <= 30 AND c.city = 'London' " + "RETURN count(p) AS numPersons" + ) + + result = _mk_graph_benchmark_t1_shape_graph().gfql(query) + assert result._nodes.to_dict(orient="records") == [] + + plan = _compiled_connected_join_plan(query) + assert "where_rows" in _post_join_functions(query) + assert plan.pattern_attach_prop_aliases == (("i",), ()) + + +@pytest.mark.parametrize( + "where_expr", + [ + "toLower(i.interest) = toLower('i')", + "toLower('i') = toLower(i.interest)", + ], +) +def test_t1_connected_comma_retains_unicode_lower_equality_residual(where_expr: str) -> None: + values = ["İ", "i", "I", "ı"] + person_ids = list(range(4)) + interest_ids = list(range(10, 14)) + city_id = 20 + nodes = pd.DataFrame({ + "id": person_ids + interest_ids + [city_id], + "node_type": ["Person"] * 4 + ["Interest"] * 4 + ["City"], + "interest": [None] * 4 + values + [None], + "city": [None] * 8 + ["London"], + }) + edges = pd.DataFrame({ + "s": person_ids + person_ids, + "d": interest_ids + [city_id] * 4, + "rel": ["HAS_INTEREST"] * 4 + ["LIVES_IN"] * 4, + }) + query = ( + "MATCH (p {node_type:'Person'})-[{rel:'HAS_INTEREST'}]->(i {node_type:'Interest'}), " + "(p)-[{rel:'LIVES_IN'}]->(c {node_type:'City'}) " + f"WHERE {where_expr} AND c.city = 'London' " + "RETURN count(p) AS n" + ) + + graph = _mk_graph(nodes, edges) + oracle_query = ( + "MATCH (i {node_type:'Interest'}) " + f"WHERE {where_expr} " + "RETURN count(i) AS n" + ) + oracle_compiled = cast(CompiledCypherQuery, compile_cypher(oracle_query)) + assert oracle_compiled.execution_extras.connected_match_join is None + oracle_rows = graph.gfql(oracle_query)._nodes.to_dict(orient="records") + assert oracle_rows[0]["n"] != len(values) + + result = graph.gfql(query) + assert result._nodes.to_dict(orient="records") == oracle_rows + + plan = _compiled_connected_join_plan(query) + assert "where_rows" in _post_join_functions(query) + assert plan.pattern_attach_prop_aliases == (("i",), ()) + + +def test_t1_connected_comma_pushes_q7_range_filters_before_join() -> 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('photography') " + "AND p.age >= 23 AND p.age <= 30 " + "AND c.country = 'France' " + "RETURN count(p) AS numPersons, c.state AS state, c.country AS country" + ) + + result = _mk_graph_benchmark_t1_shape_graph().gfql(query) + assert result._nodes.to_dict(orient="records") == [] + + plan = _compiled_connected_join_plan(query) + assert "where_rows" in _post_join_functions(query) + assert plan.pattern_attach_prop_aliases == (("i",), ("c",)) + + filters_by_alias = _compiled_connected_join_filters(query) + assert any("age" in entry.get("p", {}) for entry in filters_by_alias) + assert not any("interest" in entry.get("i", {}) for entry in filters_by_alias) + assert any(entry.get("c", {}).get("country") == "France" for entry in filters_by_alias) + + + def test_issue_1413_ic3_entity_membership_positive_same_city_friend_only() -> None: graph = _mk_ic3_cross_country_shape_graph() result = graph.gfql(