review(#1785): pin the edge sort under streaming; scope the survivor claim; kill the twin guard

Review skill found three items worth fixing; all three were claim-scope or
follow-up-hazard rather than defects in the shipped behaviour.

1. `sort(EORD)` is NOT dead code — the probe that called it unpinned was engine
   blind. It only exercised the in-memory collect, where the frame comes back
   EORD-ordered even with the sort deleted. Under `GFQL_POLARS_CPU_STREAMING=1`
   it does not, and a trailing rows(limit=)/skip would then slice the wrong rows.
   The existing 60k order test is now parametrized over the collect engine.
   Mutation-checked: deleting the sort fails [streaming] and PASSES [in-memory],
   which is exactly why the original probe saw nothing.

2. The `maintain_order=True` docstring claimed the surviving duplicate is decided
   "the same way it was before", citing a 400-combo A/B. That A/B ran on the
   default engine only; under streaming the survivor genuinely differs. Scoped the
   claim to the in-memory collect and said plainly that the survivor is a stable
   property there, not a guaranteed one.

3. `_apply_node_names` still carried the ORIGINAL spelling of the bug this branch
   exists to fix: `is_lazy(df) or df.height > 0`, where the function is always
   called with lazified steps, so `is_lazy` short-circuits True and the height test
   is unreachable. Restated against `edges_empty`, which survives lazification.
   Unlike the edges combine this guard is SEMANTIC, not a cost guard — an empty
   next edge step must not empty `named` through the gate below it. I could not
   make the dead version observable, so this is hardening, not a bug fix; but
   leaving a known-dead cardinality guard beside a freshly-revived one is precisely
   how the original defect recurred.

lint + mypy clean; 1159 polars chain/combine/semi-dedup tests pass.

Author: lmeyerov

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

@@ -358,7 +358,11 @@ def _materialize_node_rows(all_nodes, step_ids, endpoint_ids_frame, node_col):
     Row ORDER out of here is arbitrary — a polars semi-join does not preserve left-frame order —
     and the caller restores input-frame order with an explicit sort. ``maintain_order`` is kept
     verbatim from the pre-refactor call so that WHICH duplicate row survives is decided the same
-    way it was before (A/B over 400 duplicate-id combos: identical full frames)."""
+    way it was before: A/B over 400 duplicate-id combos gives identical full frames **under the
+    default in-memory collect**. Scoped deliberately — under streaming collect the survivor DOES
+    differ from the pre-refactor call (measured), so it is not a guaranteed property of this
+    helper, only a stable one on the default engine. Anything needing a specific survivor must
+    order explicitly rather than rely on this."""
     import polars as pl
     ids = pl.concat([step_ids, endpoint_ids_frame], how="vertical_relaxed")
     return all_nodes.join(ids, on=node_col, how="semi").unique(
@@ -399,7 +403,16 @@ def _apply_node_names(out, g, steps, auto_hop_col: str = _AUTO_NODE_HOP):
                     on=node_col, how="semi")
         if idx + 1 < len(step_list):
             next_op, next_step = step_list[idx + 1]
-            if isinstance(next_op, ASTEdge) and next_step._edges is not None and (is_lazy(next_step._edges) or next_step._edges.height > 0):
+            # Cardinality guard, restated against a fact that SURVIVES lazification. The old
+            # spelling was `is_lazy(df) or df.height > 0`, and `_apply_node_names` is always
+            # called with lazified steps — so `is_lazy` short-circuited True and the height
+            # test was unreachable. That is the identical silent death this commit fixes one
+            # function above; leaving a second copy of it here is how the bug recurs.
+            # Unlike the edges combine this one is SEMANTIC, not a cost guard: an empty next
+            # edge step must not empty `named` via the gate below.
+            next_edges_empty = getattr(next_step, "edges_empty", None)
+            if (isinstance(next_op, ASTEdge) and next_step._edges is not None
+                    and next_edges_empty is not True):
                 e = next_step._edges
                 if next_op.direction == "forward":
                     part = e.select(pl.col(src).alias(node_col))

graphistry/tests/compute/gfql/test_engine_polars_narrow_combine.py

@@ -337,11 +337,19 @@ def test_duplicate_node_ids_still_collapse_to_one_row():
     assert graph_sig(g_pd.chain(chain, engine="pandas")) == graph_sig(out)
 
 
-def test_output_row_order_survives_a_frame_big_enough_to_parallelize():
+@pytest.mark.parametrize("streaming", [False, True], ids=["in-memory", "streaming"])
+def test_output_row_order_survives_a_frame_big_enough_to_parallelize(streaming):
     """Order at fixture scale proves little: polars' joins happen to come back in left order on a
     handful of rows and only reorder once the hash join actually runs in parallel. Use a frame
     large enough to reorder, shuffled so input order is not id order, and pin that the combine's
-    explicit sorts put both output frames back into input-frame order."""
+    explicit sorts put both output frames back into input-frame order.
+
+    Parametrized over the collect engine because the IN-MEMORY engine hides the edge sort: with
+    `final_edges.sort(EORD)` deleted, in-memory still returns EORD-ordered rows at every size
+    probed, so an in-memory-only test would call that sort dead code. Under STREAMING it does
+    not, and a trailing rows(limit=)/skip would then slice the wrong rows. Both engines here,
+    so neither sort can be removed on the strength of the other's silence."""
+    from graphistry.compute.gfql.lazy import set_cpu_streaming
     size = 60_000
     rng = list(range(size))
     order = rng[1::2] + rng[0::2][::-1]                       # deterministic shuffle
@@ -350,8 +358,12 @@ def test_output_row_order_survives_a_frame_big_enough_to_parallelize():
                           "d": [(k * 13 + 1) % size for k in order],
                           "type": ["K" if k % 2 else "L" for k in order]})
     _, g_pl = _pair(nodes, edges)
-    out = g_pl.chain([n({"grp": 1}, name="m"), e_forward({"type": "K"}, name="r"), n(name="p")],
-                     engine="polars")
+    set_cpu_streaming(streaming)
+    try:
+        out = g_pl.chain([n({"grp": 1}, name="m"), e_forward({"type": "K"}, name="r"), n(name="p")],
+                         engine="polars")
+    finally:
+        set_cpu_streaming(None)
     assert out._nodes.height > 1000 and out._edges.height > 1000, "fixture stopped being big"
 
     node_rank = {k: i for i, k in enumerate(nodes["key"].tolist())}