feat(gfql): native polars multi-source (node-cartesian) MATCH lowering (#1273)

Multi-source MATCH like `MATCH (a {..}), (b {..})` lowers to a
`rows(binding_ops=[Node, Node, ...])` op whose node_cartesian mode was
previously deferred (returned None -> NotImplementedError) on the polars
engine. Implement it natively, mirroring the pandas
`_gfql_cartesian_node_bindings_row_table` oracle exactly:

- per-alias frame = filter_by_dict + project into the
  `_gfql_node_alias_lookup_frame` schema (bare `alias` id, `alias.id`,
  `alias.<prop>`, and the leaked named-op flag `alias.alias = True` that
  shadows a same-named real property), then a left-major cross join so the
  row order matches pandas' constant-key merge (no ORDER BY needed).
- Decline (honest NIE, no pandas bridge) outside pandas' reliable zone:
  anonymous node ops (pandas raises a spurious schema error on empty
  results) and >3 named aliases (pandas' bare-id merge residue collides
  on the 4th frame). Both engines then error, never diverge.

Verified by a differential fuzz (random small graphs x multi-source query
shapes incl. WHERE / ORDER BY / property-name-vs-alias collisions) against
the pandas oracle: 0 disagreements over ~10k lowered cases across seeds.
Adds parametrized parity + explicit pin tests. No new ruff/mypy findings.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Y6dQEcjdazEnzuvuwf73ZL

Author: lmeyerov

graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py

@@ -870,6 +870,101 @@ def select_extend_polars(g: Plottable, items: Sequence[SelectItem]) -> Optional[
     return _rewrap(g, out)
 
 
+def _cartesian_node_bindings_polars(
+    g: Plottable,
+    ops: "Sequence[Any]",
+    node_id: Optional[str],
+) -> Optional[Plottable]:
+    """Native polars cross-product for disconnected MATCH aliases (#1273).
+
+    Mirrors the pandas ``_gfql_cartesian_node_bindings_row_table`` oracle: each
+    node alias is independently filtered, projected into the ``_gfql_node_alias_lookup_frame``
+    schema (``alias``, ``alias.node_id``, ``alias.<col>``, plus the leaked named-op
+    FLAG column ``alias.alias = True``), then cross-joined in op order (left-major,
+    matching pandas' ``merge`` order so no ORDER BY is needed for parity). The bare
+    ``node_id`` residue column that pandas carries (``id_x``/``id_y`` merge suffixes)
+    is intentionally dropped: no lowered query references it, and dropping avoids
+    polars cross-join column collisions on 3+ aliases.
+
+    Returns None to DECLINE (caller raises the honest NIE) outside the supported
+    subset: node ``query=`` params, ``alias == node_id`` (pandas' flag column
+    overwrites the id column — no sane shared semantics), and seeded re-entry
+    (already gated by the caller). NO-CHEATING: never bridges to pandas.
+    """
+    import polars as pl
+    from graphistry.compute.ast import ASTNode
+    from graphistry.compute.gfql.lazy import collect as _lazy_collect
+    from .predicates import filter_by_dict_polars
+
+    nodes = g._nodes
+    if nodes is None or node_id is None:
+        return None
+    node_id = str(node_id)
+    if node_id not in nodes.columns:
+        return None
+
+    aliases = [op._name for op in ops]
+    # Decline outside pandas' RELIABLE zone (empirically derived, keeps parity):
+    #  - anonymous node op: the pandas cartesian raises a spurious schema error on
+    #    an EMPTY result when a bare `()` is present (it drops the id column) rather
+    #    than returning empty — declining avoids that divergence.
+    #  - >3 named aliases: the pandas builder's per-alias bare-id merge residue
+    #    collides on the 4th frame ("Passing 'suffixes' which cause duplicate
+    #    columns"). Both engines must not diverge, so decline to the honest NIE.
+    if any(not isinstance(a, str) for a in aliases):
+        return None
+    named = [a for a in aliases if isinstance(a, str)]
+    if len(named) > 3:
+        return None
+
+    nodes_lf = nodes.lazy()
+    # filter_by_dict_polars is frame-polymorphic (LazyFrame in -> LazyFrame out) but
+    # annotated ``pl.DataFrame``; keep the accumulator loose, as this module does.
+    per_alias: List[Any] = []
+    for op in ops:
+        if not isinstance(op, ASTNode) or op.query is not None:
+            return None
+        alias = op._name
+        if not isinstance(alias, str):
+            return None
+        if alias == node_id:
+            # pandas' named-op flag column overwrites the id column here — neither
+            # engine has sane semantics; decline (mirrors the single-entity
+            # ``rows_binding_ops_polars`` corner).
+            return None
+        try:
+            matched = filter_by_dict_polars(nodes_lf, op.filter_dict)
+        except NotImplementedError:
+            raise
+        except Exception:
+            return None
+        cols = matched.collect_schema().names()
+        # prop_cols excludes node_id and any real column named == alias: the pandas
+        # node execute() leaks a boolean FLAG into a column named ``alias``
+        # (shadowing a same-named real property), which the lookup frame surfaces
+        # as ``alias.alias = True``. Reproduce that exactly.
+        prop_cols = [c for c in cols if c != node_id and c != alias]
+        exprs = [
+            pl.col(node_id).alias(alias),
+            pl.col(node_id).alias(f"{alias}.{node_id}"),
+            pl.lit(True).alias(f"{alias}.{alias}"),
+        ]
+        exprs.extend(pl.col(c).alias(f"{alias}.{c}") for c in prop_cols)
+        per_alias.append(matched.select(exprs))
+
+    if not per_alias:
+        return None
+    state = per_alias[0]
+    for frame in per_alias[1:]:
+        # Left-major cross join → same row order as the pandas constant-key merge.
+        state = state.join(frame, how="cross")
+    try:
+        out_df = _lazy_collect(state)
+    except pl.exceptions.SchemaError:
+        return None
+    return _rewrap(g, out_df)
+
+
 def binding_rows_polars(
     g: Plottable,
     binding_ops: Sequence[Dict[str, JSONVal]],
@@ -921,7 +1016,8 @@ def _names(lf: pl.LazyFrame) -> List[str]:
     # 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)
+        # MATCH (a), (b), ... disconnected node aliases: native cross-product (#1273).
+        return _cartesian_node_bindings_polars(g, ops, node_id)
     if RowPipelineMixin._gfql_is_shortest_path_scalar_binding_ops(ops):
         return None  # shortestPath scalar contract: BFS/native backends, pandas-only
 

graphistry/tests/compute/gfql/test_engine_polars_binding_rows.py

@@ -81,6 +81,13 @@ def _assert_parity(query, *, order_sensitive=False):
     "MATCH (a)-[*2..2]->(b) RETURN count(*) AS c",           # exactly-k
     "MATCH (a)-[:F*1..2]->(b) RETURN count(*) AS c",         # typed var-length
     "MATCH (a)-[*1..2]->(b)-[]->(c) RETURN count(*) AS c",   # var-length + fixed hop
+    # node-only cartesian: disconnected multi-source MATCH (#1273). <=3 named aliases.
+    "MATCH (a), (b) RETURN a.id AS ai, b.id AS bi ORDER BY ai, bi",
+    "MATCH (a {kind: 'a'}), (b {kind: 'b'}) RETURN a.id AS ai, b.id AS bi",
+    "MATCH (a {kind: 'a'}), (b) RETURN a.id AS ai, a.age AS aa, b.id AS bi",
+    "MATCH (a {kind: 'a'}), (b {kind: 'b'}) WHERE a.age > 20 RETURN a.id AS ai, b.id AS bi",
+    "MATCH (a), (b), (c {kind: 'a'}) RETURN a.id AS ai, b.id AS bi, c.id AS ci ORDER BY ai, bi, ci",
+    "MATCH (a {kind: 'z'}), (b) RETURN a.id AS ai, b.id AS bi",   # empty (no kind=z) keeps schema
 ]
 
 # Outside the MVP subset: must raise NotImplementedError (honest NIE, no bridge,
@@ -89,6 +96,10 @@ def _assert_parity(query, *, order_sensitive=False):
     "MATCH (a)-[*]->(b) RETURN count(*) AS c",               # unbounded var-length
     "MATCH (a)-[*1..2]-(b) RETURN count(*) AS c",            # undirected var-length
     "MATCH (a)-[e]->(b) WHERE a.age < b.age RETURN a.id",    # cross-alias same-path WHERE
+    # cartesian outside the pandas-reliable subset (#1273): pandas itself errors or
+    # is fragile here, so polars declines rather than diverge.
+    "MATCH (a), (b), (c), (d) RETURN a.id, b.id, c.id, d.id",  # >3 named aliases
+    "MATCH (a), (b), () RETURN a.id, b.id",                    # anonymous companion
 ]
 
 
@@ -121,6 +132,50 @@ def test_polars_binding_rows_raw_table_meaningful_cols():
     pd.testing.assert_frame_equal(a, b, check_dtype=False)
 
 
+def test_polars_cartesian_binding_rows_raw_meaningful_cols():
+    """Raw node-only cartesian rows(binding_ops): polars carries the same meaningful
+    per-alias schema as pandas (bare ``alias`` id, ``alias.id``, ``alias.<prop>``,
+    plus the leaked named-op flag ``alias.alias = True``), values equal to pandas."""
+    bo = serialize_binding_ops([n(name="a"), n(name="b")])
+    rpd = BASE.gfql([rows(binding_ops=bo)], engine="pandas")._nodes
+    rpl = BASE.gfql([rows(binding_ops=bo)], engine="polars")._nodes
+    assert "polars" in type(rpl).__module__
+    expected = {"a", "a.id", "a.age", "a.kind", "a.a", "b", "b.id", "b.age", "b.kind", "b.b"}
+    assert expected <= set(rpl.columns)
+    assert set(rpl.columns) <= set(rpd.columns)  # no columns pandas lacks
+    key = ["a", "b"]
+    a = rpd[list(rpl.columns)].sort_values(key).reset_index(drop=True)
+    b = rpl.to_pandas().sort_values(key).reset_index(drop=True)
+    pd.testing.assert_frame_equal(a, b, check_dtype=False)
+
+
+def test_polars_cartesian_alias_name_collides_with_property():
+    """A node property named the same as a MATCH alias is shadowed by the leaked
+    named-op flag (``alias.alias = True``) on BOTH engines — polars mirrors the
+    pandas quirk exactly rather than surfacing the real property value."""
+    nodes = pd.DataFrame({"id": [0, 1, 2], "kind": ["a", "b", "a"], "a": [10, 20, 30], "b": [1, 2, 3]})
+    g = graphistry.nodes(nodes, "id").edges(pd.DataFrame({"s": [0], "d": [1]}), "s", "d")
+    q = "MATCH (a {kind: 'a'}), (b {kind: 'b'}) RETURN a.id AS ai, a.a AS aa, b.id AS bi, b.b AS bb"
+    rpd = g.gfql(q, engine="pandas")._nodes.reset_index(drop=True)
+    rpl = g.gfql(q, engine="polars")._nodes.to_pandas().reset_index(drop=True)
+    assert list(rpd["aa"]) == [True, True] and list(rpd["bb"]) == [True, True]  # flag, not 10/30
+    pd.testing.assert_frame_equal(
+        rpd.sort_values(["ai", "bi"]).reset_index(drop=True),
+        rpl[rpd.columns.tolist()].sort_values(["ai", "bi"]).reset_index(drop=True),
+        check_dtype=False,
+    )
+
+
+def test_polars_cartesian_multiplicity_three_aliases():
+    """Three-alias cross product has |a|*|b|*|c| rows in left-major order on polars,
+    identical to pandas."""
+    q = "MATCH (a {kind: 'a'}), (b {kind: 'b'}), (c {kind: 'a'}) RETURN a.id AS ai, b.id AS bi, c.id AS ci"
+    rpd = BASE.gfql(q, engine="pandas")._nodes.reset_index(drop=True)
+    rpl = BASE.gfql(q, engine="polars")._nodes.to_pandas().reset_index(drop=True)
+    assert len(rpl) == 3 * 2 * 3  # kind a = {0,2,4}, kind b = {1,3}
+    pd.testing.assert_frame_equal(rpd, rpl[rpd.columns.tolist()], check_dtype=False)  # order-exact
+
+
 def test_binding_rows_projection_pushdown_skips_unused_props():
     """#1711: a query referencing no node properties (count(*)) attaches ZERO
     property columns to the binding table; one referencing only b's property
@@ -248,7 +303,16 @@ def test_polars_binding_rows_decline_branches_direct():
     assert binding_rows_polars(seeded, bo) is None
 
     g = graphistry.nodes(pl.from_pandas(NODES), "id").edges(pl.from_pandas(EDGES), "s", "d")
-    assert binding_rows_polars(g, serialize_binding_ops([n(name="a"), n(name="b")])) is None
+    # node-only cartesian (#1273) is now natively supported for <=3 named aliases;
+    # it declines only outside the pandas-reliable subset:
+    #  - anonymous node op (pandas raises a spurious schema error on empty results)
+    assert binding_rows_polars(g, serialize_binding_ops([n(name="a"), n()])) is None
+    #  - >3 named aliases (pandas' bare-id merge residue collides on the 4th frame)
+    assert binding_rows_polars(
+        g, serialize_binding_ops([n(name="a"), n(name="b"), n(name="c"), n(name="d")])
+    ) is None
+    # ...but 2-3 named aliases now lower natively (returns a row table, not None)
+    assert binding_rows_polars(g, serialize_binding_ops([n(name="a"), n(name="b")])) is not None
     assert binding_rows_polars(g, serialize_binding_ops([n(name="a", query="id > 0"), e_forward(), n(name="b")])) is None
     assert binding_rows_polars(g, serialize_binding_ops([n(name="a"), e_forward(source_node_match={"kind": "a"}), n(name="b")])) is None
     assert binding_rows_polars(g, serialize_binding_ops([n(name="a"), e_forward(label_seeds=True), n(name="b")])) is None