diff --git a/bin/ci_cypher_surface_guard_baseline.json b/bin/ci_cypher_surface_guard_baseline.json index 75ad90165a..bbf8dbb481 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": 8478 + "lowering_py_max_lines": 8511 } diff --git a/graphistry/compute/ast.py b/graphistry/compute/ast.py index 1ca8ef3ae6..4eb0dd4281 100644 --- a/graphistry/compute/ast.py +++ b/graphistry/compute/ast.py @@ -18,6 +18,7 @@ from graphistry.compute.gfql.identifiers import validate_column_references from graphistry.util import setup_logger from graphistry.utils.json import JSONVal, is_json_serializable +from graphistry.compute.gfql.row.prefilter import AliasPrefilters from .predicates.ASTPredicate import ASTPredicate from .predicates.from_json import from_json as predicates_from_json @@ -1573,7 +1574,8 @@ def rows( table: str = "nodes", source: Optional[str] = None, alias_endpoints: Optional[Dict[str, str]] = None, - binding_ops: Optional[List[Dict[str, Any]]] = None, + binding_ops: Optional[List[Dict[str, JSONVal]]] = None, + alias_prefilters: Optional[AliasPrefilters] = None, ) -> ASTCall: """Create a row-source operation for GFQL row pipelines. @@ -1597,18 +1599,20 @@ def rows( params["alias_endpoints"] = alias_endpoints if binding_ops is not None: params["binding_ops"] = binding_ops + if alias_prefilters: + params["alias_prefilters"] = alias_prefilters return ASTCall("rows", params) -def serialize_binding_ops(ops: Sequence[ASTObject]) -> List[Dict[str, Any]]: +def serialize_binding_ops(ops: Sequence[ASTObject]) -> List[Dict[str, JSONVal]]: """Serialize node/edge bindings for ``rows(binding_ops=...)``.""" - binding_ops: List[Dict[str, Any]] = [] + binding_ops: List[Dict[str, JSONVal]] = [] for op in ops: if not isinstance(op, (ASTNode, ASTEdge)): raise ValueError( "Connected bindings-row lowering expects only ASTNode/ASTEdge ops" ) - binding_ops.append(cast(Dict[str, Any], op.to_json(validate=False))) + binding_ops.append(op.to_json(validate=False)) return binding_ops diff --git a/graphistry/compute/gfql/call/validation.py b/graphistry/compute/gfql/call/validation.py index 095a9d74da..63dcd462bf 100644 --- a/graphistry/compute/gfql/call/validation.py +++ b/graphistry/compute/gfql/call/validation.py @@ -52,6 +52,7 @@ ring_categorical_axis_payload_error, ring_continuous_axis_payload_error, ) +from graphistry.compute.gfql.row.prefilter import is_alias_prefilters from graphistry.compute.gfql.row.order_expr import ( is_order_aggregate_alias_ast, order_expr_ast_static_supported, @@ -359,12 +360,13 @@ def _semi_apply_mark_added_node_cols(params: Dict[str, object]) -> Set[str]: SAFELIST_V1: Dict[str, Dict[str, Any]] = { 'rows': _safelist_entry( - {'table', 'source', 'alias_endpoints', 'binding_ops'}, + {'table', 'source', 'alias_endpoints', 'binding_ops', 'alias_prefilters'}, param_validators={ 'table': lambda v: v in ['nodes', 'edges'], 'source': is_string_or_none, 'alias_endpoints': lambda v: isinstance(v, dict), 'binding_ops': is_list_of_dicts, + 'alias_prefilters': is_alias_prefilters, }, description='Set active row table from nodes/edges, optionally filtered by source alias', schema_effects=_schema_effects( diff --git a/graphistry/compute/gfql/cypher/lowering.py b/graphistry/compute/gfql/cypher/lowering.py index 51abd25cce..9d03a88e8e 100644 --- a/graphistry/compute/gfql/cypher/lowering.py +++ b/graphistry/compute/gfql/cypher/lowering.py @@ -8054,6 +8054,35 @@ def _attach_logical_plan_route( ) +def _maybe_pushdown_row_prefilters( + result: CompiledCypherQuery, *, has_graph_context: bool +) -> CompiledCypherQuery: + """L4: peel single-alias WHERE predicates into ``rows(alias_prefilters=...)``. + + Guarded to the simple single-chain shape. Skipped when an alias could be + NULL-filled (OPTIONAL MATCH / reentry / row guards) — pushing a predicate onto + a nullable alias would drop rows the post-join filter keeps — or for + procedure / seed-row / multi-graph programs whose chain semantics this narrow + rewrite has not been vetted against. + """ + if ( + result.procedure_call is not None + or result.seed_rows + or result.optional_null_fill is not None + or result.optional_projection_row_guard is not None + or result.optional_reentry + or result.reentry_plan is not None + or has_graph_context + ): + return result + from .row_pushdown import apply_row_prefilter_pushdown + + new_chain = apply_row_prefilter_pushdown(result.chain) + if new_chain is result.chain: + return result + return replace(result, chain=new_chain) + + def compile_cypher_query( query: Union[CypherQuery, CypherUnionQuery, CypherGraphQuery], *, @@ -8121,6 +8150,7 @@ def compile_cypher_query( _bound_scope_stack: Tuple[ScopeFrame, ...] = () def _attach_graph_context(result: CompiledCypherQuery) -> CompiledCypherQuery: + result = _maybe_pushdown_row_prefilters(result, has_graph_context=bool(compiled_bindings) or _use_ref is not None) result_extras = result.execution_extras or CompiledCypherExecutionExtras() out = result if not compiled_bindings and _use_ref is None: diff --git a/graphistry/compute/gfql/cypher/row_pushdown.py b/graphistry/compute/gfql/cypher/row_pushdown.py new file mode 100644 index 0000000000..cc80d6e4c5 --- /dev/null +++ b/graphistry/compute/gfql/cypher/row_pushdown.py @@ -0,0 +1,300 @@ +"""L4 single-alias predicate pushdown for the Cypher row pipeline. + +A disjunctive / ``searchAny`` WHERE collapses the whole MATCH onto the row +pipeline: ``rows(binding_ops=[a,e,b])`` materializes the full ``(a)-[e]->(b)`` +binding table (O(E) joins) and only THEN filters it with ``where_rows`` / +``search_any``. But a conjunct that references a single alias depends only on +that alias's columns, which the (inner) binding join copies verbatim into the +row table. So we can peel such conjuncts off the post-join filter and hand them +to the ``rows`` op as per-alias ``alias_prefilters``, which the binding builder +applies to each alias frame BEFORE the join — shrinking every downstream hop. + +Parity is exact (see ``_gfql_apply_alias_prefilter``): the SAME evaluator and the +SAME ``search_any`` kernel run, just earlier and on a subset. The pass is +deliberately conservative — anything it does not recognize is left untouched, so +the rewrite can only ever move work earlier, never change results. +""" +from __future__ import annotations + +from collections import defaultdict +from typing import DefaultDict, Dict, List, Optional, Sequence, Set, Tuple + +from graphistry.compute.ast import ASTCall, ASTObject +from graphistry.compute.chain import Chain +from graphistry.compute.exceptions import GFQLValidationError +from graphistry.compute.gfql.row.prefilter import ( + AliasPrefilterSpec, MutableAliasPrefilters, is_alias_prefilters, +) +from graphistry.utils.json import JSONVal +from typing_extensions import TypeGuard + +# Ops that consume/close the post-``rows`` filter region. Reaching any of these +# ends the contiguous filter block we are allowed to peel from. +_FILTER_REGION_OPS = frozenset({"where_rows", "search_any", "semi_apply_mark", "anti_semi_apply"}) + + +def _is_binding_ops(value: object) -> TypeGuard[List[Dict[str, JSONVal]]]: + return isinstance(value, list) and all(isinstance(item, dict) for item in value) + + +def _strip_redundant_parens(expr: str) -> str: + """Drop one or more fully-enclosing ``( ... )`` wrappers. + + The Cypher lowering emits fully-parenthesized left-associated ANDs + (``((((C1) AND (C2)) AND (C3)) AND (C4))``); the quote/bracket-aware AND + splitter only finds top-level ANDs, so the outer wrapper must come off first. + Only strips when the leading ``(`` matches the trailing ``)`` (i.e. the pair + genuinely encloses the whole string). + """ + expr = expr.strip() + while len(expr) >= 2 and expr[0] == "(" and expr[-1] == ")": + depth = 0 + encloses = True + quote: Optional[str] = None + escaped = False + for idx, ch in enumerate(expr): + if quote is not None: + # skip the char after a backslash (escaped quote) — mirror the + # sibling split_top_level_and_conjuncts scanner exactly. + if escaped: + escaped = False + continue + if ch == "\\": + escaped = True + continue + if ch == quote: + quote = None + continue + if ch in {"'", '"', "`"}: + quote = ch + continue + if ch == "(": + depth += 1 + elif ch == ")": + depth -= 1 + if depth == 0 and idx != len(expr) - 1: + encloses = False + break + if not encloses or depth != 0: + break + expr = expr[1:-1].strip() + return expr + + +def _flatten_and_conjuncts(expr: str) -> List[str]: + """Recursively split *expr* into top-level AND conjuncts. + + Returns the fully-flattened conjunct list (each with redundant parens + stripped). A leaf that is not an AND at its level comes back as a single + entry, so ``NOT (x AND y)`` and ``(p OR q)`` stay intact. + """ + from graphistry.compute.gfql.passes.predicate_pushdown import split_top_level_and_conjuncts + + inner = _strip_redundant_parens(expr) + parts = split_top_level_and_conjuncts(inner) + if len(parts) <= 1: + return [inner] + out: List[str] = [] + for part in parts: + out.extend(_flatten_and_conjuncts(part)) + return out + + +def _binding_alias_targets( + binding_ops: Sequence[Dict[str, JSONVal]], +) -> Optional[Tuple[Dict[str, ASTObject], Set[str]]]: + """Deserialize ``binding_ops`` into ``{alias: ASTObject}`` plus the set of + aliases eligible for pushdown. + + Node aliases are always eligible (inner-joined in the connected/cartesian + binding path). Edge aliases are eligible only when the relationship is a + plain single hop — a variable-length / fixed-point edge does not map onto a + single pre-join edge-property filter. Returns ``None`` if any op is + un-parseable (bail out of the whole pushdown for safety). + """ + from graphistry.compute.ast import from_json as ast_from_json, ASTEdge, ASTNode + + alias_targets: Dict[str, ASTObject] = {} + eligible: Set[str] = set() + for op_json in binding_ops: + name = op_json.get("name") if isinstance(op_json, dict) else None + # shortestPath scalar bindings route to a separate builder that does not + # honor alias_prefilters — pushing there would silently drop the filter. + # Bail out of the entire pushdown for any such pattern. + lnh = op_json.get("label_node_hops") if isinstance(op_json, dict) else None + if isinstance(lnh, str) and lnh.startswith("__cypher_shortest_path_hops__"): + return None + try: + obj = ast_from_json(op_json, validate=False) + except (AssertionError, KeyError, TypeError, ValueError, GFQLValidationError): + return None + if not isinstance(name, str): + continue + alias_targets[name] = obj + if isinstance(obj, ASTNode): + eligible.add(name) + elif isinstance(obj, ASTEdge): + hops = op_json.get("hops", 1) + if ( + op_json.get("to_fixed_point") is not True + and hops == 1 + and op_json.get("min_hops") in (None, 1) + and op_json.get("max_hops") in (None, 1) + ): + eligible.add(name) + return alias_targets, eligible + + +def _positive_search_any_markers(steps: Sequence[ASTObject], start_idx: int) -> Set[str]: + """Return searchAny marker columns used as positive top-level AND conjuncts. + + ``search_any`` row ops are positive kernels. They are safe as alias prefilters + only when the retained ``where_rows`` filter also requires that marker to be + true. ``NOT marker`` and ``marker OR ...`` must stay post-join only. + """ + markers: Set[str] = set() + idx = start_idx + while idx < len(steps): + op = steps[idx] + if not isinstance(op, ASTCall) or op.function not in _FILTER_REGION_OPS: + break + if op.function == "where_rows": + expr = op.params.get("expr") + if isinstance(expr, str): + for conjunct in _flatten_and_conjuncts(expr): + leaf = _strip_redundant_parens(conjunct) + if leaf.startswith("__gfql_search_any_") and leaf.endswith("__"): + markers.add(leaf) + idx += 1 + return markers + + +def _conjunct_single_alias( + conjunct: str, + alias_targets: Dict[str, ASTObject], +) -> Optional[str]: + """Return the sole binding alias a conjunct references, else ``None``. + + ``None`` when the conjunct touches zero binding aliases (constants, internal + marker columns), more than one alias, or any aggregate — none of which are + safe to evaluate before the join. + """ + from graphistry.compute.gfql.cypher.lowering import _expr_match_alias_usage + + # Reject any conjunct that names an INTERNAL machinery column. These reserved + # ``__gfql_`` / ``__cypher_`` names are produced by downstream ops (EXISTS / + # searchAny markers ``__gfql_where_pattern_*`` / ``__gfql_search_any_*``) or + # carry a value from another alias (reentry ``__cypher_reentry_*``); the alias + # walker hides their roots, so they falsely read as single-alias and are not + # resolvable on the pre-join alias frame. User columns never use these prefixes. + if "__gfql_" in conjunct or "__cypher_" in conjunct: + return None + + try: + non_aggregate, aggregate = _expr_match_alias_usage( + conjunct, + alias_targets=alias_targets, + params=None, + field="where", + line=0, + column=0, + ) + except GFQLValidationError: + return None + if aggregate: + return None + if len(non_aggregate) != 1: + return None + return next(iter(non_aggregate)) + + +def apply_row_prefilter_pushdown(chain: Chain) -> Chain: + """Attach single-alias predicates to the ``rows(binding_ops=...)`` op as an + advisory ``alias_prefilters`` hint — WITHOUT removing the post-join filters. + + Design note (redundant-filter safety): the peeled predicates are added as a + pre-join HINT only; the original post-join ``where_rows`` / ``search_any`` ops + are left in place untouched. Engines whose row builder honors the hint + (pandas / cuDF via RowPipelineMixin) pre-filter each alias frame before the + binding join and win big; engines that ignore it (e.g. polars' native + ``rows_binding_ops_polars``, the shortestPath scalar builder) simply run the + unchanged post-join filter and stay correct. Because a pushed conjunct is a + literal sub-conjunct of the retained AND, the pre-filter can only remove rows + the post-filter would also remove — never a result change, on any engine. + + Safe no-op for any chain without a ``rows(binding_ops)`` op, or whose filters + are all multi-alias / marker-only. + """ + steps: List[ASTObject] = list(chain.chain) + if not steps: + return chain + + rows_match: Optional[Tuple[int, ASTCall]] = None + for idx, op in enumerate(steps): + if isinstance(op, ASTCall) and op.function == "rows" and op.params.get("binding_ops"): + rows_match = (idx, op) + break + if rows_match is None: + return chain + rows_idx, rows_op = rows_match + raw_binding_ops = rows_op.params.get("binding_ops") + if not _is_binding_ops(raw_binding_ops): + return chain + binding_ops = raw_binding_ops + targets = _binding_alias_targets(binding_ops) + if targets is None: + return chain + alias_targets, eligible = targets + if not eligible: + return chain + + prefilters: DefaultDict[str, List[AliasPrefilterSpec]] = defaultdict(list) + positive_search_markers = _positive_search_any_markers(steps, rows_idx + 1) + + # Scan the contiguous filter region immediately following the rows op and + # COLLECT (never remove) pushable single-alias predicates. + idx = rows_idx + 1 + while idx < len(steps): + op = steps[idx] + if not isinstance(op, ASTCall) or op.function not in _FILTER_REGION_OPS: + break + if op.function == "search_any": + alias = op.params.get("alias") + out_col = op.params.get("out_col") + term = op.params.get("term") + if isinstance(alias, str) and isinstance(term, str) and out_col in positive_search_markers and alias in eligible: + spec: AliasPrefilterSpec = {"kind": "search_any", "term": term} + if op.params.get("case_sensitive"): + spec["case_sensitive"] = True + if op.params.get("regex"): + spec["regex"] = True + columns = op.params.get("columns") + if isinstance(columns, list) and all(isinstance(col, str) for col in columns): + spec["columns"] = columns + prefilters[alias].append(spec) + elif op.function == "where_rows": + expr = op.params.get("expr") + if isinstance(expr, str): + for conjunct in _flatten_and_conjuncts(expr): + alias = _conjunct_single_alias(conjunct, alias_targets) + if alias is not None and alias in eligible: + prefilters[alias].append({"kind": "expr", "text": conjunct}) + idx += 1 + + if not prefilters: + return chain + + # Attach the hint to the rows op; every other op is left byte-for-byte intact. + new_params = dict(rows_op.params) + merged: MutableAliasPrefilters = {} + existing_prefilters = new_params.get("alias_prefilters") + if is_alias_prefilters(existing_prefilters): + for alias, specs in existing_prefilters.items(): + merged[alias] = list(specs) + for alias, specs in prefilters.items(): + merged.setdefault(alias, []).extend(specs) + new_params["alias_prefilters"] = merged + + new_steps = list(steps) + new_steps[rows_idx] = ASTCall("rows", new_params) + return Chain(new_steps, where=chain.where) diff --git a/graphistry/compute/gfql/row/frame_ops.py b/graphistry/compute/gfql/row/frame_ops.py index 7ad3caebc9..28cb283e2f 100644 --- a/graphistry/compute/gfql/row/frame_ops.py +++ b/graphistry/compute/gfql/row/frame_ops.py @@ -5,6 +5,8 @@ import pandas as pd from graphistry.compute.dataframe_utils import df_cons as template_df_cons +from graphistry.compute.gfql.row.prefilter import AliasPrefilters +from graphistry.utils.json import JSONVal if TYPE_CHECKING: from typing import Protocol @@ -21,7 +23,11 @@ class RowPipelineCtx(Protocol): _edges: Any _edge: Any def bind(self) -> "Plottable": ... - def _gfql_binding_ops_row_table(self, binding_ops: Any) -> "Plottable": ... + def _gfql_binding_ops_row_table( + self, + binding_ops: List[Dict[str, JSONVal]], + alias_prefilters: Optional[AliasPrefilters] = None, + ) -> "Plottable": ... def _gfql_bindings_row_table(self, alias_endpoints: Any) -> "Plottable": ... @@ -146,10 +152,11 @@ def rows( table: str = "nodes", source: Optional[str] = None, alias_endpoints: Optional[Dict[str, str]] = None, - binding_ops: Optional[List[Dict[str, Any]]] = None, + binding_ops: Optional[List[Dict[str, JSONVal]]] = None, + alias_prefilters: Optional[AliasPrefilters] = None, ) -> "Plottable": if binding_ops is not None: - return cast("Plottable", ctx._gfql_binding_ops_row_table(binding_ops)) + return ctx._gfql_binding_ops_row_table(binding_ops, alias_prefilters=alias_prefilters) if alias_endpoints is not None: return cast("Plottable", ctx._gfql_bindings_row_table(alias_endpoints)) diff --git a/graphistry/compute/gfql/row/pipeline.py b/graphistry/compute/gfql/row/pipeline.py index 294d068744..892b756d08 100644 --- a/graphistry/compute/gfql/row/pipeline.py +++ b/graphistry/compute/gfql/row/pipeline.py @@ -21,6 +21,9 @@ from graphistry.compute.exceptions import ErrorCode, GFQLValidationError from graphistry.compute.dataframe_utils import concat_frames from graphistry.compute.gfql.row import frame_ops as row_frame_ops +from graphistry.compute.gfql.row.prefilter import AliasPrefilters +from graphistry.compute.typing import DataFrameT +from graphistry.utils.json import JSONVal from graphistry.compute.gfql.row.order_expr import ( extract_temporal_duration_sort_ast, is_order_aggregate_alias_ast, @@ -77,6 +80,7 @@ if TYPE_CHECKING: from graphistry.Plottable import Plottable + from graphistry.compute.ast import ASTObject from graphistry.compute.gfql.expr_parser import ExprNode @@ -3426,7 +3430,101 @@ def _state_key_cols(frame: Any) -> List[str]: merged = merged.drop_duplicates(keep="first") return merged - def _gfql_connected_bindings_state(self, ops: Sequence[Any]) -> Tuple[Any, Dict[str, Any]]: + def _gfql_apply_alias_prefilter( + self, + frame: DataFrameT, + alias: Optional[str], + alias_prefilters: Optional[AliasPrefilters], + *, + is_edge: bool = False, + src_col: Optional[str] = None, + dst_col: Optional[str] = None, + ) -> DataFrameT: + """L4 single-alias predicate pushdown: pre-filter one alias's frame BEFORE the + binding join, using residual predicates the Cypher lowering pass peeled off the + post-join ``where_rows`` / ``search_any``. + + Parity contract: a single-alias predicate's truth on a binding row depends only + on that alias's columns, which the (inner) binding join copies verbatim into the + row table. Filtering the alias frame first therefore removes exactly the rows the + post-join filter would. We reuse the SAME evaluator (``_gfql_eval_string_expr`` + + ``_gfql_bool_mask``) and the SAME ``search_any_mask`` kernel to guarantee + byte-identical semantics; only inner-joined (non-optional) aliases are ever + pushed (guaranteed by the pass). NULL is dropped (Cypher WHERE keeps only TRUE), + matching ``where_rows``. + """ + if not alias or not alias_prefilters or frame is None: + return frame + specs = alias_prefilters.get(alias) + if not specs or len(frame) == 0: + return frame + reserved = set() + if is_edge: + if src_col is not None: + reserved.add(src_col) + if dst_col is not None: + reserved.add(dst_col) + for spec in specs: + kind = spec.get("kind") + if kind == "expr": + # Rename to alias-qualified columns so the shared evaluator resolves + # ``alias.prop`` exactly as it would against the full binding table. + view = frame.rename(columns={c: f"{alias}.{c}" for c in frame.columns}) + value = self._gfql_eval_string_expr(view, spec["text"]) + mask = self._gfql_bool_mask(view, value) + frame = frame.loc[mask] + elif kind == "search_any": + from graphistry.compute.gfql.search_any import search_any_mask + term = spec.get("term") + if not isinstance(term, str): + raise GFQLValidationError( + ErrorCode.E108, + "searchAny pushdown requires a string term", + field="term", + value=term, + language="cypher", + ) + columns = spec.get("columns") + if columns is not None and not all(isinstance(col, str) for col in columns): + raise GFQLValidationError( + ErrorCode.E108, + "searchAny pushdown columns= must be a list of strings", + field="columns", + value=columns, + language="cypher", + ) + # Match the post-join op's searched column set: alias property columns + # only (src/dst are join keys, never ``alias.``-prefixed in bindings). + search_cols = [ + c for c in frame.columns + if isinstance(c, str) and c not in reserved and not c.startswith("__gfql_") + ] + sub = frame[search_cols] + mask = search_any_mask( + sub, + term, + case_sensitive=bool(spec.get("case_sensitive", False)), + regex=bool(spec.get("regex", False)), + columns=columns, + ) + if mask is None: + # columns= referenced an absent column; the pass validates this can't + # happen, but never silently drop rows — bail out of the pushdown. + raise GFQLValidationError( + ErrorCode.E108, + "searchAny pushdown columns= includes a column absent from the alias frame", + field="columns", + value=spec.get("columns"), + language="cypher", + ) + frame = frame.loc[mask] + return frame + + def _gfql_connected_bindings_state( + self, + ops: Sequence["ASTObject"], + alias_prefilters: Optional[AliasPrefilters] = None, + ) -> Tuple[DataFrameT, Dict[str, DataFrameT]]: from graphistry.compute.gfql.same_path.edge_semantics import EdgeSemantics from graphistry.compute.ast import ASTEdge, ASTNode @@ -3455,7 +3553,12 @@ def _gfql_connected_bindings_state(self, ops: Sequence[Any]) -> Tuple[Any, Dict[ base_df = self._nodes if self._nodes is not None else self._edges engine = resolve_engine(EngineAbstract.AUTO, base_df) start_nodes = getattr(self, "_gfql_start_nodes", None) - first_nodes = ops[0].execute( + first_op = ops[0] + if not isinstance(first_op, ASTNode): + self._gfql_bindings_error( + "Cypher multi-alias row bindings currently require node steps in even positions" + ) + first_nodes = first_op.execute( g=base_graph, prev_node_wavefront=start_nodes, target_wave_front=None, @@ -3463,9 +3566,14 @@ def _gfql_connected_bindings_state(self, ops: Sequence[Any]) -> Tuple[Any, Dict[ )._nodes if first_nodes is None or node_id_col not in first_nodes.columns: return self._nodes.iloc[0:0].copy(), {} + # L4 single-alias predicate pushdown: pre-filter the seed alias frame + # BEFORE it becomes the traversal wavefront (shrinks every downstream hop). + first_alias = first_op._name + first_nodes = self._gfql_apply_alias_prefilter( + first_nodes, first_alias, alias_prefilters + ) state_df = first_nodes[[node_id_col]].copy().rename(columns={node_id_col: "__current__"}) - alias_frames: Dict[str, Any] = {} - first_alias = getattr(ops[0], "_name", None) + alias_frames: Dict[str, DataFrameT] = {} if isinstance(first_alias, str): state_df[first_alias] = state_df["__current__"] alias_frames[first_alias] = first_nodes @@ -3493,6 +3601,15 @@ def _gfql_connected_bindings_state(self, ops: Sequence[Any]) -> Tuple[Any, Dict[ self._gfql_bindings_error( "Cypher multi-alias row bindings do not yet support variable-length relationship aliases" ) + # L4 pushdown: pre-filter this hop's edge frame by the edge alias's + # single-alias residual (single-hop only; searchAny/e.prop predicates) + # before orientation + join. src/dst are excluded from searchAny to match + # the post-join binding table (where they are consumed as join keys). + if not sem.is_multihop and isinstance(edge_alias, str) and edges_df_step is not None: + edges_df_step = self._gfql_apply_alias_prefilter( + edges_df_step, edge_alias, alias_prefilters, + is_edge=True, src_col=src_col, dst_col=dst_col, + ) if not sem.is_multihop and (edges_df_step is None or len(edges_df_step) == 0): return state_df.iloc[0:0], alias_frames @@ -3581,8 +3698,13 @@ def _gfql_connected_bindings_state(self, ops: Sequence[Any]) -> Tuple[Any, Dict[ )._nodes if next_nodes is None or node_id_col not in next_nodes.columns: return state_df.iloc[0:0], alias_frames + # L4 pushdown: pre-filter the endpoint alias frame by its single-alias + # residual before the isin-restrict, so only satisfying endpoints survive. + node_alias = next_node_op._name + next_nodes = self._gfql_apply_alias_prefilter( + next_nodes, node_alias, alias_prefilters + ) state_df = state_df[state_df["__current__"].isin(next_nodes[node_id_col])].copy() - node_alias = getattr(next_node_op, "_name", None) if isinstance(node_alias, str): state_df[node_alias] = state_df["__current__"] hop_column = edge_op.label_node_hops @@ -3606,7 +3728,8 @@ def _gfql_connected_bindings_state(self, ops: Sequence[Any]) -> Tuple[Any, Dict[ def _gfql_connected_bindings_row_table( self, - binding_ops: List[Dict[str, Any]], + binding_ops: List[Dict[str, JSONVal]], + alias_prefilters: Optional[AliasPrefilters] = None, ) -> "Plottable": from graphistry.compute.ast import ASTEdge, ASTNode, from_json as ast_from_json @@ -3616,7 +3739,7 @@ def _gfql_connected_bindings_row_table( ops = [ast_from_json(op_json, validate=False) for op_json in binding_ops] if self._gfql_is_shortest_path_scalar_binding_ops(ops): return self._gfql_shortest_path_scalar_bindings_row_table(ops) - return self._gfql_connected_bindings_row_table_from_ops(ops) + return self._gfql_connected_bindings_row_table_from_ops(ops, alias_prefilters=alias_prefilters) @staticmethod def _gfql_is_shortest_path_scalar_binding_ops(ops: Sequence[Any]) -> bool: @@ -3639,11 +3762,12 @@ def _gfql_is_shortest_path_scalar_binding_ops(ops: Sequence[Any]) -> bool: def _gfql_connected_bindings_row_table_from_ops( self, - ops: Sequence[Any], + ops: Sequence["ASTObject"], + alias_prefilters: Optional[AliasPrefilters] = None, ) -> "Plottable": from graphistry.compute.ast import ASTEdge - state_df, alias_frames = self._gfql_connected_bindings_state(ops) + state_df, alias_frames = self._gfql_connected_bindings_state(ops, alias_prefilters=alias_prefilters) bindings = self._gfql_connected_bindings_row_frame_from_state(ops, state_df, alias_frames) out = self._gfql_row_table(bindings) edge_aliases = { @@ -3657,10 +3781,10 @@ def _gfql_connected_bindings_row_table_from_ops( def _gfql_connected_bindings_row_frame_from_state( self, - ops: Sequence[Any], - state_df: Any, - alias_frames: Dict[str, Any], - ) -> Any: + ops: Sequence["ASTObject"], + state_df: DataFrameT, + alias_frames: Dict[str, DataFrameT], + ) -> DataFrameT: from graphistry.compute.ast import ASTNode node_id = self._node @@ -3678,9 +3802,9 @@ def _gfql_connected_bindings_row_frame_from_state( bindings = state_df.copy() node_aliases = [ - getattr(op, "_name", None) + op._name for op in ops - if isinstance(op, ASTNode) and isinstance(getattr(op, "_name", None), str) + if isinstance(op, ASTNode) and isinstance(op._name, str) ] for alias in node_aliases: alias = str(alias) @@ -3869,8 +3993,11 @@ def _gfql_shortest_path_scalar_bindings_row_table( def _gfql_cartesian_node_bindings_row_table( self, - ops: Sequence[Any], + ops: Sequence["ASTObject"], + alias_prefilters: Optional[AliasPrefilters] = None, ) -> "Plottable": + from graphistry.compute.ast import ASTNode + if self._nodes is None: return self._gfql_row_table(self._gfql_empty_frame()) @@ -3892,6 +4019,10 @@ def _gfql_cartesian_node_bindings_row_table( anonymous_cols: List[str] = [] for idx, node_op in enumerate(ops): + if not isinstance(node_op, ASTNode): + self._gfql_bindings_error( + "Cypher disconnected row bindings currently require node steps" + ) matched_nodes = node_op.execute( g=base_graph, prev_node_wavefront=start_nodes, @@ -3901,7 +4032,12 @@ def _gfql_cartesian_node_bindings_row_table( if matched_nodes is None or node_id not in matched_nodes.columns: return self._gfql_row_table(base_nodes.iloc[0:0].copy()) - alias = getattr(node_op, "_name", None) + # L4 pushdown: pre-filter each disconnected-pattern node alias frame. + alias = node_op._name + matched_nodes = self._gfql_apply_alias_prefilter( + matched_nodes, alias, alias_prefilters + ) + if isinstance(alias, str): frame = self._gfql_node_alias_lookup_frame(matched_nodes, str(node_id), alias) else: @@ -3918,15 +4054,16 @@ def _gfql_cartesian_node_bindings_row_table( def _gfql_binding_ops_row_table( self, - binding_ops: List[Dict[str, Any]], + binding_ops: List[Dict[str, JSONVal]], + alias_prefilters: Optional[AliasPrefilters] = None, ) -> "Plottable": from graphistry.compute.ast import from_json as ast_from_json ops = [ast_from_json(op_json, validate=False) for op_json in binding_ops] mode = self._gfql_binding_ops_mode(ops) if mode == "node_cartesian": - return self._gfql_cartesian_node_bindings_row_table(ops) - return self._gfql_connected_bindings_row_table(binding_ops) + return self._gfql_cartesian_node_bindings_row_table(ops, alias_prefilters=alias_prefilters) + return self._gfql_connected_bindings_row_table(binding_ops, alias_prefilters=alias_prefilters) def _gfql_bindings_row_table( self, alias_endpoints: Dict[str, str] diff --git a/graphistry/compute/gfql/row/prefilter.py b/graphistry/compute/gfql/row/prefilter.py new file mode 100644 index 0000000000..9c859e1750 --- /dev/null +++ b/graphistry/compute/gfql/row/prefilter.py @@ -0,0 +1,45 @@ +"""Typed row-pipeline alias prefilter hints.""" +from __future__ import annotations + +from typing import Dict, List, Literal, Mapping, Sequence, TypedDict +from typing_extensions import TypeGuard + + +class AliasPrefilterSpec(TypedDict, total=False): + kind: Literal["expr", "search_any"] + text: str + term: str + case_sensitive: bool + regex: bool + columns: List[str] + + +AliasPrefilters = Mapping[str, Sequence[AliasPrefilterSpec]] +MutableAliasPrefilters = Dict[str, List[AliasPrefilterSpec]] + + +def _is_string_list(value: object) -> bool: + return isinstance(value, list) and all(isinstance(item, str) for item in value) + + +def _is_alias_prefilter_spec(value: object) -> bool: + if not isinstance(value, dict): + return False + kind = value.get("kind") + if kind == "expr": + return isinstance(value.get("text"), str) + if kind == "search_any": + columns = value.get("columns") + return isinstance(value.get("term"), str) and (columns is None or _is_string_list(columns)) + return False + + +def is_alias_prefilters(value: object) -> TypeGuard[MutableAliasPrefilters]: + if not isinstance(value, dict): + return False + for alias, specs in value.items(): + if not isinstance(alias, str) or not isinstance(specs, list): + return False + if not all(_is_alias_prefilter_spec(spec) for spec in specs): + return False + return True 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 385c68f169..64d089086f 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 @@ -43,6 +43,7 @@ "graphistry/compute/gfql/cypher/reentry/scope.py": 78.72, "graphistry/compute/gfql/cypher/reentry_plan.py": 100.0, "graphistry/compute/gfql/cypher/result_postprocess.py": 58.0, + "graphistry/compute/gfql/cypher/row_pushdown.py": 88.97, "graphistry/compute/gfql/cypher/shortest_path_aliases.py": 97.37, "graphistry/compute/gfql/cypher/shortest_path_guards.py": 77.08, "graphistry/compute/gfql/row/__init__.py": 100.0, @@ -53,6 +54,7 @@ "graphistry/compute/gfql/row/order_expr.py": 81.08, "graphistry/compute/gfql/row/ordering.py": 83.92, "graphistry/compute/gfql/row/pipeline.py": 70.3, + "graphistry/compute/gfql/row/prefilter.py": 45.0, "graphistry/compute/gfql/temporal/__init__.py": 100.0, "graphistry/compute/gfql/temporal/constructors.py": 86.06, "graphistry/compute/gfql/temporal/durations.py": 75.47, diff --git a/graphistry/tests/compute/gfql/cypher/test_row_pushdown.py b/graphistry/tests/compute/gfql/cypher/test_row_pushdown.py new file mode 100644 index 0000000000..41095a3e84 --- /dev/null +++ b/graphistry/tests/compute/gfql/cypher/test_row_pushdown.py @@ -0,0 +1,205 @@ +"""Parity + guard tests for L4 single-alias predicate pushdown. + +The pushdown pass peels single-alias WHERE conjuncts (and ``searchAny`` ops) off +the post-join filter into ``rows(alias_prefilters=...)``, which pre-filters each +alias frame BEFORE the binding join. This must be a pure optimization: identical +results to the un-pushed chain, on every engine and graph shape. It must also +refuse to push in the unsafe cases (shortestPath scalar bindings, OPTIONAL / +reentry-carried values, internal marker columns) where the peeled filter would be +lost or mis-evaluated. +""" +from __future__ import annotations + +from typing import List, Tuple + +import numpy as np +import pandas as pd +import pytest + +import graphistry +import graphistry.compute.gfql.cypher.lowering as _lowering +from graphistry.compute.ast import ASTCall +from graphistry.compute.exceptions import GFQLValidationError +from graphistry.compute.gfql.cypher.api import cypher_to_gfql +from graphistry.compute.gfql.cypher.row_pushdown import ( + _flatten_and_conjuncts, + _strip_redundant_parens, + apply_row_prefilter_pushdown, +) + + +def _engines() -> List[str]: + out = ["pandas"] + try: + import cudf # noqa: F401 + out.append("cudf") + except ImportError: + pass + return out + + +ENGINES = _engines() + + +# --------------------------------------------------------------------------- # +# fixtures: directed / self-loop workload with nullable + textual columns +# --------------------------------------------------------------------------- # +@pytest.fixture(scope="module") +def g(): + rng = np.random.default_rng(7) + n = 400 + ndf = pd.DataFrame({ + "id": np.arange(n), + "kind": rng.choice(["internal", "user", "svc", "admin"], n), + "score": np.where(rng.random(n) < 0.15, np.nan, rng.random(n)), + "flag": rng.random(n) < 0.3, + "name": [f"node_{['Ember','Wire','Ash','Flux'][i % 4]}_{i}" for i in range(n)], + }) + m = 1500 + src = rng.integers(0, n, m) + dst = rng.integers(0, n, m) + # inject self-loops + src[:60] = dst[:60] + edf = pd.DataFrame({ + "src": src, + "dst": dst, + "w": rng.integers(0, 8, m), + "etype": rng.choice(["ref", "WIRE_link", "call", "owns"], m), + }) + return graphistry.nodes(ndf, "id").edges(edf, "src", "dst") + + +# queries that MUST trigger a pushdown and must be result-identical pre/post +PUSHDOWN_QUERIES: List[Tuple[str, str]] = [ + ("edge_search", "MATCH (a)-[e]->(b) WHERE searchAny(e, 'WIRE') RETURN e.w AS w ORDER BY w LIMIT 5000"), + ("node_search", "MATCH (n) WHERE searchAny(n, 'Ember') RETURN n.id AS id ORDER BY id LIMIT 5000"), + ( + "panel", + "MATCH (a)-[e]->(b) WHERE a.kind <> 'internal' AND NOT (a.flag = true AND a.score < 0.1) " + "AND (a.score > 0.25 OR a.score IS NULL) AND e.w > 2 " + "RETURN a.id AS id, a.score AS score ORDER BY id LIMIT 5000", + ), + ( + "undirected_edge_filter", + "MATCH (a)-[e]-(b) WHERE e.w > 3 AND (a.score > 0.5 OR a.score IS NULL) " + "RETURN a.id AS id ORDER BY id LIMIT 5000", + ), + ( + "combined", + "MATCH (n) WHERE n.kind <> 'internal' AND (n.score > 0.25 OR n.score IS NULL) " + "AND EXISTS { (n)--() } AND searchAny(n, 'Ash') RETURN n.id AS id ORDER BY id LIMIT 5000", + ), +] + + +def _base_chain(query: str, monkeypatch: pytest.MonkeyPatch): + """Compile WITHOUT the pushdown pass (identity-patched), to get the baseline chain.""" + monkeypatch.setattr(_lowering, "_maybe_pushdown_row_prefilters", lambda rows, **kwargs: rows) + return cypher_to_gfql(query) + + +def _norm(df) -> List[tuple]: + mod = type(df).__module__ + pdf = df.to_pandas() if ("cudf" in mod or "polars" in mod) else df + pdf = pdf.reset_index(drop=True) + rows = [] + for rec in pdf.itertuples(index=False): + rows.append(tuple( + None if (isinstance(v, float) and pd.isna(v)) or v is None + else (round(float(v), 9) if isinstance(v, float) else v) + for v in rec + )) + return sorted(rows, key=repr) + + +@pytest.mark.parametrize("engine", ENGINES) +@pytest.mark.parametrize("name,query", PUSHDOWN_QUERIES, ids=[q[0] for q in PUSHDOWN_QUERIES]) +def test_pushdown_parity(g, engine, name, query, monkeypatch): + pushed = cypher_to_gfql(query) + base = _base_chain(query, monkeypatch) + # sanity: the pass actually changed the plan for these queries + assert any( + isinstance(op, ASTCall) and op.function == "rows" and (op.params.get("alias_prefilters")) + for op in pushed.chain + ), f"{name}: expected a pushdown but none was produced" + r_push = g.gfql(pushed, engine=engine) + r_base = g.gfql(base, engine=engine) + assert _norm(r_push._nodes) == _norm(r_base._nodes), f"{name}: pushdown changed results on {engine}" + + +# --------------------------------------------------------------------------- # +# guard cases: the pass MUST NOT push (would lose / mis-evaluate the filter) +# --------------------------------------------------------------------------- # +def _has_prefilter(chain) -> bool: + return any( + isinstance(op, ASTCall) and op.function == "rows" and op.params.get("alias_prefilters") + for op in chain.chain + ) + + +@pytest.mark.parametrize("query", [ + # EXISTS marker column conjunct — produced downstream, not on the pre-join frame + "MATCH (n) WHERE EXISTS { (n)--() } RETURN n.id AS id", + # multi-alias conjunct — depends on two aliases, cannot pre-filter one frame + "MATCH (a)-[e]->(b) WHERE a.score > b.score OR a.score IS NULL RETURN a.id AS id", +]) +def test_guard_no_unsafe_pushdown(query): + assert not _has_prefilter(cypher_to_gfql(query)) + + +def test_guard_negated_search_any_not_pushed(g, monkeypatch): + q = "MATCH (n) WHERE NOT searchAny(n, 'Ember') RETURN n.id AS id ORDER BY id LIMIT 5000" + pushed = cypher_to_gfql(q) + base = _base_chain(q, monkeypatch) + assert not _has_prefilter(pushed) + for engine in ENGINES: + r_push = g.gfql(pushed, engine=engine) + r_base = g.gfql(base, engine=engine) + assert _norm(r_push._nodes) == _norm(r_base._nodes), f"NOT searchAny pushdown changed results on {engine}" + + +def test_guard_shortest_path_not_pushed(g): + # shortestPath scalar bindings route to a builder that ignores alias_prefilters + q = ( + "MATCH (p1), (p2), p = shortestPath((p1)-[*]-(p2)) " + "WHERE p1.kind <> 'internal' RETURN p1.id AS a, p2.id AS b, p IS NULL AS noPath LIMIT 100" + ) + try: + chain = cypher_to_gfql(q) + except (GFQLValidationError, NotImplementedError): + pytest.skip("shortestPath shape not single-chain in this build") + assert not _has_prefilter(chain) + + +# --------------------------------------------------------------------------- # +# pass internals +# --------------------------------------------------------------------------- # +def test_strip_redundant_parens(): + assert _strip_redundant_parens("((x))") == "x" + assert _strip_redundant_parens("(a) AND (b)") == "(a) AND (b)" # not fully enclosing + assert _strip_redundant_parens("(a AND b)") == "a AND b" + assert _strip_redundant_parens("('a )b')") == "'a )b'" # paren inside string literal + + +def test_flatten_and_conjuncts_nested(): + expr = "((((a.kind <> 'internal') AND (NOT ((a.flag = true) AND (a.score < 0.1)))) " + expr += "AND ((a.score > 0.25) OR (a.score IS NULL))) AND (e.w > 2))" + parts = _flatten_and_conjuncts(expr) + assert parts == [ + "a.kind <> 'internal'", + "NOT ((a.flag = true) AND (a.score < 0.1))", + "(a.score > 0.25) OR (a.score IS NULL)", + "e.w > 2", + ] + + +def test_flatten_single_conjunct_kept_whole(): + assert _flatten_and_conjuncts("(a.score > 0.25) OR (a.score IS NULL)") == [ + "(a.score > 0.25) OR (a.score IS NULL)" + ] + + +def test_pass_is_noop_on_plain_chain(): + from graphistry.compute.chain import Chain + ch = Chain([ASTCall("rows", {"table": "nodes"}), ASTCall("limit", {"value": 10})]) + assert apply_row_prefilter_pushdown(ch) is ch