diff --git a/CHANGELOG.md b/CHANGELOG.md index 004f9c40cd..95b0306882 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -37,6 +37,9 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm - **GFQL Cypher parser: internal cleanup — WHERE consumed directly from `MatchClause`**: the grammar bundles a trailing `WHERE` onto its `MATCH` clause; the transformer previously split it back out into a synthetic standalone item and re-attached it, so the legacy clause-assembler ran unchanged (a temporary seam that kept the LALR switch byte-identical). The assembler now consumes `MatchClause.where` directly (primary MATCH keeps its WHERE on the clause; a post-WITH re-entry MATCH's WHERE goes to `reentry_wheres`), deleting the split/re-attach round-trip and the now-unreachable standalone-WHERE handling in both `query_body` and `graph_constructor`. Pure internal refactor, **no behavior change**: verified byte-identical ASTs vs the prior parser across a 1,989-query repo corpus, and the full cypher suite passes. - **GFQL Cypher parser: Earley → LALR(1) (~70× faster parse, same AST)**: `parse_cypher` now uses a single LALR(1) parser (contextual lexer) instead of Earley (dynamic lexer) — ~0.25ms vs ~17ms per query on representative queries. The WHERE grammar was unified to one `where_clause: "WHERE" expr` rule (the former `where_predicates | expr` dual rule was a genuine reduce/reduce ambiguity that forced Earley), and the structured flat-AND predicate form is recovered by a post-parse lift that re-parses the WHERE body with a LALR sub-parser (`start="where_predicate_chain"`). The full WHERE language is preserved — OR/XOR/NOT, parentheses, arithmetic, IN, pattern predicates — nothing is restricted to a subset. The grammar was then **purified to (near-)unambiguous**: WHERE binds to its preceding clause *in the grammar* (bundled into `match_clause`; the WITH..WHERE attachment ambiguity is eliminated, not tie-broken), and name-rooted dot chains derive only via `qualified_name` (the `property_access` redundancy is gone) — dropping the redundant `label_predicate_expr` rule (`(n:Admin)` parses via `grouped_expr`), and excluding a top-level `IN` from list-literal elements (`[x IN xs ...` is comprehension syntax; allowing a bare `IN` list element made it overlap). Net: the LALR conflict profile goes from 8 shift/reduce to **ZERO conflicts** — the grammar is now **provably unambiguous LALR(1)** and builds under Lark's `strict=True` (a build-time proof, machine-checked in CI as zero conflicts, plus a strict-mode build test where the optional `interegular` dep is present). Every input has a single derivation. **Machine-checked invariants** (`test_grammar_invariants.py`): (1) **zero LALR conflicts** (dependency-free, always-on in CI) plus a `strict=True` build test (skipped where the optional `interegular` dep is absent) — a grammar edit that introduces any ambiguity fails CI; (2) semantic ambiguity is ZERO — every corpus query has exactly one Earley derivation and it transforms to a single AST, no exceptions; (3) a rule-coverage gate (`test_every_grammar_rule_is_exercised_by_the_corpus`) forces every new grammar rule through the invariants; (4) differential tests run the production pipeline under both parsers (byte-identical ASTs, identical rejections). **Full-repo differentials** (1,850+ scraped queries, LALR-vs-Earley and old-vs-new production): **zero AST divergences**, and a small set of deliberate language fixes — accept-by-accident shapes with ill-defined semantics, now honest syntax errors and pinned as tests (`DELIBERATE_LANGUAGE_FIXES`): `RETURN DISTINCT` with no items (Earley re-lexed DISTINCT as a variable name), double WHERE (the old positional attachment kept *both* predicates in different AST fields), double post-WITH WHERE, WHERE after UNWIND, WHERE before MATCH in a graph constructor, and the invalid "list of IN-booleans" (`[x IN xs, y]`; use `[(x IN xs), y]` for a genuine bool list). The lift stays an internal optimization: only a flat `AND` chain over present columns lifts to `filter_dict`; parens / OR / XOR / NOT and any absent-column case stay on `where_rows` (a correctness boundary, since `where_rows` treats an absent property as null while `filter_dict` would raise). Full cypher suite (1,681 tests) passes. Downstream execution (`filter_dict` vs `where_rows` routing) is unchanged. A pre-existing `<>`-over-null 3VL divergence between the two execution paths (surfaced by #1653's metamorphic test) is tracked separately in #1683. +### Added +- **GFQL native Polars bindings-row tables (`rows(binding_ops)`) — traversal Cypher on polars (#1709)**: the Cypher multi-alias lowering's `rows(binding_ops=...)` op (one row per matched path) now runs natively on `engine='polars'` for **fixed-length connected patterns** — unblocking traversal-shaped Cypher that previously NIE'd: multi-alias property projections (`MATCH (a)-[e]->(b) RETURN a.x, e.w, b.y`), top-k in-degree (`RETURN b.id, count(a) ... ORDER BY ... LIMIT`, graph-benchmark q1/q2), and fixed multi-hop counts (`MATCH (a)-->(b)-->(c) RETURN count(*)`, q8/q9), with forward/reverse/undirected edges, node/edge filters, and edge-alias payload columns — **plus bounded directed variable-length segments** (`-[*i..k]->`, typed `-[:TYPE*i..k]->`, exactly-k; iterative pair joins with Cypher path multiplicity and zero-hop rows), covering the q3 shape (`MATCH (a:Person)-[:FOLLOWS*1..2]->(b:Person) ... RETURN avg(b.age)`). This rung covers the q1–q4 and q8–q9 binding-table shapes exercised by its parity tests; connected q5–q7 planning/execution is layered in follow-on PRs. Also adds native `with_(extend=True)` (emitted by the bindings-path aggregate lowering) and an honest decline for `group_by(key_prefixes=...)` (whole-row bindings grouping — was a latent silent-wrong-key trap). **Honestly deferred** (`NotImplementedError`, no pandas bridge): unbounded `[*]`, undirected/aliased variable-length, shortestPath scalar bindings, node/edge `query=`/endpoint-match params, cartesian (`MATCH (a),(b)`) mode, seeded re-entry contexts. Differential parity vs the pandas oracle (+20 tests incl. path multiplicity and undirected self-loops; conformance corpus extended). The Polars-GPU engine uses the same lazy target-collected plan; exact-head DGX validation is required before making a GPU performance claim. + ### Fixed - **GFQL seeded indexes now engage for small Polars/cuDF frontiers**: NaN-free native Polars frames retain object identity during normalization, preserving resident CSR validity, while a tunable absolute floor prevents the fractional cost gate from routing tiny seeded hops over low-cardinality slices to a full scan. Scan/index result parity is unchanged; the default floor is a routing heuristic whose engine/data crossover remains subject to measurement. - **GFQL Cypher `GRAPH { }` residual predicates now fail safely or apply as graph masks**: `GRAPH { MATCH ... WHERE ... }` no longer silently drops predicates that the graph-state path cannot apply. Safe one-node/one-edge residual filters, including disjunctions and `searchAny(...)`, are applied before graph-state matching; unsupported pattern-predicate, multi-alias, and Polars graph-residual cases now raise clear validation errors instead of returning an over-broad subgraph. diff --git a/bin/ci_cypher_surface_guard_baseline.json b/bin/ci_cypher_surface_guard_baseline.json index bbf8dbb481..046156097c 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": 8511 + "lowering_py_max_lines": 8779 } diff --git a/graphistry/compute/ast.py b/graphistry/compute/ast.py index 4eb0dd4281..af15ebca0a 100644 --- a/graphistry/compute/ast.py +++ b/graphistry/compute/ast.py @@ -1576,6 +1576,7 @@ def rows( alias_endpoints: Optional[Dict[str, str]] = None, binding_ops: Optional[List[Dict[str, JSONVal]]] = None, alias_prefilters: Optional[AliasPrefilters] = None, + attach_prop_aliases: Optional[List[str]] = None, ) -> ASTCall: """Create a row-source operation for GFQL row pipelines. @@ -1601,6 +1602,8 @@ def rows( params["binding_ops"] = binding_ops if alias_prefilters: params["alias_prefilters"] = alias_prefilters + if attach_prop_aliases is not None: + params["attach_prop_aliases"] = list(attach_prop_aliases) return ASTCall("rows", params) diff --git a/graphistry/compute/chain.py b/graphistry/compute/chain.py index 96044f5c64..13b1e5ead2 100644 --- a/graphistry/compute/chain.py +++ b/graphistry/compute/chain.py @@ -953,10 +953,16 @@ def _chain_impl( GFQL_EDGE_INDEX = generate_safe_column_name('edge_index', g._edges, prefix='__gfql_', suffix='__') added_edge_index = True - indexed_edges_df = g._edges.reset_index(drop=False) - original_cols = set(g._edges.columns) - index_col_name = next(col for col in indexed_edges_df.columns if col not in original_cols) - indexed_edges_df = indexed_edges_df.rename(columns={index_col_name: GFQL_EDGE_INDEX}) + # Attach the synthetic per-edge id WITHOUT copying edge data (#1670): + # the previous reset_index(drop=False) + rename deep-copied AND + # block-consolidated the whole edge frame (~70ms @2M edges) on every + # chain call — even node-only queries. A shallow copy + assigning the + # index as a column yields the identical id values (the frame's index) + # with no O(E) data copy. The column is internal-only — dropped on + # every exit path (see added_edge_index consumers below) — so only + # uniqueness matters. + indexed_edges_df = g._edges.copy(deep=False) + indexed_edges_df[GFQL_EDGE_INDEX] = indexed_edges_df.index g = g.edges(indexed_edges_df, edge=GFQL_EDGE_INDEX) else: added_edge_index = False diff --git a/graphistry/compute/gfql/call/validation.py b/graphistry/compute/gfql/call/validation.py index 63dcd462bf..42c150396e 100644 --- a/graphistry/compute/gfql/call/validation.py +++ b/graphistry/compute/gfql/call/validation.py @@ -360,13 +360,14 @@ 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', 'alias_prefilters'}, + {'table', 'source', 'alias_endpoints', 'binding_ops', 'alias_prefilters', 'attach_prop_aliases'}, 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, + 'attach_prop_aliases': lambda v: isinstance(v, list) and all(isinstance(x, str) for x in v), }, 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 9d03a88e8e..5606a84b0d 100644 --- a/graphistry/compute/gfql/cypher/lowering.py +++ b/graphistry/compute/gfql/cypher/lowering.py @@ -81,6 +81,7 @@ collect_identifiers, parse_expr, walk_expr_nodes, + is_expr_node, ) from graphistry.compute.gfql.cypher.reentry_plan import ReentryPlan from graphistry.compute.gfql.cypher.ast import ( @@ -2235,6 +2236,215 @@ def _active_match_alias_for_stage( return next(iter(alias_targets)) +def _projection_ref_from_expr_safe( + expr_text: str, alias_targets: Mapping[str, ASTObject] +) -> Optional[Tuple[str, str]]: + """(alias, prop) if ``expr_text`` is a bare ``alias.prop`` of a known alias, else None.""" + text = (expr_text or "").strip() + if "." not in text: + return None + alias, _, prop = text.partition(".") + alias = alias.strip() + prop = prop.strip() + if not alias or not prop or "." in prop: + return None + if alias not in alias_targets: + return None + if not prop.replace("_", "").isalnum(): + return None + return alias, prop + + +def _clause_has_mixed_aggregate_item( + query: CypherQuery, + *, + alias_targets: Mapping[str, ASTObject], + params: Optional[Mapping[str, Any]], +) -> bool: + """True if any RETURN / ORDER BY item mixes a non-aggregate alias reference with + an aggregate in ONE expression (e.g. ``me.age + count(you.age)``). Such compound + cross-source items have ambiguous multiplicity and must keep the conservative + fail-fast; a clean split (``c.city`` and ``avg(p.age)`` as separate items) does + not trip this (#1273 / rejects-unsound-multi-source-overlap contract).""" + return_clause = query.return_ + exprs: List[Tuple[str, int, int]] = [] + if return_clause is not None: + for item in return_clause.items: + exprs.append((item.expression.text, item.span.line, item.span.column)) + order_by = query.order_by # top-level ORDER BY (not on ReturnClause) + if order_by is not None: + for order_item in order_by.items: + exprs.append((order_item.expression.text, order_item.span.line, order_item.span.column)) + for text, line, column in exprs: + if text == "*": + continue + try: + node = _parse_row_expr( + text, params=params, alias_targets=alias_targets, + allow_missing_params=True, field="return", line=line, column=column, + ) + except GFQLValidationError: + return True # can't analyze -> conservative (treat as mixed / fail-fast) + # An aggregate nested inside a larger expression (arithmetic / function / + # comparison) — e.g. ``me.age + count(you.age)`` or ``age + count(...)`` in + # ORDER BY — is a compound cross-source item with ambiguous multiplicity. + # A bare top-level aggregate call (``avg(p.age)``) or a pure group scalar + # (``c.city``) is clean and does NOT trip this. + if _expr_has_aggregate(node) and not _is_pure_aggregate_call(node): + return True + return False + + +def _is_pure_aggregate_call(node: ExprNode) -> bool: + return isinstance(node, FunctionCall) and node.name.lower() in _CYPHER_AGGREGATES + + +def _expr_has_aggregate(node: ExprNode) -> bool: + """True if the expression contains an aggregate function call anywhere.""" + if isinstance(node, FunctionCall) and node.name.lower() in _CYPHER_AGGREGATES: + return True + for child in getattr(node, "__dict__", {}).values(): + if is_expr_node(child): + if _expr_has_aggregate(cast(ExprNode, child)): + return True + elif isinstance(child, (list, tuple)): + for c in child: + if is_expr_node(c) and _expr_has_aggregate(cast(ExprNode, c)): + return True + return False + + +def _binding_prop_alias_set( + query: CypherQuery, + *, + alias_targets: Mapping[str, ASTObject], + params: Optional[Mapping[str, Any]], +) -> Optional[List[str]]: + """#1711 projection-pushdown: node aliases whose PROPERTIES are referenced by + the RETURN/ORDER BY, so the binding builders can skip property joins for the + rest (e.g. ``count(*)`` needs none; ``count(a)`` needs only a's bare id column). + + Returns a list of node-alias names to attach, or ``None`` = attach all (the + conservative default). Deliberately CONSERVATIVE — only computed for the simple + single-clause shape (no WITH stages, no WHERE anywhere): a WHERE predicate may + run on the binding table and need a property column, and multi-stage pipelines + carry hidden reentry/carry columns; both are hard to bound safely, so we decline + to optimize them (they keep the current attach-all behavior). The referenced set + itself is EXACT: ``_expr_match_alias_usage`` non-aggregate refs are precisely the + property / whole-entity uses; aggregate-only refs (``count(a)``) are excluded. + """ + if query.with_stages: + return None + if query.where is not None: + return None + matches = query.matches or () + if len(matches) != 1: + return None # multi-MATCH / cartesian — conservative + match_clause = matches[0] + if match_clause.where is not None or match_clause.optional: + return None + return_clause = query.return_ + if return_clause is None: + return None + + # A repeated node alias (e.g. `MATCH (n)-[:LOOP]->(n)`) enforces n==n via hidden + # bound-identity columns (`n.__gfql_node_id__`) that a RETURN-text walk can't see — + # skipping n's property join would drop them. Bail on any repeat (#1490). + try: + node_vars = [ + el.variable + for el in _match_pattern_elements(match_clause) + if isinstance(el, NodePattern) and el.variable is not None + ] + except (GFQLValidationError, RuntimeError): + return None + if len(node_vars) != len(set(node_vars)): + return None + # `collect(...)` triggers the carry/reentry machinery (hidden columns) — bail (#1413). + try: + agg_specs = _collect_aggregate_specs_for_clause( + return_clause, params=params, alias_targets=alias_targets + ) + except (GFQLValidationError, RuntimeError): + return None + if any(spec.func == "collect" for spec in agg_specs): + return None + + node_aliases = {a for a, t in alias_targets.items() if isinstance(t, ASTNode)} + if not node_aliases: + return None + + exprs: List[Tuple[str, int, int]] = [] + for item in return_clause.items: + exprs.append((item.expression.text, item.span.line, item.span.column)) + order_by = query.order_by # top-level ORDER BY (not on ReturnClause) + if order_by is not None: + for order_item in order_by.items: + exprs.append((order_item.expression.text, order_item.span.line, order_item.span.column)) + + referenced: Set[str] = set() + for text, line, column in exprs: + if text == "*": + continue + try: + non_aggregate_aliases, _agg = _expr_match_alias_usage( + text, + alias_targets=alias_targets, + params=params, + field="return", + line=line, + column=column, + ) + # Property accesses INSIDE aggregates need their columns too — e.g. + # ``avg(p.age)`` requires ``p``'s property join even though ``p`` is only + # referenced within the aggregate (#1273). ``non_aggregate_aliases`` alone + # (which excludes aggregate context) would drop it -> a missing column. + prop_aliases = _expr_property_access_node_aliases( + text, alias_targets=alias_targets, params=params, + field="return", line=line, column=column, + ) + except GFQLValidationError: + return None # can't analyze cleanly → conservative attach-all + referenced.update(a for a in non_aggregate_aliases if a in node_aliases) + referenced.update(a for a in prop_aliases if a in node_aliases) + return sorted(referenced) + + +def _expr_property_access_node_aliases( + expr_text: str, + *, + alias_targets: Mapping[str, ASTObject], + params: Optional[Mapping[str, Any]], + field: str, + line: int, + column: int, +) -> Set[str]: + """Node aliases that appear in a ``alias.prop`` property access anywhere in the + expression (INCLUDING inside aggregates). Used by #1711 projection-pushdown so a + property referenced only via ``avg(alias.prop)`` still keeps that alias's join.""" + node = _parse_row_expr( + expr_text, params=params, alias_targets=alias_targets, + allow_missing_params=True, field=field, line=line, column=column, + ) + out: Set[str] = set() + + def _visit(n: ExprNode) -> None: + if isinstance(n, PropertyAccessExpr) and isinstance(n.value, Identifier): + root = n.value.name.split(".", 1)[0] + if root in alias_targets: + out.add(root) + for child in getattr(n, "__dict__", {}).values(): + if is_expr_node(child): + _visit(cast(ExprNode, child)) + elif isinstance(child, (list, tuple)): + for c in child: + if is_expr_node(c): + _visit(cast(ExprNode, c)) + + _visit(node) + return out + + def _is_multi_source_match_alias_boundary_error( exc: GFQLValidationError, *, @@ -6551,6 +6761,45 @@ def _lower_general_row_projection( break continue if len(refs) > 1 or (len(refs) == 1 and base_active_alias not in refs): + # An aggregate over a pattern alias other than the projection's + # active alias. Two sound, benchmark-relevant shapes are routed to + # the bindings-row table (which materializes every alias, one row + # per matched path); everything else keeps the conservative + # fail-fast (the misleading "one MATCH" error is the residual). + # (a) #1708: `count()` — "matched paths binding + # this node per group" (graph-bench q1 top-k in-degree). + # (b) #1273: a CLEAN grouped aggregate `func(.)` + # (avg/sum/min/max/count) grouped by another alias's property + # (graph-bench q3/q4: `RETURN c.city, avg(p.age)`) — a + # standard GROUP BY, sound on the per-path bindings rows. + # BOTH require CLEAN agg/non-agg separation: if any RETURN/ORDER BY + # item MIXES a non-aggregate ref with an aggregate in one + # expression (`me.age + count(you.age)`), the cross-source + # multiplicity is ambiguous — keep the fail-fast (the + # rejects-unsound-multi-source-overlap contract, #1273 tests). + agg_arg = (agg_spec.expr_text or "").strip() + prop_ref = _projection_ref_from_expr_safe(agg_arg, alias_targets) + prop_alias = prop_ref[0] if prop_ref is not None else None + if ( + refs <= set(alias_targets.keys()) + and not _clause_has_mixed_aggregate_item( + query, alias_targets=alias_targets, params=params + ) + and ( + ( # (a) bare-alias non-distinct count + agg_spec.func == "count" + and not agg_spec.distinct + and agg_arg in alias_targets + and isinstance(alias_targets.get(agg_arg), ASTNode) + ) + or ( # (b) clean property aggregate over a node alias + agg_spec.func in ("avg", "sum", "min", "max", "count") + and prop_alias is not None + and isinstance(alias_targets.get(prop_alias), ASTNode) + ) + ) + ): + continue can_force_bindings = False break if can_force_bindings: @@ -6573,7 +6822,14 @@ def _lower_general_row_projection( if active_match_alias is None: row_steps: List[ASTObject] = [rows(table="nodes")] elif binding_row_aliases: - row_steps = [rows(binding_ops=serialize_binding_ops(lowered.query))] + row_steps = [ + rows( + binding_ops=serialize_binding_ops(lowered.query), + attach_prop_aliases=_binding_prop_alias_set( + query, alias_targets=alias_targets, params=params + ), + ) + ] else: row_steps = [ rows( @@ -7568,9 +7824,24 @@ def _compile_connected_match_join( shared_aliases_per_pattern.append(shared) accumulated_aliases.update(node_aliases) - if query.where is not None and query.where.expr_tree is not None: + 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) - assert synthesized is not None # gated by expr_tree is not None + if synthesized 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", + field="where", + value=None, + line=clause.span.line, + column=clause.span.column, + ) pre_join_filters.append(synthesized) for projection_clause in [stage.clause for stage in query.with_stages] + [query.return_]: diff --git a/graphistry/compute/gfql/lazy/engine/polars/chain.py b/graphistry/compute/gfql/lazy/engine/polars/chain.py index 463da88112..bfe469b056 100644 --- a/graphistry/compute/gfql/lazy/engine/polars/chain.py +++ b/graphistry/compute/gfql/lazy/engine/polars/chain.py @@ -353,7 +353,7 @@ def _try_native_row_op(g_cur, op): from graphistry.Engine import Engine from .row_pipeline import ( select_polars, with_columns_polars, order_by_polars, group_by_polars, - unwind_polars, where_rows_polars, + unwind_polars, where_rows_polars, binding_rows_polars, ) from .pattern_apply import ( rows_binding_ops_polars, semi_apply_mark_polars, anti_semi_apply_polars, @@ -361,6 +361,20 @@ def _try_native_row_op(g_cur, op): from .search import search_any_polars fn = getattr(op, "function", None) + if ( + fn == "rows" + and op.params.get("binding_ops") is not None + and op.params.get("alias_endpoints") is None + ): + # Multi-alias bindings table (#1709): native for fixed-length connected + # patterns. A decline must fall through to the pre-existing correlated + # pattern handler below (EXISTS/searchAny); returning None here would turn + # those already-native shapes into an NIE. + bindings_result = binding_rows_polars( + g_cur, op.params["binding_ops"], op.params.get("attach_prop_aliases") + ) + if bindings_result is not None: + return bindings_result if _call_native_on_polars(op): # frame ops (rows/limit/skip/distinct/drop_cols) — engine-polymorphic return op.execute(g=g_cur, prev_node_wavefront=None, target_wave_front=None, engine=Engine.POLARS) @@ -405,6 +419,11 @@ def _try_native_row_op(g_cur, op): if fn == "order_by": return order_by_polars(g_cur, op.params.get("keys", [])) if fn == "group_by": + if op.params.get("key_prefixes"): + # Whole-row grouping on a bindings table (alias-prefixed key expansion): + # silently ignoring key_prefixes would group on the wrong keys — a + # wrong answer. Decline until natively ported. + return None return group_by_polars(g_cur, op.params.get("keys", []), op.params.get("aggregations", [])) if fn == "unwind": return unwind_polars(g_cur, op.params.get("expr", ""), op.params.get("as_", "value")) diff --git a/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py b/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py index 34e6dfa72c..c7c33a940d 100644 --- a/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py +++ b/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py @@ -23,6 +23,7 @@ from graphistry.compute.gfql.expr_parser import ExprNode, FunctionCall from graphistry.Plottable import Plottable +from graphistry.utils.json import JSONVal from .dtypes import is_float as _dtype_is_float, is_int as _dtype_is_int, is_numeric as _dtype_is_numeric, is_stringlike as _dtype_is_stringlike @@ -607,6 +608,19 @@ def _lower_with_schema(table: Any, fn): _SCHEMA.reset(token) +def _project_preserving_height(table: Any, exprs: List[Any]) -> Any: + """Project ``exprs`` while preserving the frame's row cardinality. + + Cypher ``WITH``/``RETURN`` projection is a map, not a reduce. Polars + ``DataFrame.select`` collapses to one row when every projected expression is + scalar, so broadcast all-scalar projections through ``with_columns`` first. + """ + if exprs and all(len(e.meta.root_names()) == 0 for e in exprs): + names = [e.meta.output_name() for e in exprs] + return table.with_columns(exprs).select(names) + return table.select(exprs) + + def _project_polars(g: Plottable, items: Sequence[Any], extend: bool) -> Optional[Plottable]: """Shared body of ``select_polars`` / ``with_columns_polars``; None if any item isn't lowerable (honest NIE, no pandas bridge).""" @@ -614,7 +628,7 @@ def _project_polars(g: Plottable, items: Sequence[Any], extend: bool) -> Optiona exprs = _lower_with_schema(table, lambda: lower_select_items(items, list(table.columns))) if exprs is None: return None - out = table.with_columns(exprs) if extend else table.select(exprs) + out = table.with_columns(exprs) if extend else _project_preserving_height(table, exprs) if _select_emits_temporal_constructor_text(out): # decline (NIE): projected String column holds temporal-constructor text (date({...}) # etc.) that pandas normalizes to ISO, not yet native — don't leak the raw text. @@ -791,3 +805,271 @@ def unwind_polars(g: Plottable, expr: str, as_: str = "value") -> Optional[Plott rhs = pl.DataFrame({as_: values}) return _rewrap(g, table.join(rhs, how="cross")) + +def select_extend_polars(g: Plottable, items: Sequence[Any]) -> Optional[Plottable]: + """Native polars ``with_(items, extend=True)``: add/overwrite projected columns + while keeping the existing row table (pandas ``assign`` semantics). Emitted by + the bindings-path aggregate lowering (pre-aggregation group keys / agg args), + so it is required for binding-row queries (#1709). None → NIE.""" + table = _active_table(g) + exprs = _lower_with_schema(table, lambda: lower_select_items(items, list(table.columns))) + if exprs is None: + return None + out = table.with_columns(exprs) + if _select_emits_temporal_constructor_text(out): + return None + return _rewrap(g, out) + + +def binding_rows_polars( + g: Plottable, + binding_ops: Sequence[Dict[str, JSONVal]], + attach_prop_aliases: Optional[Sequence[str]] = None, +) -> Optional[Plottable]: + """Native polars bindings-row table for FIXED-LENGTH connected patterns (#1709). + + Materializes one row per matched path for an alternating ``n/e/n/...`` pattern + (the ``rows(binding_ops=...)`` op emitted by Cypher multi-alias lowering), with + the same meaningful schema as the pandas engine: bare ``alias`` id columns, + ``edge_alias.col`` edge-payload columns, and ``alias.{col}`` node-property + columns per node alias. (The pandas frame additionally carries join-residue + columns — raw ``node_id``, ``a__a_join__``, leaked ``__gfql_edge_index__`` — + that no lowered query references; those are intentionally not replicated.) + + Returns None to DECLINE (caller raises the honest NIE) for anything outside + the supported subset: variable-length/multi-hop edges, shortestPath scalar + bindings, node ``query=`` / edge query or endpoint-match params, hop labels, + HAS_-label destination disambiguation, seeded re-entry contexts, cartesian + (node-only) mode, and the legacy ``alias_endpoints`` variant. NO-CHEATING: + never bridges to pandas. Parity gate: differential tests vs the pandas oracle. + """ + import polars as pl + from graphistry.compute.ast import ASTEdge, ASTNode, ASTObject, from_json as ast_from_json + from graphistry.compute.gfql.lazy import collect as _lazy_collect + from graphistry.compute.gfql.row.pipeline import RowPipelineMixin + from graphistry.compute.gfql.same_path.edge_semantics import EdgeSemantics + from .predicates import filter_by_dict_polars + + def _names(lf: pl.LazyFrame) -> List[str]: + # LazyFrame column names WITHOUT collecting data (schema-only resolve). + return lf.collect_schema().names() + + nodes = g._nodes + edges = g._edges + node_id = g._node + src = g._source + dst = g._destination + if nodes is None or edges is None or node_id is None or src is None or dst is None: + return None + if getattr(g, "_gfql_start_nodes", None) is not None: + # Bounded re-entry seeds the first alias from carried rows — pandas-only. + return None + + ops: List[ASTObject] = [ast_from_json(op_json, validate=False) for op_json in binding_ops] + # Shared validation (engine-agnostic): raises the canonical GFQLValidationError + # for malformed op sequences / duplicate aliases — same error as pandas. + RowPipelineMixin._gfql_validate_binding_ops(ops) + if RowPipelineMixin._gfql_binding_ops_mode(ops) == "node_cartesian": + return None # MATCH (a), (b) cross joins: deferred (rare; own schema study) + if RowPipelineMixin._gfql_is_shortest_path_scalar_binding_ops(ops): + return None # shortestPath scalar contract: BFS/native backends, pandas-only + + for idx, op in enumerate(ops): + if idx % 2 == 0: + if not isinstance(op, ASTNode) or op.query is not None: + return None + else: + if not isinstance(op, ASTEdge): + return None + sem = EdgeSemantics.from_edge(op) + if sem.is_multihop: + # Bounded directed var-length (`-[*1..k]->`, graph-bench q3) is + # supported via iterative pair joins; everything else declines: + # unbounded (`[*]`, needs fixed-point + termination error), + # undirected multihop (immediate-backtrack avoidance not ported), + # and aliased var-length edges (pandas rejects those outright). + if ( + op.direction == "undirected" + or bool(op.to_fixed_point) + or (op.max_hops is None and op.hops is None) + or isinstance(op._name, str) + ): + return None + if op.direction not in ("forward", "reverse", "undirected"): + return None + if any( + value is not None + for value in ( + op.edge_query, op.source_node_match, op.destination_node_match, + op.source_node_query, op.destination_node_query, + op.label_node_hops, op.label_edge_hops, + op.output_min_hops, op.output_max_hops, + ) + ): + return None + if bool(op.label_seeds) or bool(op.include_zero_hop_seed): + return None + # Duplicate-id + HAS_