perf(gfql/polars): stop deduplicating semi-join key sides

A semi-join emits a left row iff at least one matching right row exists, so
duplicate keys on the right cannot change the result -- and unlike an inner
join it cannot multiply rows either. Every `.unique()` applied to a frame that
is used *only* as a `how="semi"` key side is therefore a full hash pass bought
for no observable effect.

On an unfiltered hop the key side IS the node table, so this put O(N) work
inside a query whose answer is O(degree): the seeded single-hop plan built two
such frames per hop, and each cost ~53 ms at 3.18M nodes -- more than the rest
of the query put together.

Removed at the sites where the frame is provably a semi key side only:
  - chain.py `_semi` (the helper hardcodes how="semi")
  - chain.py alias hop-window gate, and the next-edge endpoint gate
  - chain.py endpoint gate in the two-hop fast path
  - pattern_apply.py start-nodes gate
  - hop_eager.py `_idframe_lf` -- all four consumers are semi key sides

Deliberately NOT removed: `named` in the alias pass keeps its `.unique()`
because it feeds a how="left" join, where duplicates WOULD multiply rows. The
eager multi-hop loop's `_idframe` is untouched -- its frames also flow into
concat/anti-join bookkeeping, which is a separate argument.

Measured, 3.18M nodes / 14M edges, seeded typed hop, engine=polars:
  127.44 -> 57.12 ms (2.23x), row counts identical.
At E=1.75M: 102.83 -> 31.35 ms (3.28x) -- the removed cost scales with N, not E.

Parity: 380 differential cases, 0 divergences, over duplicate node keys, null
ids, dangling edges, duplicate start_nodes, and 11 traversal shapes. gfql +
chain test suites: identical failure sets before and after.

Author: lmeyerov

CHANGELOG.md

@@ -12,6 +12,7 @@ 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
+- **GFQL polars chain stops deduplicating semi-join key sides**: the native polars executor applied `.unique()` to every frame it fed into a `how="semi"` join. A semi-join emits a left row iff at least one matching right row exists, so duplicate keys can neither change which rows come back nor multiply them the way an inner join would — the deduplication was a full hash pass over the key column bought for no observable effect. On an unfiltered hop the key side **is** the node table, so this put **O(N) work inside a query whose answer is O(degree)**: the seeded single-hop plan built two such key frames per hop, each costing ~53 ms at 3.18M nodes — more than the rest of the query combined. The `.unique()` is now dropped everywhere the frame is provably a semi key side only (the `_semi` helper, the alias hop-window and next-edge endpoint gates, the two-hop fast path's endpoint gate, the `start_nodes` gate, and the single-hop planner's id frames). It is deliberately **kept** on the alias frame that feeds a `how="left"` join, where duplicates genuinely would multiply rows, and the eager multi-hop loop is untouched (its frames also flow into concat/anti-join bookkeeping, a separate argument). Measured on a 3.18M-node / 14M-edge polars graph (LDBC SNB SF1-shaped), a seeded typed hop goes **127.4 → 57.1 ms (2.23×)** with identical row counts; at 1.75M edges **102.8 → 31.4 ms (3.28×)**, confirming the removed cost scales with node count, not edge count. Parity held across 380 differential comparisons covering duplicate node keys, null ids, dangling edges, duplicate `start_nodes`, and 11 traversal shapes; the gfql and chain suites show identical failure sets before and after. Pinned by tests that assert the boundary rather than the speed: duplicates reaching a semi key side must not change results or multiply rows, a dangling endpoint must still be excluded (the gate is load-bearing, not vacuous), and duplicate `start_nodes` must be inert.
 - **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.

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

@@ -28,8 +28,15 @@
 
 
 def _semi(df, ids_df, df_col, id_col):
-    """Rows of df whose df_col is present in ids_df[id_col] (vectorized semi-join)."""
-    return df.join(ids_df.select(id_col).unique(), left_on=df_col, right_on=id_col, how="semi")
+    """Rows of df whose df_col is present in ids_df[id_col] (vectorized semi-join).
+
+    The key frame is NOT deduplicated: a semi-join emits a left row iff at least one
+    matching right row exists, so duplicate keys cannot change which rows come back (and
+    a semi-join never multiplies rows the way an inner join would). Deduplicating first
+    is a whole extra hash pass over the key column for no observable effect — 60.7 -> 8.6 ms
+    on a 3.18M-key build side. See the module note on semi-join key frames.
+    """
+    return df.join(ids_df.select(id_col), left_on=df_col, right_on=id_col, how="semi")
 
 
 def _align_seed_dtype(seed, node_col, ref_nodes):
@@ -326,7 +333,7 @@ def _apply_node_names(out, g, steps, auto_hop_col: str = _AUTO_NODE_HOP):
                 if max_hop is not None:
                     in_window = in_window & (hop <= max_hop)
                 named = named.join(
-                    g_step._nodes.filter(in_window).select(pl.col(node_col)).unique(),
+                    g_step._nodes.filter(in_window).select(pl.col(node_col)),
                     on=node_col, how="semi")
         if idx + 1 < len(step_list):
             next_op, next_step = step_list[idx + 1]
@@ -338,7 +345,9 @@ def _apply_node_names(out, g, steps, auto_hop_col: str = _AUTO_NODE_HOP):
                     part = e.select(pl.col(dst).alias(node_col))
                 else:
                     part = endpoint_ids(e, src, dst, node_col)
-                named = named.join(part.unique(), on=node_col, how="semi")
+                # `part` is a semi-join key side (dupes cannot change it); `named` above keeps
+                # its .unique() because it feeds a how="left" join, where they WOULD multiply.
+                named = named.join(part, on=node_col, how="semi")
         flag = named.with_columns(pl.lit(True).alias(op._name))
         out = out.join(flag, on=node_col, how="left").with_columns(pl.col(op._name).fill_null(False))
     return out
@@ -809,7 +818,7 @@ def _plain_edge(op):
                     to_ids = filter_by_dict_polars(gf._nodes, n2.filter_dict).select(pl.col(ncol))
                     edges = edges.join(to_ids, left_on=to_col, right_on=ncol, how="semi")
             endpoints = endpoint_ids(edges, scol, dcol, ncol)
-            nodes = gf._nodes.join(endpoints.unique(), on=ncol, how="semi")
+            nodes = gf._nodes.join(endpoints, on=ncol, how="semi")
             return gf.nodes(nodes, ncol).edges(_restore_edge_dtypes(edges, scol, dcol, restore), scol, dcol)
 
     if start_nodes is not None:

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

@@ -223,7 +223,22 @@ def _idframe(df: "pl.DataFrame", col: str) -> "pl.DataFrame":
         pairs_lf = _build_hop_pairs(edges_lf, direction, src, dst, node_dtype, FROM, TO, EID)
 
         def _idframe_lf(lf: "pl.LazyFrame", col: str) -> "pl.LazyFrame":
-            return lf.select(pl.col(col).cast(node_dtype).alias(NID)).unique()
+            """Id frame for a SEMI-JOIN KEY side only — deliberately NOT deduplicated.
+
+            Every consumer below (`allowed_source_lf`, `allowed_dest_lf`, `target_final_lf`,
+            `frontier_lf`) is the key side of a `how="semi"` join, and a semi-join emits a
+            row iff >=1 match exists — duplicate keys cannot change the result, and unlike an
+            inner join it cannot multiply rows either. `frontier_lf` is also the LEFT side of
+            one semi-join, whose output then feeds another as the key side; duplicates stay
+            harmless the whole way. So `.unique()` here is a full hash pass over the node-id
+            column bought for nothing: on an unfiltered hop the key side IS the node table,
+            making it O(N) work inside a query whose answer is O(degree). Measured on a
+            3.18M-node table: 65.1 -> 9.0 ms for the cast+key frame, and this path builds two
+            of them per hop.
+
+            Anything that needs a *distinct* id set must apply its own `.unique()`.
+            """
+            return lf.select(pl.col(col).cast(node_dtype).alias(NID))
 
         allowed_source_lf = (
             _idframe_lf(filter_by_dict_polars(nodes if nodes is not None else all_nodes,

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

@@ -67,7 +67,7 @@ def rows_binding_ops_polars(g: Plottable, binding_ops: Sequence[Dict[str, JSONVa
         # decline the whole wavefront-constrained case (wave-2 W2-3).
         sn = start_nodes.collect() if isinstance(start_nodes, pl.LazyFrame) else start_nodes
         try:
-            nodes = nodes.join(sn.select(node_id).unique(), on=node_id, how="semi")  # type: ignore[operator]  # polars op on a DataFrameT-typed seed
+            nodes = nodes.join(sn.select(node_id), on=node_id, how="semi")  # type: ignore[operator]  # polars op on a DataFrameT-typed seed
         except Exception:
             return None
     try:

graphistry/tests/compute/gfql/test_engine_polars_semi_key_dedup.py

@@ -0,0 +1,136 @@
+"""Semi-join key sides carry duplicates on purpose — pin the invariant that makes that safe.
+
+The polars chain used to `.unique()` every frame it fed into a `how="semi"` join. That is a
+no-op for the RESULT (a semi-join emits a left row iff >=1 match exists; duplicates neither
+change which rows come back nor multiply them) but a full hash pass over the key column — and
+on an unfiltered hop the key side IS the node table, i.e. O(N) work inside an O(degree) query.
+
+These tests pin the boundary rather than the speed:
+  * duplicate keys reaching a SEMI key side must not change results or multiply rows, and
+  * the one frame that still needs `.unique()` — the alias frame feeding a how="left" join —
+    must keep it, because THERE duplicates do multiply.
+Pandas is the oracle throughout, so a divergence fails as a parity break, not a guess.
+"""
+import pandas as pd
+import pytest
+
+import graphistry
+from graphistry.compute.ast import n, e_forward, e_reverse, e_undirected
+
+pl = pytest.importorskip("polars")
+
+from graphistry.tests.compute.gfql.polars_test_utils import graph_sig  # noqa: E402
+
+
+def _pair(nodes_pd, edges_pd):
+    """Same graph as pandas frames and as polars frames."""
+    g_pd = graphistry.edges(edges_pd, "s", "d").nodes(nodes_pd, "key")
+    g_pl = graphistry.edges(pl.from_pandas(edges_pd), "s", "d").nodes(
+        pl.from_pandas(nodes_pd), "key")
+    return g_pd, g_pl
+
+
+def _clean_frames():
+    nodes = pd.DataFrame({"key": [0, 1, 2, 3, 4],
+                          "id": ["a", "b", "c", "d", "e"],
+                          "grp": [1, 2, 2, 1, 2]})
+    edges = pd.DataFrame({"s": [0, 0, 1, 2, 3],
+                          "d": [1, 2, 3, 3, 4],
+                          "type": ["K", "K", "L", "K", "K"]})
+    return nodes, edges
+
+
+def _dup_key_frames():
+    """A node table with the SAME key twice — the row that reaches a semi key side twice."""
+    nodes, edges = _clean_frames()
+    nodes = pd.concat([nodes, nodes[nodes["key"].isin([1, 3])]], ignore_index=True)
+    return nodes, edges
+
+
+def _dangling_frames():
+    """An edge whose destination is absent from the node table: the endpoint gate must still
+    exclude it, so this is the case where the semi-join is NOT vacuous."""
+    nodes, edges = _clean_frames()
+    edges = pd.concat([edges, pd.DataFrame({"s": [0], "d": [999], "type": ["K"]})],
+                      ignore_index=True)
+    return nodes, edges
+
+
+SHAPES = {
+    "fwd_typed": [n({"id": "a"}, name="m"), e_forward({"type": "K"}, name="r"), n(name="p")],
+    "rev_typed": [n({"id": "d"}, name="m"), e_reverse({"type": "K"}, name="r"), n(name="p")],
+    "undirected": [n({"id": "a"}, name="m"), e_undirected({"type": "K"}, name="r"), n(name="p")],
+    "two_hop": [n({"id": "a"}, name="m"), e_forward({"type": "K"}, hops=2, name="r"), n(name="p")],
+    "dest_filter": [n({"id": "a"}, name="m"), e_forward({"type": "K"}, name="r"),
+                    n({"grp": 1}, name="p")],
+    "src_and_dest_filter": [n({"grp": 1}, name="m"), e_forward({"type": "K"}, name="r"),
+                            n({"grp": 2}, name="p")],
+    "untyped": [n({"id": "a"}, name="m"), e_forward(name="r"), n(name="p")],
+    "fixed_point": [n({"id": "a"}, name="m"),
+                    e_forward({"type": "K"}, to_fixed_point=True, name="r"), n(name="p")],
+}
+
+BUILDERS = {"clean": _clean_frames, "dup_keys": _dup_key_frames, "dangling": _dangling_frames}
+
+
+@pytest.mark.parametrize("frames", sorted(BUILDERS))
+@pytest.mark.parametrize("shape", sorted(SHAPES))
+def test_polars_matches_pandas_when_semi_key_sides_carry_duplicates(frames, shape):
+    """Duplicate / dangling keys flow into the semi key sides; polars must still match pandas."""
+    g_pd, g_pl = _pair(*BUILDERS[frames]())
+    chain = list(SHAPES[shape])
+    assert graph_sig(g_pd.chain(chain, engine="pandas")) == \
+        graph_sig(g_pl.chain(chain, engine="polars")), \
+        f"polars diverged from the pandas oracle [{frames}/{shape}]"
+
+
+@pytest.mark.parametrize("shape", sorted(SHAPES))
+def test_duplicate_node_keys_do_not_multiply_result_rows(shape):
+    """The regression a missing `.unique()` on a how="left" key side would cause.
+
+    Duplicating node rows must not duplicate OUTPUT rows beyond what the duplicated input
+    itself accounts for: the alias-flag join is a left join, so an undeduplicated alias frame
+    would multiply every matching node. Bound the output by the input duplication instead of
+    hardcoding a count, so the test survives changes to the fixture.
+    """
+    chain = list(SHAPES[shape])
+    _, g_clean = _pair(*_clean_frames())
+    _, g_dup = _pair(*_dup_key_frames())
+    out_clean = g_clean.chain(chain, engine="polars")
+    out_dup = g_dup.chain(chain, engine="polars")
+
+    # each key appears at most twice in the duplicated node table
+    counts = out_dup._nodes["key"].value_counts()
+    worst = int(counts[counts.columns[-1]].max()) if counts.height else 0
+    assert worst <= 2, f"[{shape}] a node key came back {worst}x — alias join multiplied rows"
+    assert set(out_clean._nodes["key"].to_list()) == set(out_dup._nodes["key"].to_list()), \
+        f"[{shape}] duplicating node rows changed WHICH nodes matched"
+
+
+def test_dangling_endpoint_is_still_excluded():
+    """The semi-join is load-bearing, not vacuous: an endpoint missing from the node table
+    must not appear in the output. A 'the gate accepts everything, drop it' optimization
+    would break exactly here."""
+    g_pd, g_pl = _pair(*_dangling_frames())
+    chain = [n({"id": "a"}, name="m"), e_forward({"type": "K"}, name="r"), n(name="p")]
+    out = g_pl.chain(chain, engine="polars")
+    assert 999 not in out._nodes["key"].to_list(), "dangling endpoint leaked into the nodes"
+    assert graph_sig(g_pd.chain(chain, engine="pandas")) == graph_sig(out)
+
+
+@pytest.mark.parametrize("shape", ["fwd_typed", "undirected", "two_hop"])
+def test_duplicate_start_nodes_do_not_change_the_result(shape):
+    """`start_nodes` reaches a semi key side (pattern_apply); repeating rows in it must be
+    inert. This is the one key side a CALLER controls, so it can carry duplicates on any
+    schema, not just a malformed graph."""
+    _, g_pl = _pair(*_clean_frames())
+    chain = list(SHAPES[shape])
+    uniq = pl.DataFrame({"key": [0, 1, 2]})
+    dup = pl.DataFrame({"key": [0, 1, 1, 2, 2, 2]})
+    try:
+        a = g_pl.chain(chain, engine="polars", start_nodes=uniq)
+        b = g_pl.chain(chain, engine="polars", start_nodes=dup)
+    except TypeError:
+        pytest.skip("this chain surface does not accept start_nodes")
+    assert graph_sig(a) == graph_sig(b), \
+        f"[{shape}] duplicate start_nodes changed the result"