refactor(gfql): move engine-agnostic frame/series helpers to Engine.py

The reentry executor accumulated a cluster of generic pandas/cuDF/polars
dispatch helpers (null mask, series/frame filter, order-preserving left join,
constant-column assign, drop-columns, row-as-mapping, series-to-pylist) that
have no reentry semantics and belong with safe_merge / df_concat / df_unique in
the engine layer. gfql_unified.py had already re-implemented series_to_pylist,
proving the leak. Move the eight primitives to graphistry.Engine (public names,
no underscore), drop the reentry-local defs, and consolidate the duplicate onto
the more-defensive version. SeriesT/DataFrameT typing preserved via a
TYPE_CHECKING import + string annotations so Engine.py (imported very early)
never triggers graphistry.compute package init at runtime (would be circular).

No behavior change: byte-identical dispatch; reentry suite 19/19, cypher/entity
subset green, ruff clean, mypy-neutral.

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

Author: lmeyerov

graphistry/Engine.py

@@ -3,12 +3,18 @@
 import numpy as np
 import pandas as pd
 import pyarrow as pa
-from typing import Any, List, Optional, Union
+from typing import TYPE_CHECKING, Any, Dict, List, Mapping, Optional, Sequence, Union
 from typing_extensions import Literal
 from enum import Enum
 
 from graphistry.models.types import ValidationParam
 
+if TYPE_CHECKING:
+    # Frame aliases (pandas types for mypy; ``Any`` at runtime). Imported under TYPE_CHECKING
+    # and referenced via string annotations below so Engine.py — imported very early — never
+    # triggers ``graphistry.compute`` package init at runtime (would be circular).
+    from graphistry.compute.typing import DataFrameT, SeriesT
+
 
 class Engine(Enum):
     PANDAS = 'pandas'
@@ -850,3 +856,103 @@ def safe_merge(
         raise ValueError("Must specify either 'on' or both 'left_on' and 'right_on'")
 
     return result
+
+
+# ---------------------------------------------------------------------------
+# Engine-agnostic series / frame primitives (pandas / cuDF / polars dispatch).
+#
+# Pure per-row/per-column dispatch helpers with no domain knowledge — the polars
+# branches genuinely return polars objects mypy can't narrow to the pandas frame
+# aliases, so the localized ``# type: ignore`` lives here at the dispatch point and
+# callers get a clean ``SeriesT`` / ``DataFrameT`` contract with no ``cast()``.
+# Annotations are strings so the TYPE_CHECKING-only alias import stays runtime-free.
+# ---------------------------------------------------------------------------
+
+
+def is_series_like(s: object) -> bool:
+    """True for a pandas/cuDF Series (``.dropna``) or a polars Series (module check).
+
+    Some engine-agnostic callers accept an ``ids`` Series that is pandas under
+    ``engine='pandas'`` and polars under ``engine='polars'``; both are valid."""
+    return hasattr(s, "dropna") or is_polars_df(s)
+
+
+def series_not_null_mask(s: "SeriesT") -> "SeriesT":
+    """Non-null boolean mask, engine-aware (polars ``is_not_null`` vs pandas ``notna``)."""
+    if is_polars_df(s):
+        return s.is_not_null()  # type: ignore[attr-defined,no-any-return]
+    return s.notna()
+
+
+def series_filter(s: "SeriesT", mask: "SeriesT") -> "SeriesT":
+    """Filter a Series by a boolean mask, engine-aware, dropping the old index (pandas)."""
+    if is_polars_df(s):
+        return s.filter(mask)  # type: ignore[attr-defined,no-any-return]
+    return s[mask].reset_index(drop=True)
+
+
+def frame_filter(df: "DataFrameT", mask: "SeriesT") -> "DataFrameT":
+    """Filter a DataFrame's rows by a boolean mask, engine-aware, dropping the old index."""
+    if is_polars_df(df):
+        return df.filter(mask)  # type: ignore[attr-defined,no-any-return]
+    return df.loc[mask].reset_index(drop=True)
+
+
+def ordered_left_join(left: "DataFrameT", right: "DataFrameT", *, on: str) -> "DataFrameT":
+    """Left join preserving ``left`` row order, engine-aware. Polars ``.merge`` does not exist;
+    ``safe_merge`` (pandas/cuDF ``.merge``) cannot run on polars frames, so branch to
+    ``.join(..., maintain_order='left')`` which pins the left-row ordering the caller needs.
+
+    ``right`` may arrive on a different engine than ``left`` (e.g. natively-projected polars
+    ``left`` against a still-pandas base table), so align ``right`` onto ``left``'s engine
+    before the polars join."""
+    if is_polars_df(left):
+        if not is_polars_df(right):
+            right = df_to_engine(right, Engine.POLARS)
+        return left.join(right, on=on, how="left", maintain_order="left")  # type: ignore[call-arg,no-any-return]
+    return safe_merge(left, right, on=on, how="left")
+
+
+def row_as_mapping(rows: "DataFrameT", row_index: int) -> Mapping[str, Any]:
+    """One frame row as a col->scalar mapping, engine-aware (``row[col]`` works for
+    both the pandas Series and the polars named-row dict)."""
+    if is_polars_df(rows):
+        return rows.row(row_index, named=True)  # type: ignore[attr-defined,no-any-return]
+    return rows.iloc[row_index]
+
+
+def assign_constant_columns(df: "DataFrameT", values: Dict[str, Any]) -> "DataFrameT":
+    """Broadcast scalar ``values`` as constant columns, engine-aware."""
+    if not values:
+        return df
+    if is_polars_df(df):
+        import polars as pl
+        return df.with_columns([pl.lit(v).alias(k) for k, v in values.items()])  # type: ignore[attr-defined,no-any-return]
+    return df.assign(**values)
+
+
+def drop_columns(df: "DataFrameT", cols: Sequence[str]) -> "DataFrameT":
+    """Drop columns by name, engine-aware (polars ``drop(list)`` vs pandas ``drop(columns=)``)."""
+    if is_polars_df(df):
+        return df.drop(list(cols))  # type: ignore[no-any-return]
+    return df.drop(columns=list(cols))
+
+
+def series_to_pylist(values: "SeriesT") -> List[Any]:
+    """Series -> python list, engine-aware, with defensive arrow/pandas fallbacks."""
+    if hasattr(values, "to_arrow"):
+        try:
+            return list(values.to_arrow().to_pylist())
+        except Exception:
+            pass
+    if hasattr(values, "to_pandas"):
+        try:
+            return list(values.to_pandas().tolist())
+        except Exception:
+            pass
+    if hasattr(values, "tolist"):
+        try:
+            return list(values.tolist())
+        except Exception:
+            pass
+    return list(values)

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

@@ -4,7 +4,22 @@
 
 from typing import Any, Dict, List, Mapping, Optional, Sequence, Tuple, Union, cast
 
-from graphistry.Engine import EngineAbstract, df_concat, df_cons, resolve_engine, safe_merge
+from graphistry.Engine import (
+    EngineAbstract,
+    assign_constant_columns,
+    df_concat,
+    df_cons,
+    drop_columns,
+    frame_filter,
+    is_series_like,
+    ordered_left_join,
+    resolve_engine,
+    row_as_mapping,
+    safe_merge,
+    series_filter,
+    series_not_null_mask,
+    series_to_pylist,
+)
 from graphistry.Plottable import Plottable
 from graphistry.compute.exceptions import GFQLValidationError, ErrorCode
 from graphistry.compute.gfql.cypher.reentry.naming import _reentry_hidden_column_name
@@ -23,79 +38,6 @@
 from graphistry.Engine import is_polars_df as _is_polars_df
 
 
-def _series_is_series_like(s: object) -> bool:
-    """True for a pandas/cuDF Series (``.dropna``) or a polars Series (module check).
-
-    The reentry executor recovers carried node identities from a projection-meta ``ids`` Series
-    that is pandas under ``engine='pandas'`` and polars under ``engine='polars'``; both are valid."""
-    return hasattr(s, "dropna") or _is_polars_df(s)
-
-
-# The four helpers below are engine-agnostic (pandas/cuDF/polars); `SeriesT`/`DataFrameT` are the
-# repo's frame aliases (pandas types for mypy). The polars branches genuinely return polars objects
-# mypy can't narrow to the alias, so the suppression lives here at the dispatch point — callers get
-# a clean `SeriesT`/`DataFrameT` contract with no cast().
-def _series_not_null_mask(s: SeriesT) -> SeriesT:
-    """Non-null boolean mask, engine-aware (polars ``is_not_null`` vs pandas ``notna``)."""
-    if _is_polars_df(s):
-        return s.is_not_null()  # type: ignore[attr-defined,no-any-return]
-    return s.notna()
-
-
-def _series_filter(s: SeriesT, mask: SeriesT) -> SeriesT:
-    """Filter a Series by a boolean mask, engine-aware, dropping the old index (pandas)."""
-    if _is_polars_df(s):
-        return s.filter(mask)  # type: ignore[attr-defined,no-any-return]
-    return s[mask].reset_index(drop=True)
-
-
-def _frame_filter(df: DataFrameT, mask: SeriesT) -> DataFrameT:
-    """Filter a DataFrame's rows by a boolean mask, engine-aware, dropping the old index."""
-    if _is_polars_df(df):
-        return df.filter(mask)  # type: ignore[attr-defined,no-any-return]
-    return df.loc[mask].reset_index(drop=True)
-
-
-def _ordered_left_join(left: DataFrameT, right: DataFrameT, *, on: str) -> DataFrameT:
-    """Left join preserving ``left`` row order, engine-aware. Polars ``.merge`` does not exist;
-    ``safe_merge`` (pandas/cuDF ``.merge``) cannot run on polars frames, so branch to
-    ``.join(..., maintain_order='left')`` which pins the WITH-row ordering the reentry seed needs.
-
-    Under ``engine='polars'`` the carried ids come from the natively-projected (polars) prefix
-    while the base node table can still be pandas (the base graph is converted lazily), so align
-    ``right`` onto ``left``'s engine before the polars join."""
-    if _is_polars_df(left):
-        from graphistry.Engine import Engine, df_to_engine
-        if not _is_polars_df(right):
-            right = df_to_engine(right, Engine.POLARS)
-        return left.join(right, on=on, how="left", maintain_order="left")  # type: ignore[call-arg,no-any-return]
-    return safe_merge(left, right, on=on, how="left")
-
-
-def _reentry_row(prefix_rows: DataFrameT, row_index: int) -> Mapping[str, Any]:
-    """One prefix row as a col->scalar mapping, engine-aware (``row[col]`` works for
-    both the pandas Series and the polars named-row dict)."""
-    if _is_polars_df(prefix_rows):
-        return prefix_rows.row(row_index, named=True)  # type: ignore[attr-defined,no-any-return]
-    return prefix_rows.iloc[row_index]
-
-
-def _assign_constant_columns(df: DataFrameT, values: Dict[str, Any]) -> DataFrameT:
-    """Broadcast scalar ``values`` as constant columns, engine-aware."""
-    if not values:
-        return df
-    if _is_polars_df(df):
-        import polars as pl
-        return df.with_columns([pl.lit(v).alias(k) for k, v in values.items()])  # type: ignore[attr-defined,no-any-return]
-    return df.assign(**values)
-
-
-def _drop_columns(df: Any, cols: Sequence[str]) -> Any:
-    if _is_polars_df(df):
-        return df.drop(list(cols))
-    return df.drop(columns=list(cols))
-
-
 def _bind_reentry_graph(graph: Plottable, node_rows: Optional[DataFrameT], *, empty_edges: bool = False) -> Plottable:
     out = graph.bind()
     out._nodes = node_rows
@@ -232,22 +174,14 @@ def _optional_reentry_key(record: Dict[str, Any], columns: Tuple[str, ...]) -> T
 
 
 def _records_for_columns(df: DataFrameT, columns: Tuple[str, ...]) -> List[Dict[str, Any]]:
-    values_by_column = {column: _series_to_pylist(cast(SeriesT, df[column])) for column in columns}
+    values_by_column = {column: series_to_pylist(cast(SeriesT, df[column])) for column in columns}
     row_count = len(df)
     return [
         {column: values_by_column[column][row_index] for column in columns}
         for row_index in range(row_count)
     ]
 
 
-def _series_to_pylist(values: SeriesT) -> List[Any]:
-    if hasattr(values, "to_arrow"):
-        return cast(List[Any], values.to_arrow().to_pylist())
-    if hasattr(values, "tolist"):
-        return cast(List[Any], values.tolist())
-    return list(values)
-
-
 def _optional_reentry_key_value(value: Any) -> Any:
     try:
         if value != value:
@@ -323,7 +257,7 @@ def compiled_query_reentry_state(
         )
     ids = meta["ids"]
     id_column = meta["id_column"]
-    if not _series_is_series_like(ids):
+    if not is_series_like(ids):
         raise reentry_validation_error(
             "Cypher MATCH after WITH could not recover carried node identities from the prefix stage",
             value=output_name,
@@ -457,10 +391,10 @@ def compiled_query_scalar_reentry_state(
             value=missing_column,
             suggestion="Project the scalar column explicitly before MATCH re-entry.",
         )
-    row = _reentry_row(prefix_rows, row_index)
+    row = row_as_mapping(prefix_rows, row_index)
     node_rows = cast(
         DataFrameT,
-        _assign_constant_columns(
+        assign_constant_columns(
             base_nodes,
             {
                 _reentry_hidden_column_name(output_name): row[output_name]
@@ -480,7 +414,7 @@ def freeform_broadcast_row_to_nodes(
     row_index: int,
 ) -> Plottable:
     """Broadcast one free-form prefix row's hidden carries onto the base nodes."""
-    row = _reentry_row(prefix_rows, row_index)
+    row = row_as_mapping(prefix_rows, row_index)
     broadcast_values: Dict[str, Any] = {
         _reentry_hidden_column_name(col): row[col]
         for col in plan.scalar_columns
@@ -495,11 +429,11 @@ def freeform_broadcast_row_to_nodes(
     if broadcast_values:
         existing_hidden = [c for c in base_nodes.columns if isinstance(c, str) and c.startswith("__cypher_reentry_")]
         node_rows = (
-            cast(DataFrameT, _drop_columns(base_nodes, existing_hidden))
+            cast(DataFrameT, drop_columns(base_nodes, existing_hidden))
             if existing_hidden
             else base_nodes
         )
-        node_rows = cast(DataFrameT, _assign_constant_columns(node_rows, broadcast_values))
+        node_rows = cast(DataFrameT, assign_constant_columns(node_rows, broadcast_values))
     else:
         node_rows = cast(DataFrameT, base_nodes)
 
@@ -548,18 +482,18 @@ def aligned_reentry_rows(
             value=output_name,
             suggestion="Retry with a direct whole-row carry through WITH or inspect intermediate row-shaping before MATCH re-entry.",
         )
-    if not _series_is_series_like(ids):
+    if not is_series_like(ids):
         raise reentry_validation_error(
             "Cypher MATCH after WITH could not align carried node identities from the prefix stage",
             value=output_name,
             suggestion=REENTRY_WHOLE_ROW_SUGGESTION,
         )
 
-    non_null_mask = _series_not_null_mask(ids)
-    carried_ids = _series_filter(ids, non_null_mask)
+    non_null_mask = series_not_null_mask(ids)
+    carried_ids = series_filter(ids, non_null_mask)
     if prefix_rows is None:
         return carried_ids, None
-    return carried_ids, _frame_filter(prefix_rows, non_null_mask)
+    return carried_ids, frame_filter(prefix_rows, non_null_mask)
 
 
 def reentry_carry_payload(
@@ -593,4 +527,4 @@ def ordered_reentry_start_nodes(
     id_column: str,
 ) -> DataFrameT:
     # MATCH re-entry must preserve the WITH row order, not the base node-table order.
-    return _ordered_left_join(carried_node_ids, node_rows, on=id_column)
+    return ordered_left_join(carried_node_ids, node_rows, on=id_column)

graphistry/compute/gfql_unified.py

@@ -6,7 +6,7 @@
 from types import MappingProxyType
 from typing import Any, Dict, List, Literal, Mapping, Optional, Sequence, Set, Tuple, Union, cast
 from graphistry.Plottable import Plottable
-from graphistry.Engine import Engine, EngineAbstract, POLARS_ENGINES, df_concat, df_cons, df_to_engine, df_unique, is_polars_df, resolve_engine
+from graphistry.Engine import Engine, EngineAbstract, POLARS_ENGINES, df_concat, df_cons, df_to_engine, df_unique, is_polars_df, resolve_engine, series_to_pylist
 from graphistry.util import setup_logger
 from .ast import ASTObject, ASTLet, ASTNode, ASTEdge, ASTCall
 from .chain import Chain, chain as chain_impl
@@ -83,25 +83,6 @@
 logger = setup_logger(__name__)
 
 
-def _series_to_pylist(values: Any) -> List[Any]:
-    if hasattr(values, "to_arrow"):
-        try:
-            return list(values.to_arrow().to_pylist())
-        except Exception:
-            pass
-    if hasattr(values, "to_pandas"):
-        try:
-            return list(values.to_pandas().tolist())
-        except Exception:
-            pass
-    if hasattr(values, "tolist"):
-        try:
-            return list(values.tolist())
-        except Exception:
-            pass
-    return list(values)
-
-
 def _is_duplicate_carried_rows_reentry_error(exc: GFQLValidationError) -> bool:
     context = getattr(exc, "context", None)
     if exc.code != ErrorCode.E108 or not isinstance(context, dict):
@@ -217,8 +198,8 @@ def _apply_optional_null_fill(
             language="cypher",
         )
 
-    base_ids = _series_to_pylist(base_rows_df[node_col])
-    matched_id_list = _series_to_pylist(matched_ids)
+    base_ids = series_to_pylist(base_rows_df[node_col])
+    matched_id_list = series_to_pylist(matched_ids)
     if len(base_ids) == actual_rows and base_ids == matched_id_list:
         return result