perf(gfql): empty-left merge shrink + polars EORD/EID row-index dedup (#1783)

* perf(gfql): shrink the right side of a left-merge when the left is empty

`_lean_prefilter_right` shrinks `right` to the keys present in `left` before a
how='left' merge. It declined to do so when `left` was EMPTY -- the one case where
shrinking is both maximally profitable and trivially correct: no left key can match
anything, and a left merge keeps only the right rows that DO match, so the result is
empty whatever `right` holds.

The cost of declining is graph-sized. Instrumenting a single-node named query on a
3.18M-node / 14M-edge graph showed the dominant work was:

    safe_merge(left=0x1, right=14000000x4, how='left')

i.e. a zero-row intermediate joined against every edge in the graph. Returning a
zero-row slice of `right` keeps the identical columns and dtypes -- so the merge still
produces the same schema -- without materializing the frame.

Measured on that graph: a named seed-only query goes 112.64 -> 17.29 ms (6.5x), same
result shape. Patterns whose intermediates are non-empty are unaffected.

Parity: 650 differential comparisons (pandas + polars x 13 shapes x 25 seeds, dense
graph, null-carrying columns) -- 0 divergences. Index + lean-combine suites: no new
failures (38 pre-existing GPU-image failures unchanged, 184 passed).

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VARwnAqmvczhz5EmP7TYyZ

* perf(gfql/polars): reuse the synthetic edge id as the stable edge order

The native polars chain executor attaches a synthetic edge id (`EID`) via
`with_row_index` when the graph has no edge binding, then — a few hundred lines later —
materializes a SECOND `with_row_index` column (`EORD`) over that same frame to restore
eager edge order for the fused combine. Both are 0..n-1 over the same frame in the same
order, because the second runs on the frame the first produced, so the second is pure
duplication of a graph-sized materialization.

Reuse EID as EORD when it was attached. Both columns were already dropped before
returning, so the single drop now covers it (guarded on `EID != EORD` for the case where
the graph brought its own edge binding and EORD is still a separate column).

Measured on a 3.18M-node / 14M-edge polars graph with polars-engine indexes, a seeded
typed pattern: 148.22 -> 135.43 ms (-8.6%). `with_row_index` was ~34 ms/query across the
eager calls before this.

Parity: 108 polars-path cases (9 traversal shapes x 12 random seeds, engine='polars' with
polars-engine indexes) hashed over column ORDER, dtypes, row count and full content --
0 differ, 0 raised. This is a new sweep: the existing 650-case differential runs
engine=auto, which resolves to pandas and never exercises this executor.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VARwnAqmvczhz5EmP7TYyZ

* docs+test(gfql): CHANGELOG and both-direction tests for the lean-combine shrink

Owner's merge conditions for this PR were a CHANGELOG entry, correctness tests in
both directions, and a pyg-bench lane (the last is separate).

Tests pin both sides of the new empty-left branch:
  - empty left shrinks to zero rows AND keeps columns/dtypes identical, so the
    downstream merge still produces the same schema;
  - the shrunk merge equals the unshrunk merge at the call site's actual how='left';
  - non-empty lefts (1, 5, 999, 1000 rows — spanning both sides of the ratio gate)
    are byte-identical to the unshrunk merge, i.e. the new branch perturbs nothing;
  - a non-empty left whose keys miss entirely still yields one null-filled row per
    left row, rather than being collapsed;
  - end-to-end: a seed matching nothing gives lean-on == lean-off through chain().

Mutation-checked. Reverting the branch to `return right` fails the shrink test.
A second mutation (also shrink when no keys overlap) is NOT caught — correctly:
under how='left' both forms give null-filled rows, so there is no behaviour to
distinguish. The docstring says so rather than implying coverage it does not have.

Also records the safety precondition in the CHANGELOG: the shrink is used at exactly
one call site, which is how='left'; a how='right'/'outer' merge retains unmatched
right rows and would NOT be safe to shrink this way.

* review(#1783): fix a silent no-op on float indexes; drop a dead guard; pin edge order

Review skill found no correctness defect (the adversarial question — is how='left'
really the only consumer of _lean_prefilter_right — came back CONFIRMED: one
non-test caller, the local is dead on the next line, and every other merge in
chain.py builds its right side independently). These are the quality items.

1. `right[0:0]` is LABEL-based slicing on a float index — pandas routes those
   through slice_indexer — so it returned ONE row and the shrink silently did
   nothing. Now `right.iloc[0:0]`, positional on every index type. The RESULT was
   always correct (a 0-row left yields a 0-row how='left' merge either way), which
   is why this needed its own test: the bug was an invisible perf no-op, not a
   wrong answer. Mutation-checked — reverting to `[0:0]` fails the new test.
2. Docstring said "Only shrinks when `left` is materially smaller", which the new
   empty-left branch contradicts. Restated, and now records that the whole helper
   is pandas-only in practice (`_lean_engine_ok` is checked first), so there is no
   cross-engine exposure.
3. Removed `if added_edge_index and EID != EORD:` — dead by construction, since the
   only branch setting added_edge_index also sets EORD = EID.

lint clean; 1022 lean-combine + polars-chain tests pass.

---------

Co-authored-by: Claude Opus 5 <noreply@anthropic.com>

Author: lmeyerov

CHANGELOG.md

@@ -12,6 +12,8 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 - **GFQL secondary node property indexes (`create_index('node_prop', column=...)` / `g.gfql_index_node_props([...])`)**: a seed predicate on a NON-key column — `MATCH (m {id: 42})` where the graph's node id binding is some other column — previously cost a full node scan, because the registry only indexed the node-id binding and the CSR adjacencies. A property index is the same pay-as-you-go sidecar as the existing kinds: sorted distinct values over node **row positions** (CSR, so duplicate values are indexable), never reorders `.nodes`, fingerprint-validated so a `.nodes()` rebind is treated as absent (safe miss, never a wrong answer), engine-polymorphic (numpy host / cupy on-device), and policy-gated (`off`/`use`/`auto`/`force`). The seeded fixed-hop planner picks the **most selective** indexed scalar predicate in the seed filter using a free CSR-offset estimate, gathers those candidates, and applies every remaining predicate to them — so results are identical whether the index is present, absent, stale, or cost-gated out. `show_indexes()` lists property indexes; `drop_index('node_prop', column=...)` drops one. Only integer columns are indexable today (float NaN ordering, strings on cupy, and nulls all decline to the scan); widening that is additive. **Perf (dgx-spark, official LDBC SNB SF1, 3.18M nodes, warm median, value-identical 19-row result):** interactive-short IS7 `71.6 ms -> 19.5 ms` (**3.7x**), with a one-time `112 ms` build — the seed lookup itself goes from a `51.2 ms` scan to `0.096 ms`.
 
 ### Performance
+- **Seeded chain combine stops joining against the full frame when the intermediate is empty**: `_lean_prefilter_right` shrinks the big side of the combine's `how='left'` merge to the keys actually present on the left, but it declined to do so in the one case where shrinking is both maximally profitable and trivially correct — an **empty** left. A left merge keeps only the right rows that match, so a zero-row left yields a zero-row result whatever the right side holds; the merge was nonetheless materializing the whole graph-sized frame. It now hands back a zero-row slice of `right` (same columns and dtypes, so the merge still produces an identical schema). Measured: a single-node query whose 0-row intermediate was joined against 14M edges went **112.64 → 17.29 ms**. The shrink is used at exactly one call site, which is `how='left'`; a merge that retains unmatched right rows (`right`/`outer`) would NOT be safe to shrink this way, and the tests pin both directions — the empty-left case must return an empty result, and the non-empty cases must be byte-identical to the unshrunk merge.
+- **Native polars chain reuses the synthetic edge id as its stable row order**: when the executor had already added a synthetic edge-index column it also added a second, separate row-index column purely to restore input order at the end — two full `with_row_index` passes over the edge frame, and a redundant column carried through every intermediate. The existing synthetic id is already a contiguous, order-preserving row index, so it is now reused as the sort key and dropped once at the end. Only the pre-existing column is reused; when no synthetic id was added the separate order column is still created, so graphs that bring their own edge id are unaffected. Value-identical including row order (148.22 → 135.43 ms on a 3.18M-node / 14M-edge polars graph).
 - **GFQL indexed typed-edge traversals stop scanning the whole edge frame (#1658)**: the index path built the simple-equality `edge_match` mask over ALL `E` edges — one `(series == val)` over the full column — and then read that mask only at `rows[edge_keep[rows]]`, the handful of positions the CSR adjacency lookup had already returned. That put an **O(E) predicate scan inside an O(degree) traversal**, which is why an indexed seeded typed hop scaled with the graph while the underlying `g.hop()` stayed flat. The predicate is now evaluated on the gathered candidate rows instead: `_build_edge_row_filter` does the schema-level validation up front (the dtype-mismatch decline that keeps error parity with the scan is O(1) and unchanged) and `_EdgeMatchRowFilter.mask_for(rows)` applies `col == val` to just those rows, using each frame's native `==` exactly as before (so cuDF string columns stay on the cuDF layer). This is a cost tradeoff, not a free win, and it is bounded rather than assumed: candidate-row evaluation beats one whole-column compare only while the candidates stay a small fraction of the frame, and a fixed-point walk that reaches most of the graph inverts it — the out- and in-indices are filtered separately, so it can gather up to 2E elements by random access against the eager form's single sequential pass over E (measured before the guard: an undirected `to_fixed_point` typed walk gathered 1.94×E and ran 1.2–1.6× SLOWER than master). A cumulative-cost guard now switches to the whole-column mask once gathered rows reach E/8, so total predicate work is bounded at ~1.125×E in that regime while a seeded hop — which gathers ~degree — never approaches the threshold and never builds it. An evaluation failure mid-traversal abandons the indexed path entirely so the scan fallback stays parity-safe. Measured on a 3.18M-node / 14M-edge polars graph (LDBC SNB SF1-shaped), a seeded typed-edge query with alias markers goes **309.9 → 57.4 ms (5.4×)**, value-identical, with the full-column compare (previously 248 ms, 81% of the query) gone from the profile. Parity held across 650 differential comparisons — pandas + polars × 13 traversal shapes × 25 random seeds on a dense graph with null-carrying string and numeric columns — comparing columns, dtypes, row counts and full content, including the decline paths (membership `edge_match` still routes to the scan; a dtype mismatch still raises the same `GFQLSchemaError`). One honest limit on error parity: only the O(1) dtype gates are parity-exact. A *data-dependent* comparison failure (e.g. a list-valued cell in an object column) is now observed only if it lands in the gathered candidates, so the indexed path can succeed where the scan raises. That needs pathological cell values to reach, but it is a real narrowing of the previous guarantee. Pinned by two structural regression tests that assert the *shape* rather than a wall-clock number: the predicate must see only candidate rows, and growing the graph 8× at fixed degree must not grow the number of rows it examines.
 - **GFQL indexed fixed-hop bindings now dispatch BEFORE the canonical traversal (pandas / cuDF / native Polars)**: the resident-index fixed-hop path (`rows(binding_ops=...)`, the Cypher multi-alias `MATCH … RETURN` shape) was consulted only *inside* row materialization — after the engine had already run the full canonical traversal — so its compact path bag was computed on top of the graph-sized work it exists to avoid. Both the pandas/cuDF chain boundary and the native Polars chain now ask the same shared, structural gate first and, only when it can serve the WHOLE middle exactly, skip the canonical traversal and hand the compact state to the unchanged row materializer. Engagement stays operator/index/dtype/cost based — no query, schema, or hop-count recognition — and every unsupported, seeded, prefiltered, policy-bearing, shortest-path, or cost-gated shape still declines to canonical execution with identical results. On a standard LDBC SNB SF1 interactive-short profile this removed the redundant two-hop typed-mask and 16-merge buckets and cut the profiled call ~11.9× (1263 ms → 106 ms), exact 19-row oracle preserved.
 - **Indexed seed lookup now covers a unique node id PLUS extra scalar constraints**: `{node_id: v, other: w}` previously fell back to a full node-table scan even though the unique node-id index can gather at most one row; it now gathers that one row and applies the remaining constraints to it. Missing ids and non-matching extra constraints return the same empty result as before, and duplicate-id graphs (which cannot build the index) are unaffected.

graphistry/compute/chain_lean_combine.py

@@ -91,7 +91,12 @@ def _lean_prefilter_right(left: DataFrameT, right: DataFrameT, key: str, engine:
     """Shrink ``right`` to the keys present in ``left`` before a ``how='left'``
     merge. A left merge discards unmatched ``right`` rows anyway, so this is
     byte-identical (row order = ``left`` order; matched right rows preserved,
-    including any fan-out). Only shrinks when ``left`` is materially smaller.
+    including any fan-out). Shrinks when ``left`` is EMPTY (nothing can match) or
+    materially smaller than ``right``; otherwise the membership pass costs more
+    than the join it would save.
+
+    pandas-only in practice: ``_lean_engine_ok`` is checked first and returns
+    ``right`` unchanged for cuDF / polars / dask, so nothing here is cross-engine.
     """
     if not _lean_combine_enabled() or not _lean_engine_ok(engine):
         return right
@@ -102,6 +107,20 @@ def _lean_prefilter_right(left: DataFrameT, right: DataFrameT, key: str, engine:
         right_len = len(right)
     except Exception:
         return right
-    if left_len == 0 or left_len * _LEAN_SHRINK_RATIO > right_len:
+    if left_len == 0:
+        # No left key can match anything, and a how='left' merge keeps only the right
+        # rows that DO match, so the result is empty whatever `right` holds. Hand back
+        # a zero-row slice -- same columns and dtypes, so the merge still produces the
+        # identical schema -- instead of letting the join materialize the whole frame.
+        # This is the case where shrinking is both maximally profitable and trivially
+        # correct, and it was the one case the gate below declined: an empty
+        # intermediate against a graph-sized side (measured: a 0-row left joined
+        # against 14M edges dominated a 112ms single-node query).
+        # `.iloc` and not `right[0:0]`: bare slicing is LABEL-based on a float index
+        # (pandas routes those through slice_indexer), so `right[0:0]` returns ONE row
+        # there and the optimization silently does nothing. `.iloc` is positional on
+        # every index type.
+        return right.iloc[0:0]
+    if left_len * _LEAN_SHRINK_RATIO > right_len:
         return right
     return right[right[key].isin(left[key])]

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

@@ -928,8 +928,17 @@ def _plain_edge(op):
     from graphistry.compute.util import generate_safe_column_name
     from graphistry.compute.gfql.lazy import collect_all
     NORD = generate_safe_column_name("__gfql_norder__", g._nodes, prefix="__gfql_", suffix="__")
-    EORD = generate_safe_column_name("__gfql_eorder__", g._edges, prefix="__gfql_", suffix="__")
-    g_lz = _LazyShim(g._nodes.with_row_index(NORD).lazy(), g._edges.with_row_index(EORD).lazy(),
+    if added_edge_index:
+        # EID was attached above as `with_row_index` over THIS frame in THIS order, so it
+        # already IS the stable edge order. Materializing a second identical 0..n-1 column
+        # over a graph-sized edge frame is pure duplication (measured: `with_row_index`
+        # cost ~34ms/query across the eager calls on a 14M-edge graph). Reuse EID.
+        EORD = EID
+        edges_lz = g._edges.lazy()
+    else:
+        EORD = generate_safe_column_name("__gfql_eorder__", g._edges, prefix="__gfql_", suffix="__")
+        edges_lz = g._edges.with_row_index(EORD).lazy()
+    g_lz = _LazyShim(g._nodes.with_row_index(NORD).lazy(), edges_lz,
                      node_col, src, dst, g._edge)
     steps_lz = [(op, _LazyShim.step(p)) for op, p in steps]
     edge_steps_lz = [(op, _LazyShim.step(p)) for op, p in edge_steps]
@@ -946,9 +955,10 @@ def _plain_edge(op):
     final_nodes = _apply_node_names(final_nodes, g_lz, steps_lz, auto_hop_col=auto_hop_col)
 
     final_nodes = final_nodes.sort(NORD).drop(NORD)
+    # EORD IS EID whenever we synthesized the id (that is the point of this change), so the
+    # single drop above removes it. There is no `added_edge_index and EID != EORD` case left
+    # to handle: the only branch that sets added_edge_index also sets EORD = EID.
     final_edges = final_edges.sort(EORD).drop(EORD)
-    if added_edge_index:
-        final_edges = final_edges.drop(EID)
     final_edges, final_nodes = collect_all([final_edges, final_nodes])
     final_edges = _restore_edge_dtypes(final_edges, src, dst, _endpoint_restore)
     return self.nodes(final_nodes, node_col).edges(final_edges, src, dst)

graphistry/tests/compute/test_chain_lean_combine.py

@@ -200,3 +200,114 @@ def _spy_prefilter(left, right, key, engine):
     # both lean paths must fire on a seeded chain (else parity tests are vacuous)
     assert hits["intersect"] >= 1
     assert hits["prefilter"] >= 1
+
+
+# --- empty-left shrink (#1783): both directions --------------------------------------
+# The shrink returns a ZERO-ROW slice of `right` when `left` is empty. That is only sound
+# because the sole call site merges how='left', which discards unmatched right rows anyway.
+# These pin both directions: the empty case must shrink AND stay schema-identical, and the
+# non-empty cases must be untouched by the new branch.
+
+def _wide_right(n_rows: int = 1000) -> pd.DataFrame:
+    rng = np.random.default_rng(5)
+    return pd.DataFrame({
+        "id": np.arange(n_rows, dtype=np.int64),
+        "f": rng.random(n_rows),
+        "s": np.array([f"v{i % 7}" for i in range(n_rows)]),
+        "b": rng.integers(0, 2, n_rows).astype(bool),
+    })
+
+
+def test_empty_left_shrinks_to_zero_rows_preserving_schema():
+    """The optimization itself: nothing survives a left merge from an empty left."""
+    right = _wide_right()
+    left = right.iloc[:0][["id"]]
+    out = _lean_prefilter_right(left, right, "id", Engine.PANDAS)
+
+    assert len(out) == 0, "empty left must not carry the full right frame into the merge"
+    # Schema-identical, or the downstream merge would produce different columns/dtypes.
+    assert list(out.columns) == list(right.columns)
+    assert list(out.dtypes) == list(right.dtypes)
+
+
+def test_empty_left_merge_result_matches_the_unshrunk_merge():
+    """End-to-end equivalence at the call site's actual merge type."""
+    right = _wide_right()
+    left = right.iloc[:0][["id"]]
+    shrunk = _lean_prefilter_right(left, right, "id", Engine.PANDAS)
+
+    got = left.merge(shrunk, on="id", how="left")
+    expected = left.merge(right, on="id", how="left")
+    pd.testing.assert_frame_equal(got, expected)
+
+
+@pytest.mark.parametrize("n_left", [1, 5, 999, 1000])
+def test_non_empty_left_is_unaffected_by_the_empty_branch(n_left):
+    """The other direction: adding the empty-left case must not perturb any non-empty
+    one — neither the shrink-eligible small lefts nor the ones the ratio gate declines."""
+    right = _wide_right()
+    left = right[["id"]].head(n_left)
+    out = _lean_prefilter_right(left, right, "id", Engine.PANDAS)
+
+    got = left.merge(out, on="id", how="left").sort_values("id").reset_index(drop=True)
+    expected = left.merge(right, on="id", how="left").sort_values("id").reset_index(drop=True)
+    pd.testing.assert_frame_equal(got, expected)
+    assert len(got) == n_left
+
+
+def test_empty_left_with_no_matching_keys_is_still_empty():
+    """A NON-empty left whose keys miss entirely must still yield null-filled rows, one
+    per left row — not a zero-row frame.
+
+    Note on what this does and does not pin: it constrains the RESULT, not the branch
+    taken. Shrinking on 'no key overlap' would also be sound under how='left' (both give
+    null-filled rows), so this test deliberately cannot distinguish those two
+    implementations — verified by mutation. What it does catch is the result-level error:
+    dropping the unmatched left rows entirely.
+    """
+    right = _wide_right()
+    left = pd.DataFrame({"id": np.array([10_000, 10_001], dtype=np.int64)})
+    out = _lean_prefilter_right(left, right, "id", Engine.PANDAS)
+
+    got = left.merge(out, on="id", how="left")
+    expected = left.merge(right, on="id", how="left")
+    pd.testing.assert_frame_equal(got, expected)
+    assert len(got) == 2 and got["f"].isna().all()
+
+
+def test_empty_left_shrink_holds_through_the_chain():
+    """The shape the fix exists for: a seed that matches nothing, so the intermediate is
+    empty and the combine would otherwise join it against the whole frame. Lean-on must
+    equal lean-off."""
+    g, _ = _seeded_graph()
+    q = [n({"id": -12345}), e_forward(), n()]
+
+    os.environ['GFQL_LEAN_COMBINE'] = '0'
+    off = g.chain(q, engine='pandas')
+    os.environ['GFQL_LEAN_COMBINE'] = '1'
+    on = g.chain(q, engine='pandas')
+
+    pd.testing.assert_frame_equal(
+        off._nodes.sort_values(list(off._nodes.columns)).reset_index(drop=True),
+        on._nodes.sort_values(list(on._nodes.columns)).reset_index(drop=True),
+    )
+    assert len(on._nodes) == 0
+
+
+def test_empty_left_shrink_works_on_a_float_index():
+    """`right[0:0]` is LABEL-based on a float index — pandas routes those through
+    slice_indexer — so the bare-slice spelling returns ONE row there and the shrink
+    silently does nothing. The result stays correct either way (a 0-row left still yields
+    a 0-row how='left' merge), which is exactly why this needs its own test: the bug is a
+    silent perf no-op, invisible to any assertion about the merge result."""
+    right = _wide_right()
+    right.index = np.arange(len(right), dtype=float)
+    left = right.iloc[:0][["id"]]
+
+    out = _lean_prefilter_right(left, right, "id", Engine.PANDAS)
+    assert len(out) == 0, (
+        f"shrink returned {len(out)} rows on a float index — bare [0:0] label-sliced "
+        "instead of position-sliced, so the optimization did not fire"
+    )
+    pd.testing.assert_frame_equal(
+        left.merge(out, on="id", how="left"), left.merge(right, on="id", how="left"))