From 414a0abdf02537749d644dae29f51bd0d0d1375c Mon Sep 17 00:00:00 2001 From: Codex Worker Date: Tue, 27 Oct 2026 00:46:38 -0700 Subject: [PATCH 1/9] fix(gfql): preserve scalar reentry across continuation matches --- graphistry/compute/chain.py | 60 ++++++++++++++-- .../compute/gfql/cypher/reentry/execution.py | 2 +- graphistry/compute/gfql/row/pipeline.py | 4 ++ .../compute/gfql/cypher/test_lowering.py | 72 ++++++++++++++++++- 4 files changed, 131 insertions(+), 7 deletions(-) diff --git a/graphistry/compute/chain.py b/graphistry/compute/chain.py index 130ac44ecf..1f14a9be08 100644 --- a/graphistry/compute/chain.py +++ b/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,63 @@ 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] + 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) + carry_map = carry_map.drop_duplicates(subset=[id] + prev_carry_cols) + 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 +384,8 @@ 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)) + out_df = concat(dfs_to_concat).drop_duplicates(subset=dedup_cols) label_cols = set() for step_df in extra_step_dfs: @@ -676,7 +729,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] diff --git a/graphistry/compute/gfql/cypher/reentry/execution.py b/graphistry/compute/gfql/cypher/reentry/execution.py index 8a6b452ec3..8367d4eeca 100644 --- a/graphistry/compute/gfql/cypher/reentry/execution.py +++ b/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: diff --git a/graphistry/compute/gfql/row/pipeline.py b/graphistry/compute/gfql/row/pipeline.py index f1d68dc2c7..69db691420 100644 --- a/graphistry/compute/gfql/row/pipeline.py +++ b/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()) diff --git a/graphistry/tests/compute/gfql/cypher/test_lowering.py b/graphistry/tests/compute/gfql/cypher/test_lowering.py index f515576b02..19ab099485 100644 --- a/graphistry/tests/compute/gfql/cypher/test_lowering.py +++ b/graphistry/tests/compute/gfql/cypher/test_lowering.py @@ -22888,8 +22888,6 @@ def test_lowering_coverage_helper_direct_margin_paths() -> None: boundary_exc = GFQLValidationError(ErrorCode.E108, "forced", field="return", value=["a", "b"]) assert _is_multi_source_match_alias_boundary_error(boundary_exc, alias_targets={"a": ASTNode(name="a"), "b": ASTNode(name="b")}) - assert not _is_multi_source_match_alias_boundary_error(GFQLValidationError(ErrorCode.E201, "forced", field="return", value=["a", "b"]), alias_targets={"a": ASTNode(name="a"), "b": ASTNode(name="b")}) - assert _expr_property_access_node_aliases( "[a.city, r.weight]", parse_row_expr=_parse_row_expr, @@ -22913,3 +22911,73 @@ def test_string_cypher_parenthesized_and_preserves_missing_property_as_null() -> # IS NOT NULL on the same absent property yields no rows (absent -> null) none = g.gfql("MATCH (n) WHERE n.missing IS NOT NULL AND (n.x = 1) RETURN n.id AS id") assert none._nodes["id"].tolist() == [] if "id" in none._nodes.columns else len(none._nodes) == 0 + + +@pytest.mark.parametrize("engine", ["pandas", "polars"]) +def test_issue_2252_scalar_only_prefix_survives_match_on_pandas_and_polars(engine: str) -> None: + nodes_df, edges_df = _mk_reentry_carried_scalar_data() + nodes_df = nodes_df.assign(label__B=[False, False, True, True]) + graph = _mk_polars_graph(nodes_df, edges_df) if engine == "polars" else _mk_graph(nodes_df, edges_df) + + result = graph.gfql( + "MATCH (a:A) WITH a.num AS property MATCH (a)-->(b:B) " + "RETURN property, count(DISTINCT b.id) AS messageCount ORDER BY property", + engine=engine, + language="cypher", + ) + + records = _to_pandas_df(result._nodes).to_dict(orient="records") + assert records == [{"property": 1.0, "messageCount": 2}, {"property": 2.0, "messageCount": 2}] + + +@pytest.mark.parametrize("engine", ["pandas", "polars"]) +def test_issue_2252_scalar_only_prefix_negative_is_empty_on_pandas_and_polars(engine: str) -> None: + nodes_df, edges_df = _mk_reentry_carried_scalar_data() + nodes_df = nodes_df.assign(label__B=[False, False, True, True]) + graph = _mk_polars_graph(nodes_df, edges_df) if engine == "polars" else _mk_graph(nodes_df, edges_df) + + result = graph.gfql( + "MATCH (a:A {num: 99}) WITH a.num AS property MATCH (a)-->(b:B) " + "RETURN property, count(DISTINCT b.id) AS messageCount", + engine=engine, + language="cypher", + ) + + assert _to_pandas_df(result._nodes).to_dict(orient="records") == [] + + +def test_issue_2252_no_reentry_match_retains_multiplicity_on_pandas_and_polars() -> None: + nodes_df, edges_df = _mk_reentry_carried_scalar_data() + nodes_df = nodes_df.assign(label__B=[False, False, True, True]) + + for engine in ("pandas", "polars"): + graph = _mk_polars_graph(nodes_df, edges_df) if engine == "polars" else _mk_graph(nodes_df, edges_df) + result = graph.gfql( + "MATCH (a:A)-[:R]->(b:B) RETURN count(*) AS pathCount", + engine=engine, + language="cypher", + ) + assert _to_pandas_df(result._nodes).to_dict(orient="records") == [{"pathCount": 2}] + + +@pytest.mark.parametrize("engine", ["pandas", "polars"]) +def test_issue_2252_typed_carried_scalar_survives_match_on_pandas_and_polars(engine: str) -> None: + nodes_df, edges_df = _mk_reentry_carried_scalar_data() + nodes_df = nodes_df.assign( + label__B=[False, False, True, True], + ts=pd.to_datetime(["2020-01-01", "2020-01-02", "2020-01-03", "2020-01-04"]), + ) + graph = _mk_polars_graph(nodes_df, edges_df) if engine == "polars" else _mk_graph(nodes_df, edges_df) + + result = graph.gfql( + "MATCH (a:A) WITH a.ts AS event MATCH (a)-->(b:B) " + "RETURN event, count(DISTINCT b.id) AS messageCount", + engine=engine, + language="cypher", + ) + + records = _to_pandas_df(result._nodes).to_dict(orient="records") + assert records == [ + {"event": pd.Timestamp("2020-01-01"), "messageCount": 2}, + {"event": pd.Timestamp("2020-01-02"), "messageCount": 2}, + ] From 73764ce426d68a025adc5eb84e606d1cacf385a2 Mon Sep 17 00:00:00 2001 From: Codex Worker Date: Tue, 27 Oct 2026 00:49:50 -0700 Subject: [PATCH 2/9] fix(gfql): type reentry edge operation --- graphistry/compute/chain.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/graphistry/compute/chain.py b/graphistry/compute/chain.py index 1f14a9be08..fd62fb492e 100644 --- a/graphistry/compute/chain.py +++ b/graphistry/compute/chain.py @@ -326,7 +326,7 @@ def apply_output_slice(op: ASTObject, op_label: ASTObject, df): ): prev_nodes = label_steps[idx - 2][1]._nodes prev_edges = label_steps[idx - 1][1]._edges - edge_op = label_steps[idx - 1][0] + edge_op = cast(ASTEdge, label_steps[idx - 1][0]) prev_carry_cols = ( [c for c in prev_nodes.columns if c.startswith(REENTRY_HIDDEN_COLUMN_PREFIX)] if prev_nodes is not None else [] From 176e4e8b16a8b93f4e7aeb921365814732e1668b Mon Sep 17 00:00:00 2001 From: Codex Worker Date: Tue, 27 Oct 2026 00:51:53 -0700 Subject: [PATCH 3/9] fix(gfql): narrow reentry edge operation --- graphistry/compute/chain.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/graphistry/compute/chain.py b/graphistry/compute/chain.py index fd62fb492e..ec0c3291bc 100644 --- a/graphistry/compute/chain.py +++ b/graphistry/compute/chain.py @@ -326,7 +326,8 @@ def apply_output_slice(op: ASTObject, op_label: ASTObject, df): ): prev_nodes = label_steps[idx - 2][1]._nodes prev_edges = label_steps[idx - 1][1]._edges - edge_op = cast(ASTEdge, label_steps[idx - 1][0]) + 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 [] From ee108da6ba6c1484d979253578aa6b4567eaa9a9 Mon Sep 17 00:00:00 2001 From: Codex Worker Date: Tue, 27 Oct 2026 00:59:24 -0700 Subject: [PATCH 4/9] fix(gfql): tolerate unhashable scalar reentry values --- graphistry/compute/chain.py | 11 +++++++++-- 1 file changed, 9 insertions(+), 2 deletions(-) diff --git a/graphistry/compute/chain.py b/graphistry/compute/chain.py index ec0c3291bc..8107ac06b7 100644 --- a/graphistry/compute/chain.py +++ b/graphistry/compute/chain.py @@ -356,7 +356,10 @@ def apply_output_slice(op: ASTObject, op_label: ASTObject, df): ).drop(columns=[from_col, id]).rename(columns={to_col: id}) ) carry_map = df_concat(engine)(carry_maps) - carry_map = carry_map.drop_duplicates(subset=[id] + prev_carry_cols) + 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) @@ -386,7 +389,11 @@ def apply_output_slice(op: ASTObject, op_label: ASTObject, df): concat = df_concat(engine) dedup_cols = [id] + list(dict.fromkeys(carry_cols)) - out_df = concat(dfs_to_concat).drop_duplicates(subset=dedup_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: From f2385b6b80e1e05c6765d9f5a2c7712d0729e18c Mon Sep 17 00:00:00 2001 From: Codex Worker Date: Tue, 27 Oct 2026 01:47:58 -0700 Subject: [PATCH 5/9] fix: preserve carried Cypher scalars across Polars traversal --- CHANGELOG.md | 2 + .../compute/gfql/lazy/engine/polars/chain.py | 25 +++++++++++ graphistry/compute/gfql_unified.py | 29 ++++++------ .../compute/gfql/cypher/test_lowering.py | 44 +++++++++++++++++++ 4 files changed, 84 insertions(+), 16 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index fc19263c98..adca068e53 100644 --- a/CHANGELOG.md +++ b/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 and connected-scalar 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. diff --git a/graphistry/compute/gfql/lazy/engine/polars/chain.py b/graphistry/compute/gfql/lazy/engine/polars/chain.py index 3e0dc881cc..dc07313e10 100644 --- a/graphistry/compute/gfql/lazy/engine/polars/chain.py +++ b/graphistry/compute/gfql/lazy/engine/polars/chain.py @@ -1200,4 +1200,29 @@ 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: + edge_ops = [op for op in ops if isinstance(op, ASTEdge)] + edge_parts = [] + if len(edge_ops) == 1: + if edge_ops[0].direction != "reverse": + edge_parts.append((src, dst)) + if edge_ops[0].direction != "forward": + edge_parts.append((dst, src)) + carry_maps = [] + for from_col, to_col in edge_parts: + carry_maps.append( + final_edges_eager.select([from_col, to_col]) + .join(start_nodes.select([node_col, *seed_carry_cols]), + left_on=from_col, right_on=node_col, how="inner") + .drop(from_col) + .rename({to_col: node_col}) + ) + carry_map = pl.concat(carry_maps, how="vertical") if len(carry_maps) > 1 else carry_maps[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) diff --git a/graphistry/compute/gfql_unified.py b/graphistry/compute/gfql_unified.py index 3ab447bf64..f1f933cee3 100644 --- a/graphistry/compute/gfql_unified.py +++ b/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: @@ -2011,20 +2015,7 @@ def _execute_compiled_query_with_reentry_impl( prefix_row_count_for_whole_row = ( len(prefix_rows_for_whole_row) if prefix_rows_for_whole_row is not None else 0 ) - try: - compiled_base_graph, start_nodes = _compiled_query_reentry_state( - base_graph, - plan, - prefix_result, - engine=engine, - ) - except GFQLValidationError as exc: - if not ( - plan.scalar_columns - and prefix_row_count_for_whole_row > 1 - and _is_duplicate_carried_rows_reentry_error(exc) - ): - raise + if plan.scalar_columns and prefix_row_count_for_whole_row > 1: if compiled_query.optional_reentry: raise _reentry_validation_error( "Cypher OPTIONAL MATCH after a multi-row whole-row WITH prefix is not yet supported" @@ -2032,7 +2023,7 @@ def _execute_compiled_query_with_reentry_impl( value=prefix_row_count_for_whole_row, suggestion="Use MATCH instead of OPTIONAL MATCH, or reduce the WITH prefix to a single row", field="optional_reentry", - ) from exc + ) row_results = [] for i in range(prefix_row_count_for_whole_row): row_prefix_result = _slice_reentry_prefix_result_row( @@ -2058,6 +2049,12 @@ def _execute_compiled_query_with_reentry_impl( return _union_scalar_reentry_results( row_results, base_graph=base_graph, engine=engine ) + compiled_base_graph, start_nodes = _compiled_query_reentry_state( + base_graph, + plan, + prefix_result, + engine=engine, + ) result = _execute_compiled_query( compiled_base_graph, compiled_query=compiled_query, diff --git a/graphistry/tests/compute/gfql/cypher/test_lowering.py b/graphistry/tests/compute/gfql/cypher/test_lowering.py index 19ab099485..f7833e0aa0 100644 --- a/graphistry/tests/compute/gfql/cypher/test_lowering.py +++ b/graphistry/tests/compute/gfql/cypher/test_lowering.py @@ -22888,6 +22888,10 @@ def test_lowering_coverage_helper_direct_margin_paths() -> None: boundary_exc = GFQLValidationError(ErrorCode.E108, "forced", field="return", value=["a", "b"]) assert _is_multi_source_match_alias_boundary_error(boundary_exc, alias_targets={"a": ASTNode(name="a"), "b": ASTNode(name="b")}) + assert not _is_multi_source_match_alias_boundary_error( + GFQLValidationError(ErrorCode.E201, "forced", field="return", value=["a", "b"]), + alias_targets={"a": ASTNode(name="a"), "b": ASTNode(name="b")}, + ) assert _expr_property_access_node_aliases( "[a.city, r.weight]", parse_row_expr=_parse_row_expr, @@ -22913,6 +22917,26 @@ def test_string_cypher_parenthesized_and_preserves_missing_property_as_null() -> assert none._nodes["id"].tolist() == [] if "id" in none._nodes.columns else len(none._nodes) == 0 +@pytest.mark.parametrize("engine", ["pandas", "polars"]) +def test_issue_2252_whole_plus_scalar_prefix_survives_match_on_pandas_and_polars(engine: str) -> None: + nodes_df, edges_df = _mk_reentry_carried_scalar_data() + nodes_df = nodes_df.assign(label__B=[False, False, True, True]) + graph = _mk_polars_graph(nodes_df, edges_df) if engine == "polars" else _mk_graph(nodes_df, edges_df) + + result = graph.gfql( + "MATCH (a:A) WITH a, a.num AS property MATCH (a)-->(b:B) " + "RETURN property, count(DISTINCT b.id) AS messageCount ORDER BY property", + engine=engine, + language="cypher", + ) + + records = _to_pandas_df(result._nodes).to_dict(orient="records") + assert records == [ + {"property": 1.0, "messageCount": 1}, + {"property": 2.0, "messageCount": 1}, + ] + + @pytest.mark.parametrize("engine", ["pandas", "polars"]) def test_issue_2252_scalar_only_prefix_survives_match_on_pandas_and_polars(engine: str) -> None: nodes_df, edges_df = _mk_reentry_carried_scalar_data() @@ -22930,6 +22954,26 @@ def test_issue_2252_scalar_only_prefix_survives_match_on_pandas_and_polars(engin assert records == [{"property": 1.0, "messageCount": 2}, {"property": 2.0, "messageCount": 2}] +@pytest.mark.parametrize("engine", ["pandas", "polars"]) +def test_issue_2252_connected_scalar_prefix_survives_match_on_pandas_and_polars(engine: str) -> None: + nodes_df, edges_df = _mk_connected_reentry_carried_scalar_data() + graph = _mk_polars_graph(nodes_df, edges_df) if engine == "polars" else _mk_graph(nodes_df, edges_df) + + result = graph.gfql( + "MATCH (a:A)-[:R]->(b:B) WITH b, b.score AS property " + "MATCH (b)-[:S]->(c:C) " + "RETURN property, count(DISTINCT c.id) AS messageCount ORDER BY property", + engine=engine, + language="cypher", + ) + + records = _to_pandas_df(result._nodes).to_dict(orient="records") + assert records == [ + {"property": 10.0, "messageCount": 1}, + {"property": 20.0, "messageCount": 1}, + ] + + @pytest.mark.parametrize("engine", ["pandas", "polars"]) def test_issue_2252_scalar_only_prefix_negative_is_empty_on_pandas_and_polars(engine: str) -> None: nodes_df, edges_df = _mk_reentry_carried_scalar_data() From ea04b7aea37352199e949c62cf273aafe2312620 Mon Sep 17 00:00:00 2001 From: Codex Worker Date: Tue, 27 Oct 2026 01:59:19 -0700 Subject: [PATCH 6/9] fix: correlate carried Cypher scalars across Polars edge paths --- CHANGELOG.md | 2 +- .../compute/gfql/lazy/engine/polars/chain.py | 39 ++++++----- graphistry/compute/gfql_unified.py | 23 ++++--- .../compute/gfql/cypher/test_lowering.py | 65 +++++++++++++++++++ 4 files changed, 104 insertions(+), 25 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index adca068e53..0024a8e5f2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,7 +10,7 @@ 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 and connected-scalar 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 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. diff --git a/graphistry/compute/gfql/lazy/engine/polars/chain.py b/graphistry/compute/gfql/lazy/engine/polars/chain.py index dc07313e10..8b1dbead2b 100644 --- a/graphistry/compute/gfql/lazy/engine/polars/chain.py +++ b/graphistry/compute/gfql/lazy/engine/polars/chain.py @@ -1205,23 +1205,30 @@ def _chain_traversal_polars(self: Plottable, ops, start_nodes: Optional[Any] = N if isinstance(col, str) and col.startswith("__cypher_reentry_") ] if seed_carry_cols: - edge_ops = [op for op in ops if isinstance(op, ASTEdge)] - edge_parts = [] - if len(edge_ops) == 1: - if edge_ops[0].direction != "reverse": - edge_parts.append((src, dst)) - if edge_ops[0].direction != "forward": - edge_parts.append((dst, src)) - carry_maps = [] - for from_col, to_col in edge_parts: - carry_maps.append( - final_edges_eager.select([from_col, to_col]) - .join(start_nodes.select([node_col, *seed_carry_cols]), - left_on=from_col, right_on=node_col, how="inner") - .drop(from_col) - .rename({to_col: node_col}) + 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: + continue + edge_src, edge_dst = edge_step._source, edge_step._destination + assert edge_src is not None and edge_dst is not None + 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] ) - carry_map = pl.concat(carry_maps, how="vertical") if len(carry_maps) > 1 else carry_maps[0] final_nodes_eager = final_nodes_eager.drop(seed_carry_cols).join( carry_map, on=node_col, how="left" ) diff --git a/graphistry/compute/gfql_unified.py b/graphistry/compute/gfql_unified.py index f1f933cee3..59c3ae08d4 100644 --- a/graphistry/compute/gfql_unified.py +++ b/graphistry/compute/gfql_unified.py @@ -2015,7 +2015,20 @@ def _execute_compiled_query_with_reentry_impl( prefix_row_count_for_whole_row = ( len(prefix_rows_for_whole_row) if prefix_rows_for_whole_row is not None else 0 ) - if plan.scalar_columns and prefix_row_count_for_whole_row > 1: + try: + compiled_base_graph, start_nodes = _compiled_query_reentry_state( + base_graph, + plan, + prefix_result, + engine=engine, + ) + except GFQLValidationError as exc: + if not ( + plan.scalar_columns + and prefix_row_count_for_whole_row > 1 + and _is_duplicate_carried_rows_reentry_error(exc) + ): + raise if compiled_query.optional_reentry: raise _reentry_validation_error( "Cypher OPTIONAL MATCH after a multi-row whole-row WITH prefix is not yet supported" @@ -2023,7 +2036,7 @@ def _execute_compiled_query_with_reentry_impl( value=prefix_row_count_for_whole_row, suggestion="Use MATCH instead of OPTIONAL MATCH, or reduce the WITH prefix to a single row", field="optional_reentry", - ) + ) from exc row_results = [] for i in range(prefix_row_count_for_whole_row): row_prefix_result = _slice_reentry_prefix_result_row( @@ -2049,12 +2062,6 @@ def _execute_compiled_query_with_reentry_impl( return _union_scalar_reentry_results( row_results, base_graph=base_graph, engine=engine ) - compiled_base_graph, start_nodes = _compiled_query_reentry_state( - base_graph, - plan, - prefix_result, - engine=engine, - ) result = _execute_compiled_query( compiled_base_graph, compiled_query=compiled_query, diff --git a/graphistry/tests/compute/gfql/cypher/test_lowering.py b/graphistry/tests/compute/gfql/cypher/test_lowering.py index f7833e0aa0..974ea2551c 100644 --- a/graphistry/tests/compute/gfql/cypher/test_lowering.py +++ b/graphistry/tests/compute/gfql/cypher/test_lowering.py @@ -22974,6 +22974,71 @@ def test_issue_2252_connected_scalar_prefix_survives_match_on_pandas_and_polars( ] +@pytest.mark.parametrize("engine", ["pandas", "polars"]) +def test_issue_2252_multi_edge_prefix_keeps_path_scalar_on_pandas_and_polars(engine: str) -> None: + nodes_df = pd.DataFrame( + { + "id": ["a1", "a2", "b1", "b2", "c1", "c2"], + "label__A": [True, True, False, False, False, False], + "label__B": [False, False, True, True, False, False], + "label__C": [False, False, False, False, True, True], + "num": [1, 2, None, None, None, None], + } + ) + edges_df = pd.DataFrame( + { + "s": ["a1", "a2", "b1", "b2"], + "d": ["b1", "b2", "c1", "c2"], + "type": ["R", "R", "S", "S"], + } + ) + graph = _mk_polars_graph(nodes_df, edges_df) if engine == "polars" else _mk_graph(nodes_df, edges_df) + + result = graph.gfql( + "MATCH (a:A) WITH a, a.num AS p " + "MATCH (a)-[:R]->(b:B)-[:S]->(c:C) " + "RETURN p, c.id AS cid ORDER BY p", + engine=engine, + language="cypher", + ) + + assert _to_pandas_df(result._nodes).to_dict(orient="records") == [ + {"p": 1.0, "cid": "c1"}, + {"p": 2.0, "cid": "c2"}, + ] + + +@pytest.mark.parametrize("engine", ["pandas", "polars"]) +def test_issue_2252_multi_edge_prefix_negative_is_empty_on_pandas_and_polars(engine: str) -> None: + nodes_df = pd.DataFrame( + { + "id": ["a1", "a2", "b1", "b2", "c1", "c2"], + "label__A": [True, True, False, False, False, False], + "label__B": [False, False, True, True, False, False], + "label__C": [False, False, False, False, True, True], + "num": [1, 2, None, None, None, None], + } + ) + edges_df = pd.DataFrame( + { + "s": ["a1", "a2", "b1", "b2"], + "d": ["b1", "b2", "c1", "c2"], + "type": ["R", "R", "S", "S"], + } + ) + graph = _mk_polars_graph(nodes_df, edges_df) if engine == "polars" else _mk_graph(nodes_df, edges_df) + + result = graph.gfql( + "MATCH (a:A {num: 99}) WITH a, a.num AS p " + "MATCH (a)-[:R]->(b:B)-[:S]->(c:C) " + "RETURN p, c.id AS cid", + engine=engine, + language="cypher", + ) + + assert _to_pandas_df(result._nodes).to_dict(orient="records") == [] + + @pytest.mark.parametrize("engine", ["pandas", "polars"]) def test_issue_2252_scalar_only_prefix_negative_is_empty_on_pandas_and_polars(engine: str) -> None: nodes_df, edges_df = _mk_reentry_carried_scalar_data() From b0982c5751ff65e69a77e2cac6005c3b9323a2c2 Mon Sep 17 00:00:00 2001 From: Codex Worker Date: Tue, 27 Oct 2026 02:23:13 -0700 Subject: [PATCH 7/9] fix: preserve convergent Cypher carry paths --- graphistry/compute/chain.py | 13 +++++++- .../compute/gfql/cypher/test_lowering.py | 30 +++++++++++++++++-- 2 files changed, 40 insertions(+), 3 deletions(-) diff --git a/graphistry/compute/chain.py b/graphistry/compute/chain.py index 8107ac06b7..b35bc2db1b 100644 --- a/graphistry/compute/chain.py +++ b/graphistry/compute/chain.py @@ -1565,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}), @@ -1578,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 diff --git a/graphistry/tests/compute/gfql/cypher/test_lowering.py b/graphistry/tests/compute/gfql/cypher/test_lowering.py index 974ea2551c..b949979058 100644 --- a/graphistry/tests/compute/gfql/cypher/test_lowering.py +++ b/graphistry/tests/compute/gfql/cypher/test_lowering.py @@ -280,8 +280,9 @@ def _mk_connected_reentry_carried_scalar_data() -> Tuple[pd.DataFrame, pd.DataFr pd.DataFrame( { "s": ["a1", "a2", "b1", "b2"], - "d": ["b1", "b2", "c1", "c2"], + "d": ["b1", "b2", "c1", "c1"], "type": ["R", "R", "S", "S"], + "weight": [None, None, 10, 20], } ), ) @@ -12716,7 +12717,7 @@ def test_string_cypher_executes_with_match_reentry_where_xor_on_carried_and_trai ) result = _mk_connected_reentry_carried_scalar_graph().gfql(query) - assert result._nodes.to_dict(orient="records") == [] + assert result._nodes.to_dict(orient="records") == [{"aid": "a2", "cid": "c1"}] def test_string_cypher_executes_with_match_reentry_preserves_orderby_single_column_limit_prefix() -> None: @@ -22974,6 +22975,31 @@ def test_issue_2252_connected_scalar_prefix_survives_match_on_pandas_and_polars( ] +@pytest.mark.parametrize("engine", ["pandas", "polars"]) +def test_issue_2252_convergent_destination_suffix_filters_stay_path_scoped(engine: str) -> None: + nodes_df, edges_df = _mk_connected_reentry_carried_scalar_data() + graph = _mk_polars_graph(nodes_df, edges_df) if engine == "polars" else _mk_graph(nodes_df, edges_df) + + partial = graph.gfql( + "MATCH (a:A)-[:R]->(b:B) WITH b, b.score AS property " + "MATCH (b)-[:S {weight: 10}]->(c:C) " + "RETURN property, count(DISTINCT c.id) AS messageCount ORDER BY property", + engine=engine, + language="cypher", + ) + assert _to_pandas_df(partial._nodes).to_dict(orient="records") == [ + {"property": 10.0, "messageCount": 1} + ] + + empty = graph.gfql( + "MATCH (a:A)-[:R]->(b:B) WITH b, b.score AS property " + "MATCH (b)-[:S {weight: 99}]->(c:C) " + "RETURN property, count(DISTINCT c.id) AS messageCount", + engine=engine, + language="cypher", + ) + assert _to_pandas_df(empty._nodes).to_dict(orient="records") == [] + @pytest.mark.parametrize("engine", ["pandas", "polars"]) def test_issue_2252_multi_edge_prefix_keeps_path_scalar_on_pandas_and_polars(engine: str) -> None: nodes_df = pd.DataFrame( From 4bcbbe5e59f8febd7b0d5fb07d4130e16738dc94 Mon Sep 17 00:00:00 2001 From: Codex Worker Date: Tue, 27 Oct 2026 02:30:11 -0700 Subject: [PATCH 8/9] fix: narrow Polars edge endpoints --- graphistry/compute/gfql/lazy/engine/polars/chain.py | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/graphistry/compute/gfql/lazy/engine/polars/chain.py b/graphistry/compute/gfql/lazy/engine/polars/chain.py index 8b1dbead2b..869a5f0228 100644 --- a/graphistry/compute/gfql/lazy/engine/polars/chain.py +++ b/graphistry/compute/gfql/lazy/engine/polars/chain.py @@ -1207,10 +1207,14 @@ def _chain_traversal_polars(self: Plottable, ops, start_nodes: Optional[Any] = N 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: + 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 - assert edge_src is not None and edge_dst is not None edge_parts = [] if edge_op.direction != "reverse": edge_parts.append((edge_src, edge_dst)) From cb2f6e0d0f9509c677f2b1fd8e6731fb5de980e3 Mon Sep 17 00:00:00 2001 From: Codex Worker Date: Tue, 27 Oct 2026 02:55:51 -0700 Subject: [PATCH 9/9] test: cover convergent cuDF carries --- .../compute/gfql/cypher/test_lowering.py | 32 +++++++++++++++++++ 1 file changed, 32 insertions(+) diff --git a/graphistry/tests/compute/gfql/cypher/test_lowering.py b/graphistry/tests/compute/gfql/cypher/test_lowering.py index b949979058..d15a6f3d41 100644 --- a/graphistry/tests/compute/gfql/cypher/test_lowering.py +++ b/graphistry/tests/compute/gfql/cypher/test_lowering.py @@ -23000,6 +23000,38 @@ def test_issue_2252_convergent_destination_suffix_filters_stay_path_scoped(engin ) assert _to_pandas_df(empty._nodes).to_dict(orient="records") == [] + +def test_issue_2252_convergent_destination_keeps_carry_on_cudf() -> None: + """cuDF parity for path-scoped carries entering the same destination (#2252).""" + pytest.importorskip("cudf") + _require_cudf_runtime() + graph = _mk_connected_reentry_carried_scalar_graph_cudf() + + positive = graph.gfql( + "MATCH (a:A)-[:R]->(b:B) WITH b, b.score AS property " + "MATCH (b)-[:S]->(c:C) " + "RETURN property, count(DISTINCT c.id) AS messageCount ORDER BY property", + engine="cudf", + language="cypher", + ) + assert type(positive._nodes).__module__.startswith("cudf") + assert _to_pandas_df(positive._nodes).to_dict(orient="records") == [ + {"property": 10.0, "messageCount": 1}, + {"property": 20.0, "messageCount": 1}, + ] + + partial = graph.gfql( + "MATCH (a:A)-[:R]->(b:B) WITH b, b.score AS property " + "MATCH (b)-[:S {weight: 10}]->(c:C) " + "RETURN property, count(DISTINCT c.id) AS messageCount ORDER BY property", + engine="cudf", + language="cypher", + ) + assert type(partial._nodes).__module__.startswith("cudf") + assert _to_pandas_df(partial._nodes).to_dict(orient="records") == [ + {"property": 10.0, "messageCount": 1} + ] + @pytest.mark.parametrize("engine", ["pandas", "polars"]) def test_issue_2252_multi_edge_prefix_keeps_path_scalar_on_pandas_and_polars(engine: str) -> None: nodes_df = pd.DataFrame(