feat(gfql): native polars UNWIND of a carried collect() list column (#1752)

* feat(gfql): native polars UNWIND of a carried collect() list column

Extend `unwind_polars` to explode a List-dtype column reference (a `collect()`
output / carried binding), not just a scalar-literal list. This is the row-
pipeline analogue of the IC6 `WITH collect(x) AS xs UNWIND xs AS y` shape:
previously the second UNWIND declined (NIE) because the expr was an Identifier,
not a ListLiteral, forcing engine='pandas'.

Mirrors the pandas oracle (`RowPipelineMixin.unwind` list-column branch)
exactly: empty-list and null cells contribute 0 rows; nulls WITHIN a list
survive as real elements; the source column is retained and exploded values
append as `as_`. Implemented as with_columns(copy) -> filter(list.len() > 0)
-> explode, which also makes the result independent of the polars 2.0
`empty_as_null` default. Non-list columns, name collisions, unknown
identifiers, and nested-list literals still decline (NIE) — no pandas bridge,
no silent-wrong.

NOTE (out of scope): the collect->UNWIND->MATCH re-entry form (IC6 with a
trailing MATCH) is compile-time rewritten to whole-row MATCH-after-WITH
re-entry; polars declines there because the whole-entity projection does not
attach `_cypher_entity_projection_meta` and `binding_rows_polars` declines on
`_gfql_start_nodes`. That is the separate WITH-MATCH re-entry residual, not a
row-pipeline unwind gap.

Correctness gate: differential fuzz vs pandas (2000 queries, 0 disagreements)
plus parity/pin/decline tests. ruff clean; no new mypy errors; new lines
fully covered.

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

* docs(gfql): CHANGELOG entry for native polars collect()->UNWIND

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

---------

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: lmeyerov

CHANGELOG.md

@@ -12,6 +12,7 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 - **GFQL engine-selection docs (pandas / polars / cuDF / polars-gpu)**: New :doc:`Choosing a GFQL Engine <gfql/engines>` page — a numbers-first, persona-tested guide to the four interchangeable engines. Adds the one-keyword `engine='polars'` speedup (up to ~38× over pandas on real graphs, no GPU), a motivating warm-median comparison table on real public graphs (LiveJournal 35M / Orkut 117M), a decision matrix (workload shape × size × hardware → engine, with the measured ~10K-edge CPU crossover, GPU-work-bound rule, polars-gpu memory-pressure caveat, and GPU-or-error contract), a cuDF-vs-polars-gpu disambiguation (eager-op vs fused-lazy; cuDF is not deprecated), an honest "when *not* to use Polars" section, the differential-parity guarantee, and a methodology + reproducer-script disclosure. Rewrote the top of `gfql/performance.rst` to lead with the engine comparison (de-marketed the prose), wired the new page into the GFQL toctree + recommended paths, and added Polars/polars-gpu to the engine examples in `gfql/quick.rst` and `gfql/about.rst` (previously only pandas/cuDF were documented). Driven by 4-persona doc user-testing (pandas DS, RAPIDS user, perf engineer, skeptical evaluator).
 
 ### Fixed
+- **GFQL polars engine natively runs `UNWIND` of a carried `collect()` list column**: `... WITH collect(x) AS xs UNWIND xs AS y RETURN ...` (exploding a list-valued column produced by `collect()`, the row-pipeline analogue of the IC6 unwind shape) previously declined on polars — `unwind_polars` only accepted a scalar literal list. Now explodes a `List`-dtype column reference (`with_columns` copy → `filter(list.len() > 0)` → `explode`), matching the pandas oracle exactly: empty-list/null cells → 0 rows, nulls *within* a list survive, source column retained; the pre-filter also makes it independent of the polars-2.0 `empty_as_null` default. Non-list columns, name collisions, unknown identifiers, and nested-list literals still decline (honest NIE, no pandas bridge). Differential fuzz vs pandas over ~3500 collect→UNWIND→{RETURN, grouped, filtered, ORDER BY, aggregate, nested-unwind} queries: 0 disagreements. (The `MATCH ... WITH collect(..) UNWIND .. MATCH` form — IC6 with a trailing re-traversal — is compile-rewritten into WITH→MATCH reentry, a separate path.) Tests in `test_engine_polars_row_pipeline.py`.
 - **GFQL polars engine natively runs multi-source (node-cartesian) MATCH (#1273)**: comma-separated disconnected aliases — `g.gfql("MATCH (a {..}), (b {..}) RETURN a.id, b.id")` — previously declined on polars (the `rows(binding_ops)` `node_cartesian` branch hard-NIE'd) while pandas handled them. New `_cartesian_node_bindings_polars` mirrors the pandas oracle `_gfql_cartesian_node_bindings_row_table`: each alias is independently filtered, projected into the per-alias lookup schema (bare `alias` id, `alias.id`, `alias.<prop>`, and pandas' leaked `alias.alias=True` flag incl. property-shadowing), then left-major cross-joined for row-order parity. Scalar/aliased/`count(*)` projections over the cartesian match pandas exactly (verified end-to-end); whole-entity `RETURN a, b` stays an honest NIE (separate projection surface). Two parity-safe declines mirror shapes where pandas itself errors (proven on master, so both engines fail identically, never diverge): an anonymous node op, and ≥4 named aliases (pandas' bare-id merge residue collides). Differential fuzz vs pandas over ~10k lowered cases: 0 disagreements. Tests in `test_engine_polars_binding_rows.py`.
 - **GFQL polars chain — variable-length node aliases are hop-distance gated; the chain is now silent-wrong-free (#1741)**: a node named after a variable-length edge (`g.gfql("MATCH (a)-[*1..2]-(b) RETURN b")`) previously carried its alias regardless of hop distance, so an undirected walk that backtracked into the seed wrongly returned it (pandas correctly excludes it — trail semantics). The polars chain now auto-injects the #1741 hop-distance label (name resolved against the user's node columns via the `reserved_columns` registry, so a user column named like the internal one is never clobbered) and applies pandas' alias `[min_hop, max_hop]` window in `_apply_node_names`. The one shape the gate can't yet cover — a node alias after a **forward/reverse `min_hops>1`** edge, whose labels need pandas' layered backward walk (not ported) — now **declines with an honest `NotImplementedError` (#1748)** instead of returning nodes outside the window; the decline is precise (differential vs pandas: 30/30 named shapes NIE, 30/30 unnamed run and match, 0 over-decline). 4-engine A/B on the stacked build: pandas/cudf 144/144, polars/polars-gpu 112 agree / **0 disagree** / 32 honest-NIE — zero silent-wrong shapes remain on the polars chain. Adapts the mechanism from the retired #1742 (declined undirected varlen+alias, now natively gated). Tests: `TestVarlenAliasHopGate` (pandas-parity across directions, the `*2..3` decline + unnamed-still-runs pins, a column-collision test).
 - **GFQL native polars `label_node_hops` on the plain BFS — correct, direction-dependent hop labels (#1741 groundwork)**: the polars eager hop now emits pandas-parity node hop-distance labels instead of declining `label_node_hops`. The labeling rule is DIRECTION-DEPENDENT (the subtle part, derived by differential fuzz vs the pandas oracle, not by reading pandas alone): forward/reverse label EVERY destination of a hop first-wins (matching pandas `hop.py` `new_node_ids`), so a seed re-entered at hop 1 IS labeled; undirected labels destinations MINUS everything already visited and pre-seeds the seen-set with the seeds, so a seed re-reached by a backtracking walk stays NULL (its shortest-path distance). Two correctness fixes over the first cut: (1) the undirected seed pre-seed must NOT depend on `return_as_wave_front` — a seed filtered out of the frontier by `source_node_match` or suppressed in wave-front mode was being re-labeled (A/B vs pandas over direction × hops × `return_as_wave_front` × `source_node_match` × `destination_node_match` × 10 graphs: was 456/24, now 480/480); (2) a requested `label_node_hops` name that collides with an existing column is redirected to `<name>_1` like pandas' `resolve_label_col`, not the polars left-join auto-suffix `<name>_right`/`DuplicateError`. `label_edge_hops` and `min_hops>1` labels stay honest NIEs. Amplified `test_engine_polars_hop.py` (`TestHopLabelsDifferential` over the seed-filter / wave-front / multi-seed axes, an explicit direction-asymmetry contrast pin, and a collision test).

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

@@ -831,11 +831,21 @@ def group_by_polars(
 
 
 def unwind_polars(g: Plottable, expr: str, as_: str = "value") -> Optional[Plottable]:
-    """Native UNWIND for a literal list: cross-join each row with the values (cypher per-row
-    expansion; empty list → 0 rows); None → caller NIEs. List-column / expression unwinds
-    (null/empty-element semantics) decline (NIE) for now."""
+    """Native UNWIND for two shapes; None → caller NIEs.
+
+    1. Literal scalar list (``UNWIND [1, 2] AS x``): cross-join each row with the values
+       (cypher per-row expansion; empty list → 0 rows).
+    2. Carried list column (``WITH collect(x) AS xs UNWIND xs AS y``, i.e. a ``collect()``
+       output or any List-dtype binding): explode the list column. Mirrors the pandas oracle
+       (``RowPipelineMixin.unwind`` list-column branch) exactly — an empty-list or null cell
+       contributes 0 rows; nulls WITHIN a list survive as real elements; the source column is
+       retained and the exploded values are appended as ``as_``.
+
+    Everything else — nested-list literals, scalar/non-list columns (whose single-element-list
+    Cypher coercion is not yet ported), function/arithmetic results — still declines (NIE)
+    rather than risk diverging from pandas."""
     import polars as pl
-    from graphistry.compute.gfql.expr_parser import ListLiteral, Literal
+    from graphistry.compute.gfql.expr_parser import Identifier, ListLiteral, Literal
 
     if not isinstance(expr, str):
         return None
@@ -846,14 +856,29 @@ def unwind_polars(g: Plottable, expr: str, as_: str = "value") -> Optional[Plott
         node = parse(expr)
     except Exception:
         return None
-    if not isinstance(node, ListLiteral) or not all(isinstance(it, Literal) for it in node.items):
-        return None
     table = _active_table(g)
     if as_ in table.columns:
         return None
-    values = [it.value for it in node.items if isinstance(it, Literal)]
-    rhs = pl.DataFrame({as_: values})
-    return _rewrap(g, table.join(rhs, how="cross"))
+    if isinstance(node, ListLiteral) and all(isinstance(it, Literal) for it in node.items):
+        values = [it.value for it in node.items if isinstance(it, Literal)]
+        rhs = pl.DataFrame({as_: values})
+        return _rewrap(g, table.join(rhs, how="cross"))
+    if isinstance(node, Identifier) and node.name in table.columns:
+        col = node.name
+        if not isinstance(table.schema[col], pl.List):
+            # Non-list column: Cypher UNWIND coerces a scalar to a 1-element list (null → 0
+            # rows). Those semantics aren't ported here yet, so decline rather than diverge.
+            return None
+        # pandas oracle: empty/null list cells drop out (0 rows); nulls within a list survive.
+        # Copy the source into ``as_`` first (keeping the source column, like pandas), filter
+        # out empty/null cells (``list.len()`` is null for a null cell → excluded by the
+        # predicate), then explode. Pre-filtering empties also makes the explode independent of
+        # the polars ``empty_as_null`` default (stable across polars versions).
+        out = table.with_columns(pl.col(col).alias(as_))
+        out = out.filter(pl.col(as_).list.len() > 0)
+        out = out.explode(as_)
+        return _rewrap(g, out)
+    return None
 
 
 def select_extend_polars(g: Plottable, items: Sequence[SelectItem]) -> Optional[Plottable]:

graphistry/tests/compute/gfql/test_engine_polars_row_pipeline.py

@@ -139,6 +139,100 @@ def test_polars_row_pipeline_deferred_raises(query):
         BASE.gfql(query, engine="polars")
 
 
+# Native polars UNWIND of a CARRIED list column (collect() output / carried binding), the row-
+# pipeline analogue of the IC6 `WITH collect(...) AS xs UNWIND xs AS y` shape. These are literal-
+# seeded (a MATCH-introduced alias before UNWIND-after-WITH is parser-gated for BOTH engines), so
+# the graph is irrelevant; the collect() result is what the UNWIND explodes. Parity gate: pandas.
+COLLECT_UNWIND = [
+    "UNWIND [1, 2, 3] AS x WITH collect(x) AS xs UNWIND xs AS y RETURN y",
+    "UNWIND [3, 1, 2, 1] AS x WITH collect(DISTINCT x) AS xs UNWIND xs AS y RETURN y",
+    "UNWIND [5, 1, 9, 2] AS x WITH collect(x) AS xs UNWIND xs AS y RETURN y ORDER BY y DESC",
+    "UNWIND [1, 2, 3, 4] AS x WITH x % 2 AS k, collect(x) AS xs UNWIND xs AS y RETURN k, y ORDER BY k, y",
+    "UNWIND [1, 2, 3, 4] AS x WITH collect(x) AS xs UNWIND xs AS y RETURN count(y) AS c, sum(y) AS s",
+    "UNWIND [1, 2, 3, 4] AS x WITH x % 2 AS k, collect(x) AS xs UNWIND xs AS y RETURN k, count(y) AS c ORDER BY k",
+    # nested pipeline: collect -> unwind -> collect -> unwind
+    "UNWIND [1, 1, 2, 2, 3] AS x WITH collect(DISTINCT x) AS xs UNWIND xs AS y "
+    "WITH collect(y * 10) AS ys UNWIND ys AS z RETURN z ORDER BY z",
+]
+
+
+@pytest.mark.parametrize("query", COLLECT_UNWIND)
+def test_polars_collect_unwind_parity(query):
+    """collect() -> UNWIND (list-column explode) matches the pandas oracle exactly."""
+    _assert_parity(query, order_sensitive="ORDER BY" in query)
+
+
+@pytest.mark.parametrize("query", COLLECT_UNWIND)
+def test_polars_collect_unwind_is_polars_typed(query):
+    """collect() -> UNWIND stays native (polars-typed, no pandas round-trip)."""
+    assert "polars" in type(BASE.gfql(query, engine="polars")._nodes).__module__
+
+
+def test_polars_unwind_list_column_semantics_unit():
+    """unwind_polars list-column branch == pandas oracle on empty/null/nested cells.
+
+    These cells (empty list, null cell, null WITHIN a list) can't be produced through Cypher
+    collect() — which strips nulls and yields ``[]`` for empty groups — so exercise them by
+    building the list column directly and comparing the native explode to the pandas oracle
+    (``RowPipelineMixin.unwind``)."""
+    import pandas as _pd
+    import polars as _pl
+    from graphistry.compute.gfql.lazy.engine.polars.row_pipeline import unwind_polars
+    from graphistry.compute.gfql.row.pipeline import RowPipelineMixin
+
+    rows = {"k": [0, 1, 2, 3, 4], "xs": [[10, 20], [], [30, None, 40], None, [50]]}
+
+    class _Oracle(RowPipelineMixin):
+        def __init__(self, df):
+            self._df = df
+
+        def _gfql_get_active_table(self):
+            return self._df
+
+        def _gfql_row_table(self, df):
+            return df
+
+    oracle = _Oracle(_pd.DataFrame(rows)).unwind("xs", as_="v").reset_index(drop=True)
+
+    g = graphistry.nodes(_pd.DataFrame({"id": [0]}), "id").bind()
+    g._nodes = _pl.DataFrame(rows)
+    native = unwind_polars(g, "xs", "v")
+    assert "polars" in type(native._nodes).__module__
+    got = native._nodes.to_pandas().reset_index(drop=True)
+
+    # empty list + null cell drop out (0 rows); nulls WITHIN a list survive as real elements.
+    assert list(got.columns) == list(oracle.columns)
+    assert len(got) == len(oracle) == 6
+
+    def _canon(df):
+        # normalize NaN/None null representation (polars -> NaN, pandas oracle -> None) so the
+        # comparison is null-representation agnostic; keep the surviving in-list null as a row.
+        # Compare only the meaningful key/exploded columns (the retained ``xs`` list column
+        # carries an in-list null whose NaN/None rendering differs harmlessly per engine).
+        out = df.sort_values(["k", "v"], na_position="last").reset_index(drop=True)[["k", "v"]]
+        out["v"] = [None if pd.isna(x) else int(x) for x in out["v"]]
+        return out
+
+    pd.testing.assert_frame_equal(_canon(got), _canon(oracle), check_dtype=False)
+
+
+def test_polars_unwind_declines_non_list_and_collisions():
+    """Non-list column / name collision / unknown identifier UNWIND declines (None -> NIE),
+    never a wrong-shape explode."""
+    import pandas as _pd
+    import polars as _pl
+    from graphistry.compute.gfql.lazy.engine.polars.row_pipeline import unwind_polars
+
+    g = graphistry.nodes(_pd.DataFrame({"id": [0]}), "id").bind()
+    g._nodes = _pl.DataFrame({"a": [1, 2, 3], "xs": [[1], [2], [3]]})
+    assert unwind_polars(g, "a", "v") is None          # scalar (non-list) column -> decline
+    assert unwind_polars(g, "xs", "a") is None          # as_ collides with existing column
+    assert unwind_polars(g, "nope", "v") is None        # unknown identifier
+    # nested-list literal is still declined (only scalar-literal lists lowered natively)
+    with pytest.raises(NotImplementedError):
+        BASE.gfql("UNWIND [[1, 2], [3, 4]] AS pair UNWIND pair AS z RETURN z", engine="polars")
+
+
 def test_row_expr_lowering_unit():
     """lower_expr_str / lower_select_items / lower_order_by_keys edge cases."""
     from graphistry.compute.gfql.lazy.engine.polars.row_pipeline import (