fix(gfql): fixes found by the route sweeps: source dtypes on every route, polars prune decline, polars-gpu fused lane (#2058, #2053, #2064)

CHANGELOG.md

@@ -25,6 +25,8 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 * GFQL: a multi-hop step (`hops=2`, `to_fixed_point`) whose edge alias shares its name with the column its own filter uses raised `incompatible-column-type` on pandas and cuDF; the backward re-execution now reads the graph's edge columns for that step, so the eager engines serve it like the single-hop form and return the same rows as polars (#2049).
 * GFQL: one alias-shadowing contract on every engine for op-list chains: an alias named like a column of the frame it marks becomes the boolean marker under that name (pandas, cuDF and now polars, which used to keep the user values and leak a `_right` join suffix). The shadowed values ride under the internal restore column the Cypher row pipeline already resolves, so Cypher keeps reading the user value through the variable (`MATCH (a)-[type:K]->(b) RETURN type.type`) on both engines; the polars residual decline for that shape is gone.
 * GFQL polars: an unnamed, untyped single-hop chain kept duplicate node rows (a node table carrying the same id twice) where pandas, cuDF and every other polars shape collapse them; the plain single-hop branch now returns one row per node id (#2051).
+* GFQL polars-gpu: the fused single-hop grouped-aggregate lane no longer takes the `value_counts` formulation on the GPU target; its `unnest` node has no cudf-polars implementation, so every fused `count(*)` shape declined to the generic route on `engine='polars-gpu'`. The `group_by` formulation is GPU-executable and answers the same rows (#2064).
+
 - **cuDF 26.2 compatibility: `cudf.from_pandas` replaces the removed `cudf.DataFrame.from_pandas` at the five cuDF-only product sites (`ai_utils`, `umap_utils`, `feature_utils`) and in the test fixtures (#2043)**; the three cuDF chain differential cases that disagree with the full path on cuDF 26.2 are marked expected-failure on that line with the tracking issue, so the GPU lane reports them instead of crashing before them.
 - **GFQL pandas/cuDF: several single-alias `IN` (and other pushed-down) predicates across a hop no longer raise `Unalignable boolean Series` (#2020)**: the predicate pushdown filtered an alias frame by label after an earlier pushdown had already narrowed it, while the mask it evaluated carried a fresh positional index. Rows are now kept by position, which is the contract of a mask computed on the same rows; results equal the polars engine and the scalar `=` form.
 - **GFQL polars: a native chain whose edge alias shares its name with the column that step filters on is served instead of raising `incompatible-column-type` (#2039)**: the backward pass and the pruned re-execution now run each step on the graph's original edge columns, and a node step whose alias names its own filtered column filters the graph's node values rather than the marker of an earlier pass, so a stamped alias marker is never re-filtered as that column; pandas and polars return the same rows. A cross-engine collision matrix pins the single-hop shapes and records the multi-hop (#2049) and binding-column (#2050) forms as expected failures. The Cypher rows-route projection of such an alias on polars still declines with a typed error (pinned) and stays tracked.
@@ -41,6 +43,8 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 
 ### Changed
 
+* GFQL polars: `prune_to_endpoints` on a single-hop edge is a typed decline (`NotImplementedError`, use `engine='pandas'` or `'cudf'`); the plain single-hop branch and the full polars traversal used to admit it and silently return both endpoints where pandas keeps the arrival side. Variable-length hops keep their native pruning (#2053).
+* GFQL: the general chain path keeps integer and boolean node attribute dtypes on a closed graph: the endpoint-closure backfill now appends only endpoints missing from the node frame instead of concatenating every endpoint id and deduplicating afterwards, which widened the attribute columns to float on every route that fell through the hot paths (#2058, chain seam; the row-pipeline pivot dtypes are a separate item).
 * GFQL: the chain specializations move into `graphistry/compute/chain_specializations/{admission,hotpaths}.py` (pandas/cuDF single-node lane, seeded typed single hop, seeded typed RETURN-destination) and `graphistry/compute/gfql/lazy/engine/polars/chain_specializations/{admission,hotpaths}.py` (polars plain single-hop branches, seeded lane, RETURN-destination), each lane next to the admission predicate the dispatcher calls (`native_fast_path_admits`, `polars_plain_single_hop_admits`, `polars_seeded_lane_admits`); `chain.py` and the polars chain only dispatch, `chain_fast_paths.py` keeps the shared seed/index helpers. No route admits or declines anything it did not before. Tests mirror the new paths and filter one shared shape corpus per route with the route's own gate; `GFQL_ROUTES_OFF=<route,...>` (test conftest) makes named hot paths decline so every existing test replays through the other routes, and `bin/test-routes-off.sh` reports the per-route divergences.
 * GFQL: the wavefront seed-rediscovery rule moved out of `hop.py` into `graphistry/compute/gfql/seed_rediscovery.py` (pandas/cuDF) and `graphistry/compute/gfql/lazy/engine/polars/seed_rediscovery.py` (polars); `undirected_rediscovered_seed_ids` (an internal helper) is gone.
 

graphistry/compute/chain.py

@@ -1351,7 +1351,11 @@ def _chain_impl(
                     sort=False,
                 ).drop_duplicates(subset=[g_out._node])
                 endpoints = align_shared_column_dtypes(g_out._nodes, endpoints)
-                g_out = g_out.nodes(safe_row_concat([g_out._nodes, endpoints], ignore_index=True, sort=False).drop_duplicates(subset=[g_out._node]))
+                missing = endpoints[~endpoints[g_out._node].isin(g_out._nodes[g_out._node])]
+                nodes_out = g_out._nodes
+                if len(missing) > 0:  # only a dangling endpoint is backfilled; a present id would widen the attribute dtypes
+                    nodes_out = safe_row_concat([nodes_out, missing], ignore_index=True, sort=False)
+                g_out = g_out.nodes(nodes_out.drop_duplicates(subset=[g_out._node]))
 
             success = True
 

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

@@ -942,6 +942,11 @@ def _chain_traversal_polars(self: Plottable, ops, start_nodes: Optional[Any] = N
     if isinstance(ops[-1], ASTEdge):
         ops = ops + [ASTNode()]
 
+    if any(isinstance(op, ASTEdge) and op.prune_to_endpoints and op.is_simple_single_hop() for op in ops):
+        raise NotImplementedError(
+            "polars chain engine: prune_to_endpoints on a single-hop edge (arrival-side pruning "
+            "by hop label) is not implemented; use engine='pandas' or engine='cudf'"
+        )
     if any(
         isinstance(op, ASTEdge) and not op.is_simple_single_hop() and not _is_native_multihop(op)
         for op in ops

graphistry/compute/gfql/lazy/engine/polars/chain_specializations/admission.py

@@ -18,7 +18,7 @@ def _plain_edge(op: ASTObject) -> bool:
             and op.edge_match is None and op.source_node_match is None
             and op.destination_node_match is None and op._name is None
             and op.source_node_query is None and op.destination_node_query is None
-            and op.edge_query is None and not op.include_zero_hop_seed)
+            and op.edge_query is None and not op.include_zero_hop_seed and not op.prune_to_endpoints)
 
 
 def polars_plain_single_hop_admits(ops: Sequence[ASTObject], start_nodes: Optional[object]) -> Optional[PolarsPlainSingleHopShape]:

graphistry/compute/gfql/row/pipeline.py

@@ -3675,6 +3675,25 @@ def _gfql_unshadow_alias_marker_column(
             base_frame[[key_col, alias]], on=key_col, how="left"
         )
 
+    @staticmethod
+    def _gfql_restore_alias_property_dtypes(bindings: "DataFrameT", base_nodes: Optional["DataFrameT"], alias: str) -> "DataFrameT":
+        """A traversal frame carries id-only stub rows, so its property columns arrive widened; once the bound rows hold no null, the node table's dtype is the answer's dtype."""
+        if base_nodes is None:
+            return bindings
+        for prop in base_nodes.columns:
+            col = f"{alias}.{prop}"
+            if col not in bindings.columns:
+                continue
+            want, have = str(base_nodes[prop].dtype), str(bindings[col].dtype)
+            if want == have or have not in ("float64", "float32", "object"):
+                continue
+            if want not in ("int8", "int16", "int32", "int64", "uint8", "uint16", "uint32", "uint64", "bool"):
+                continue
+            if bool(bindings[col].isna().any()):
+                continue
+            bindings[col] = bindings[col].astype(want)
+        return bindings
+
     @staticmethod
     def _gfql_node_alias_lookup_frame(lookup_source: Any, node_id: str, alias: str) -> Any:
         lookup = lookup_source[[node_id]].copy()
@@ -4371,33 +4390,6 @@ def _gfql_add_missing_binding_columns(
                     bindings[col] = source.iloc[0:0].reindex(bindings.index)
                 else:
                     bindings[col] = self._gfql_broadcast_scalar(bindings, None)
-        # Downstream node aliases (every node op after the first) reach the row via a hop
-        # whose unmatched rows introduce NaN, so the non-empty path widens the columns that
-        # cannot hold NaN: numpy int -> float64, numpy bool -> object. The 0-row path sourced
-        # them from the base node frame (int/bool), so match that widening or an emptied
-        # `sum(b.i)`/`max(b.bv)` returns int64/bool where the non-empty run and master give
-        # float64/object -- observable through UNION ALL (#31, Wave 37 Finding 2). Extension
-        # dtypes (`Int64`, `boolean`) hold NA natively and stay put in the non-empty path, so
-        # they must NOT be touched here (widening them re-introduced the divergence -- Wave 37
-        # Finding 1).
-        import pandas as _pd
-
-        node_ops = [op for op in ops if isinstance(op, ASTNode)]
-        for op in node_ops[1:]:
-            alias = getattr(op, "_name", None)
-            if not isinstance(alias, str):
-                continue
-            for col in [c for c in bindings.columns if c == alias or str(c).startswith(f"{alias}.")]:
-                try:
-                    dtype = bindings[col].dtype
-                    if _pd.api.types.is_extension_array_dtype(dtype):
-                        continue
-                    if dtype.kind in ("i", "u"):
-                        bindings[col] = bindings[col].astype("float64")
-                    elif dtype.kind == "b":
-                        bindings[col] = bindings[col].astype("object")
-                except Exception:
-                    continue
         return bindings
 
     def _gfql_connected_bindings_row_frame_from_state(
@@ -4453,6 +4445,7 @@ def _gfql_connected_bindings_row_frame_from_state(
             dup_col = f"{node_id}__{alias}_join__"
             if dup_col in bindings.columns:
                 bindings = bindings.drop(columns=[dup_col])
+            bindings = self._gfql_restore_alias_property_dtypes(bindings, base_nodes, alias)
             for hop_col in [col for col in bindings.columns if is_shortest_path_hops_column(str(col))]:
                 alias_hop_col = f"{alias}.{hop_col}"
                 if alias_hop_col in bindings.columns:

graphistry/compute/gfql_fast_paths.py

@@ -1993,6 +1993,9 @@ def _low_cardinality_pure_count_plan(
         return None
     if edge_rows > _LOWCARD_COUNT_MAX_INPUT_ROWS:
         return None
+    from graphistry.compute.gfql.lazy import ExecutionTarget, active_target
+    if active_target() == ExecutionTarget.GPU:
+        return None  # cudf-polars has no unnest map function; the group_by formulation is GPU-executable
 
     # ``name=`` (polars >= 1.0, and the declared floor is 1.29) keeps the count column out
     # of a rename, so a group key literally named ``count`` is served rather than crashing.
@@ -3336,26 +3339,22 @@ def _pandas_frame_has_extension_dtype(frame: DataFrameT) -> bool:
 def _pivot_parity_casts(
     rows: DataFrameT, items: Sequence[Tuple[str, str]], node: str, *, keep_source_dtypes: bool,
 ) -> Optional[Dict[str, str]]:
-    """Per-output casts that reproduce the pandas rows-pivot dtypes (int/float -> float64,
-    bool -> object) for a lean projection; None for a dtype class the pivot parity does
-    not cover (datetimes, extension dtypes), which must take the full path."""
+    """The lean projection's dtype gate: source dtypes are kept on every route, so no
+    cast is produced; None for a dtype class the projection does not cover (datetimes,
+    extension dtypes), which must take the full path."""
     import numpy as np
     casts: Dict[str, str] = {}
     for out_name, prop in items:
-        if prop == node and len(rows) > 0:
-            continue  # the pivot keeps the id column's dtype except on an empty frame
+        if prop == node:
+            continue
         d = rows[prop].dtype
         if isinstance(d, pd.StringDtype):
             continue
         if not isinstance(d, np.dtype):
             return None
-        if d == np.dtype(bool):
-            if not keep_source_dtypes:
-                casts[out_name] = "object"
-        elif d.kind in "iuf":
-            if not keep_source_dtypes:
-                casts[out_name] = "float64"
-        elif d.kind != "O":
+        if d == np.dtype(bool) or d.kind in "iuf":
+            continue
+        if d.kind != "O":
             return None
     return casts
 
@@ -3802,8 +3801,7 @@ def _execute_seeded_typed_hop_fast_path(
         hop_details=[{"hop": 1}] if index_ctx is not None else None,
     )
     p_rows, _edges, seed_rows, kernel_admits = dst_res
-    # the canonical bag path keeps source dtypes only where the indexed kernel serves it
-    canonical_keeps_source_dtypes = "cudf" in type(p_rows).__module__ or (bag_rows and kernel_admits)
+    canonical_keeps_source_dtypes = True  # the general chain path keeps source dtypes on every route
     if wants_multi_alias:
         assert select_items is not None
         out_frame = _seeded_typed_hop_two_alias_frame(

graphistry/tests/compute/gfql/cypher/test_grouped_aggregate_fused_polars.py

@@ -1027,3 +1027,24 @@ def test_grouped_aggregate_projection_differential_all_shapes_all_engines(
 
     result = _records(_graph(engine, nodes, edges).gfql(query, engine=engine))
     assert result == oracle, f"{label}: projection changed the answer on {engine}"
+
+
+def test_low_cardinality_count_plan_declines_on_the_gpu_target() -> None:
+    """The value_counts formulation lowers to an ``unnest`` node cudf-polars cannot execute,
+    so on the GPU target the fused lane keeps the group_by formulation; on CPU the
+    value_counts plan is still taken for the same input."""
+    pl = _require_polars()
+    from graphistry.compute.gfql.lazy import ExecutionTarget, target_mode
+    work = pl.LazyFrame({"id": [1, 2, 3], "city": ["LA", "NY", "LA"]})
+    kwargs = dict(
+        node_col="id", group_keys=["city"], agg_specs=[("n", "count", None)],
+        needed_by_alias={"c": [("city", "city")]},
+        frames_by_alias={"c": pl.DataFrame({"id": [1, 2, 3], "city": ["LA", "NY", "LA"]})},
+        edge_rows=3,
+    )
+    with target_mode(ExecutionTarget.CPU):
+        cpu_plan = gfql_fast_paths_module._low_cardinality_pure_count_plan(work, **kwargs)
+    with target_mode(ExecutionTarget.GPU):
+        gpu_plan = gfql_fast_paths_module._low_cardinality_pure_count_plan(work, **kwargs)
+    assert cpu_plan is not None and "UNNEST" in cpu_plan.explain()
+    assert gpu_plan is None

graphistry/tests/compute/gfql/cypher/test_lowering.py

@@ -15829,12 +15829,10 @@ def test_connected_join_empty_edge_aggregate_keeps_numeric_dtype(ret: str, dtype
 @pytest.mark.parametrize(
     "ret,dtype",
     [
-        # A downstream (non-anchor) node reaches the row via a hop whose NaN widens its
-        # integer columns to float in the non-empty run; the 0-row path must match, or an
-        # emptied sum(b.iv) returns int64 and escapes via UNION ALL (#31). The anchor never
-        # NaN-widens, so it stays int; float columns are unchanged.
-        ("sum(b.iv) AS c", "float64"),
-        ("max(b.iv) AS c", "float64"),
+        # the 0-row path must carry the non-empty run's dtypes (UNION ALL, #31): source
+        # dtypes on every alias since the traversal stubs stopped widening them (#2058)
+        ("sum(b.iv) AS c", "int64"),
+        ("max(b.iv) AS c", "int64"),
         ("sum(a.iv) AS c", "int64"),
         ("sum(b.fv) AS c", "float64"),
     ],
@@ -15890,7 +15888,7 @@ def test_connected_join_empty_node_aggregate_keeps_nullable_int(ret: str, dtype:
         # numpy bool cannot hold the hop's left-join NaN, so the non-empty path widens a
         # downstream node's bool column to object; the 0-row path must match, or an emptied
         # max(b.bv) returns raw bool where non-empty and master give object (Wave 37 Finding 2).
-        ("bool", "object"),
+        ("bool", "bool"),
         # Nullable `boolean` holds NA natively and stays `boolean` in the non-empty path, so it
         # must be left untouched -- widening it would be the Finding-1 over-reach in bool form.
         ("boolean", "boolean"),

graphistry/tests/compute/gfql/lazy/engine/polars/chain_specializations/test_polars_admission.py

@@ -25,7 +25,7 @@
     "plain single hop, seeded, destination filter": "skip-combine",
     "plain single hop, undirected, unconstrained": "skip-combine",
     "plain single hop, undirected, seeded": None,
-    "single hop, prune to endpoints": "seeded-index",
+    "single hop, prune to endpoints": None,
 }
 
 
@@ -51,8 +51,6 @@ def _sig(res):
 
 @pytest.mark.parametrize("name", [k for k, v in EXPECTED.items() if v is not None])
 def test_admitted_shapes_match_the_pandas_full_path(name, request):
-    if name == "single hop, prune to endpoints":
-        request.applymarker(pytest.mark.xfail(strict=True, reason="graphistry/pygraphistry#2053"))
     ops = by_name()[name].ops()
     g_pd = graphistry.nodes(NODES, "key").edges(EDGES, "s", "d", "eid")
     g_pl = graphistry.nodes(pl.from_pandas(NODES), "key").edges(pl.from_pandas(EDGES), "s", "d", "eid")
@@ -111,3 +109,21 @@ def test_seeded_lane_called_directly_serves_every_admitted_non_colliding_shape(n
     ops = by_name()[name].ops()
     res = hot._try_seeded_chain_polars(_indexed_polars_graph(), ops)
     assert (res is not None) == (name in SEEDED_LANE_SERVES_DIRECTLY)
+
+
+@pytest.mark.parametrize("ops_name", ["single hop, prune to endpoints"])
+def test_prune_to_endpoints_is_a_typed_decline_on_polars_and_served_on_pandas(ops_name):
+    """A single-hop edge has no hop labels to prune by on polars, so prune_to_endpoints there
+    raises the engine's NotImplementedError (variable-length hops keep their native pruning);
+    pandas answers the shape."""
+    ops = by_name()[ops_name].ops()
+    g_pl = graphistry.nodes(pl.from_pandas(NODES), "key").edges(pl.from_pandas(EDGES), "s", "d", "eid")
+    with pytest.raises(NotImplementedError, match="prune_to_endpoints"):
+        g_pl.gfql(ops, engine="polars")
+    unseeded = [n(), e_forward(prune_to_endpoints=True), n()]
+    with pytest.raises(NotImplementedError, match="prune_to_endpoints"):
+        g_pl.gfql(unseeded, engine="polars")
+    g_pd = graphistry.nodes(NODES, "key").edges(EDGES, "s", "d", "eid")
+    assert _sig(g_pd.gfql(ops, engine="pandas"))[0] == [2, 3]
+    assert polars_plain_single_hop_admits(ops, None) is None
+    assert polars_plain_single_hop_admits([n({"key": 1}), e_forward(), n()], None) == "seeded-index"