perf(gfql/cypher): single-alias predicate pushdown in the row pipeline (L4)

bin/ci_cypher_surface_guard_baseline.json

@@ -13,5 +13,5 @@
       "max_properties": 0
     }
   },
-  "lowering_py_max_lines": 8478
+  "lowering_py_max_lines": 8511
 }

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
 
 

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(

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: