fix(gfql): preserve scalar reentry across continuation matches

CHANGELOG.md

@@ -10,6 +10,8 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 
 ### Fixed
 
+- **GFQL Cypher scalar carries after `WITH`**: a scalar carried through `WITH` remains available when a later `MATCH` reconnects the earlier alias, including whole-node-plus-scalar, connected-scalar, and multi-edge path prefixes. Grouping now keeps each carried scalar's multiplicity on pandas and Polars instead of returning null groups or dropping a connected result group (#2252).
+
 - **GFQL Cypher conversion operand validation**: statically invalid conversion operands, such as lists, maps, and graph entities, are rejected at compile time before the engine sees them. Dynamic operands, null propagation, and list-comprehension lexical shadowing keep their existing behavior.
 
 - **GFQL remote Let/DAG compatibility**: `gfql_remote` no longer substitutes an empty `gfql_operations` array for a Let/DAG query, so unsupported older servers receive `gfql_query` and reject it explicitly instead of silently running an identity program.

graphistry/compute/chain.py

@@ -279,6 +279,7 @@ def combine_steps(
             steps = new_steps
 
     logger.debug('-----------[ combine %s ---------------]', kind)
+    from graphistry.compute.gfql.cypher.reentry.naming import REENTRY_HIDDEN_COLUMN_PREFIX
 
     if label_steps is None:
         label_steps = steps
@@ -307,6 +308,7 @@ def apply_output_slice(op: ASTObject, op_label: ASTObject, df):
     dfs_to_concat = []
     extra_step_dfs = []
     base_cols = set((g._nodes if kind == "nodes" else g._edges).columns)
+    carry_cols: List[str] = []
     for idx, (op, g_step) in enumerate(steps):
         op_label = label_steps[idx][0] if idx < len(label_steps) else op
         step_df = apply_output_slice(op, op_label, (g_step._nodes if kind == "nodes" else g_step._edges))
@@ -315,13 +317,67 @@ def apply_output_slice(op: ASTObject, op_label: ASTObject, df):
             raise ValueError(f"Column '{id}' not found in {kind} step DataFrame. "
                            f"Step has id='{step_id}', available columns: {list(step_df.columns)}. "
                            f"Operation: {op}")
-        dfs_to_concat.append(step_df[[id]])
+
+        if (
+            kind == 'nodes'
+            and idx > 0
+            and isinstance(op, ASTNode)
+            and isinstance(label_steps[idx - 1][0], ASTEdge)
+        ):
+            prev_nodes = label_steps[idx - 2][1]._nodes
+            prev_edges = label_steps[idx - 1][1]._edges
+            edge_op = label_steps[idx - 1][0]
+            assert isinstance(edge_op, ASTEdge)
+            prev_carry_cols = (
+                [c for c in prev_nodes.columns if c.startswith(REENTRY_HIDDEN_COLUMN_PREFIX)]
+                if prev_nodes is not None else []
+            )
+            if (
+                prev_nodes is not None
+                and prev_edges is not None
+                and prev_carry_cols
+                and len(step_df)
+                and len(prev_edges)
+            ):
+                edge_parts = []
+                if edge_op.direction != 'reverse':
+                    edge_parts.append((g._source, g._destination))
+                if edge_op.direction != 'forward':
+                    edge_parts.append((g._destination, g._source))
+
+                carry_maps = []
+                for from_col, to_col in edge_parts:
+                    carry_maps.append(
+                        prev_edges[[from_col, to_col]].merge(
+                            prev_nodes[[id] + prev_carry_cols],
+                            left_on=from_col,
+                            right_on=id,
+                            how='inner',
+                        ).drop(columns=[from_col, id]).rename(columns={to_col: id})
+                    )
+                carry_map = df_concat(engine)(carry_maps)
+                try:
+                    carry_map = carry_map.drop_duplicates(subset=[id] + prev_carry_cols)
+                except TypeError:
+                    carry_map = carry_map.drop_duplicates(subset=[id])
+                step_df = step_df.drop(columns=[c for c in prev_carry_cols if c in step_df.columns])
+                step_df = safe_merge(step_df, carry_map, on=id, how='left', engine=engine)
+
+        step_carry_cols = [
+            column for column in step_df.columns
+            if column.startswith(REENTRY_HIDDEN_COLUMN_PREFIX)
+        ]
+        carry_cols.extend(step_carry_cols)
+        dfs_to_concat.append(step_df[[id] + step_carry_cols])
 
     for _, (_, g_step) in enumerate(label_steps):
         step_df = (g_step._nodes if kind == "nodes" else g_step._edges)
         if id not in step_df.columns:
             continue
-        extra_cols = [c for c in step_df.columns if c != id and c not in base_cols and 'hop' in c]
+        extra_cols = [
+            c for c in step_df.columns
+            if c != id and c not in base_cols and 'hop' in c
+        ]
         if extra_cols:
             extra_step_dfs.append(step_df[[id] + extra_cols])
 
@@ -332,7 +388,12 @@ def apply_output_slice(op: ASTObject, op_label: ASTObject, df):
             dfs_to_concat = [df_to_engine(df, engine) for df in dfs_to_concat]
 
     concat = df_concat(engine)
-    out_df = concat(dfs_to_concat).drop_duplicates(subset=[id])
+    dedup_cols = [id] + list(dict.fromkeys(carry_cols))
+    combined_df = concat(dfs_to_concat)
+    try:
+        out_df = combined_df.drop_duplicates(subset=dedup_cols)
+    except TypeError:
+        out_df = combined_df.drop_duplicates(subset=[id])
 
     label_cols = set()
     for step_df in extra_step_dfs:
@@ -676,7 +737,6 @@ def _plan_indexed_middle(
     comparable instead of two inline condition chains that drift.
     """
     from .gfql.index.handoff import IndexedBindingsHandoff
-
     if not (middle and suffix) or prefix or start_nodes is not None or policy:
         return None
     call = suffix[0]
@@ -1505,6 +1565,9 @@ def _chain_impl(
 
             if g_out._edges is not None and len(g_out._edges) > 0:
                 concat_fn = df_concat(engine_concrete)
+                from graphistry.compute.gfql.cypher.reentry.naming import (
+                    REENTRY_HIDDEN_COLUMN_PREFIX,
+                )
                 endpoints = concat_fn(
                     [
                         g_out._edges[[g_out._source]].rename(columns={g_out._source: g_out._node}),
@@ -1518,7 +1581,15 @@ def _chain_impl(
                 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]))
+                node_dedup_cols = [g_out._node] + [
+                    column for column in nodes_out.columns
+                    if column.startswith(REENTRY_HIDDEN_COLUMN_PREFIX)
+                ]
+                try:
+                    nodes_out = nodes_out.drop_duplicates(subset=node_dedup_cols)
+                except TypeError:
+                    nodes_out = nodes_out.drop_duplicates(subset=[g_out._node])
+                g_out = g_out.nodes(nodes_out)
 
             success = True
 

graphistry/compute/gfql/cypher/reentry/execution.py

@@ -641,7 +641,7 @@ def compiled_query_scalar_reentry_state(
             return base_graph, None
         return _bind_reentry_graph(
             base_graph,
-            cast(DataFrameT, base_nodes.iloc[0:0]),
+            cast(DataFrameT, base_nodes.head(0) if _is_polars_df(base_nodes) else base_nodes.iloc[0:0]),
             empty_edges=True,
         ), None
     if base_nodes is None:

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

@@ -1200,4 +1200,40 @@ def _chain_traversal_polars(self: Plottable, ops, start_nodes: Optional[Any] = N
     # error, and silencing it would cost the lazy/eager distinction the shim exists to keep.
     final_edges_eager, final_nodes_eager = collect_all([final_edges, final_nodes])
     final_edges_eager = _restore_edge_dtypes(final_edges_eager, src, dst, _endpoint_restore)
+    seed_carry_cols = [
+        col for col in (start_nodes.columns if start_nodes is not None else [])
+        if isinstance(col, str) and col.startswith("__cypher_reentry_")
+    ]
+    if seed_carry_cols:
+        carry_map = start_nodes.select([node_col, *seed_carry_cols])
+        for edge_op, edge_step in edge_steps:
+            if (
+                not isinstance(edge_op, ASTEdge)
+                or edge_step._edges is None
+                or edge_step._source is None
+                or edge_step._destination is None
+            ):
+                continue
+            edge_src, edge_dst = edge_step._source, edge_step._destination
+            edge_parts = []
+            if edge_op.direction != "reverse":
+                edge_parts.append((edge_src, edge_dst))
+            if edge_op.direction != "forward":
+                edge_parts.append((edge_dst, edge_src))
+            next_carry_frames = []
+            for from_col, to_col in edge_parts:
+                next_carry_frames.append(
+                    carry_map.join(
+                        edge_step._edges.select([from_col, to_col]),
+                        left_on=node_col, right_on=from_col, how="inner")
+                    .drop(node_col)
+                    .rename({to_col: node_col})
+                )
+            carry_map = (
+                pl.concat(next_carry_frames, how="vertical")
+                if len(next_carry_frames) > 1 else next_carry_frames[0]
+            )
+        final_nodes_eager = final_nodes_eager.drop(seed_carry_cols).join(
+            carry_map, on=node_col, how="left"
+        )
     return self.nodes(final_nodes_eager, node_col).edges(final_edges_eager, src, dst)

graphistry/compute/gfql/row/pipeline.py

@@ -1334,6 +1334,9 @@ def _gfql_eval_expr_ast_values(self, table_df: Any, node: Any) -> Tuple[bool, An
         if isinstance(node, PropertyAccessExpr):
             if isinstance(node.value, Identifier):
                 alias_name = node.value.name
+                from graphistry.compute.gfql.cypher.reentry.naming import _is_hidden_reentry_property
+                if node.property in table_df.columns and _is_hidden_reentry_property(node.property):
+                    return True, table_df[node.property]
                 if alias_name == node.property:
                     restore_col = shadow_restore_column(alias_name)
                     if restore_col in table_df.columns:
@@ -2500,6 +2503,7 @@ def _gfql_series_to_pylist(values: Any) -> List[Any]:
                 return list(values.to_arrow().to_pylist())
             except Exception:
                 pass
+
         if hasattr(values, "to_pandas"):
             try:
                 return list(values.to_pandas().tolist())

graphistry/compute/gfql_unified.py

@@ -121,7 +121,11 @@ def _slice_reentry_prefix_result_row(
     if rows_df is None:
         return prefix_result
     out = prefix_result.bind()
-    out._nodes = cast(DataFrameT, rows_df.iloc[row_index:row_index + 1].reset_index(drop=True))
+    if _is_polars_frame(rows_df):
+        row_nodes = cast(DataFrameT, rows_df.slice(row_index, 1))
+    else:
+        row_nodes = cast(DataFrameT, rows_df.iloc[row_index:row_index + 1].reset_index(drop=True))
+    out._nodes = row_nodes
     out._cypher_entity_projection_presence = entity_projection_presence_for_rows(prefix_result, [row_index])
     entry = prefix_result._cypher_entity_projection_meta.get(output_name)
     if entry is not None: